脚本如下
#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ YouTube 下载工具 (基于 yt-dlp) 支持：单个视频、播放列表、仅音频、画质选择、进度显示、错误重试等 """ import yt_dlp import os import sys import argparse import subprocess import re from pathlib import Path # 控制台颜色（可选，用于美化输出） class Colors: GREEN = '\033[92m' YELLOW = '\033[93m' RED = '\033[91m' BLUE = '\033[94m' RESET = '\033[0m' def print_info(msg): print(f"{Colors.BLUE}[INFO]{Colors.RESET} {msg}") def print_success(msg): print(f"{Colors.GREEN}[SUCCESS]{Colors.RESET} {msg}") def print_error(msg): print(f"{Colors.RED}[ERROR]{Colors.RESET} {msg}") def print_warning(msg): print(f"{Colors.YELLOW}[WARNING]{Colors.RESET} {msg}") def progress_hook(d): """下载进度回调函数""" if d['status'] == 'downloading': # 获取下载进度 if d.get('total_bytes'): total = d['total_bytes'] downloaded = d.get('downloaded_bytes', 0) percent = (downloaded / total) * 100 speed = d.get('speed', 0) if speed: speed_mb = speed / 1024 / 1024 print(f"\r 下载进度: {percent:.1f}% | 速度: {speed_mb:.2f} MB/s", end='') else: print(f"\r 下载进度: {percent:.1f}%", end='') elif d.get('total_bytes_estimate'): total = d['total_bytes_estimate'] downloaded = d.get('downloaded_bytes', 0) percent = (downloaded / total) * 100 print(f"\r 下载进度: {percent:.1f}% (估算)", end='') elif d['status'] == 'finished': print("\n ✅ 下载完成，正在进行后续处理（合并/转换）...") elif d['status'] == 'error': print_error("下载过程中出现错误") def merge_video_audio_with_ffmpeg(video_file, audio_file, output_file): """ 使用 FFmpeg 合并视频和音频文件 参数: video_file: 视频文件路径 audio_file: 音频文件路径 output_file: 输出文件路径 返回: bool: 合并成功返回 True，否则返回 False """ try: print_info(f"开始合并视频和音频...") print_info(f" 视频: {os.path.basename(video_file)}") print_info(f" 音频: {os.path.basename(audio_file)}") # FFmpeg 合并命令 cmd = [ 'ffmpeg', '-i', video_file, '-i', audio_file, '-c:v', 'copy', # 视频流直接复制（不重新编码） '-c:a', 'aac', # 音频编码为 AAC '-map', '0:v:0', # 使用第一个输入文件的第一个视频流 '-map', '1:a:0', # 使用第二个输入文件的第一个音频流 '-shortest', # 以较短的流为准 '-y', # 覆盖输出文件（如果存在） output_file ] # 执行命令 result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode == 0: print_success(f"合并成功！输出文件: {output_file}") # 可选：删除原始的分开文件 try: os.remove(video_file) os.remove(audio_file) print_info("已删除原始的分开文件") except: pass return True else: print_error(f"合并失败: {result.stderr}") return False except Exception as e: print_error(f"合并过程中出现错误: {str(e)}") return False def find_video_audio_files(directory, video_pattern=None, audio_pattern=None): """ 在目录中查找视频和音频文件 参数: directory: 目录路径 video_pattern: 视频文件匹配模式（正则表达式） audio_pattern: 音频文件匹配模式（正则表达式） 返回: tuple: (video_file, audio_file) 或 (None, None) """ files = os.listdir(directory) # 如果没有指定模式，尝试常见的格式 if video_pattern is None: # 查找 .mp4 视频文件（通常包含 f数字 格式） video_files = [f for f in files if '.mp4' in f and not '.f251' in f and not 'audio' in f.lower()] # 优先选择包含 f399 或 f400 等高质量的文件 video_files.sort(reverse=True) video_file = video_files[0] if video_files else None else: video_matches = [f for f in files if re.search(video_pattern, f)] video_file = video_matches[0] if video_matches else None if audio_pattern is None: # 查找 .webm 或 .m4a 音频文件 audio_files = [f for f in files if ('.webm' in f or '.m4a' in f) and ('f251' in f or 'audio' in f.lower())] audio_file = audio_files[0] if audio_files else None else: audio_matches = [f for f in files if re.search(audio_pattern, f)] audio_file = audio_matches[0] if audio_matches else None return (os.path.join(directory, video_file) if video_file else None, os.path.join(directory, audio_file) if audio_file else None) def download_media(url, output_dir="Downloads", format_spec="best", audio_only=False, audio_format="mp3", audio_quality="192", playlist=False, embed_subs=False, subs_lang="en", max_height=None, proxy=None, retries=10, merge_manually=False): """ 通用下载函数 参数: url: YouTube 视频或播放列表 URL output_dir: 输出目录 format_spec: 画质/格式规格 (例如: 'best', 'bestvideo+bestaudio', 'worst') audio_only: 是否仅下载音频 audio_format: 音频格式 (mp3, m4a, opus, etc.) audio_quality: 音频比特率 (如 '128', '192', '320') playlist: 是否强制按播放列表下载 (即使URL不是播放列表形式) embed_subs: 是否嵌入字幕 subs_lang: 字幕语言代码 (如 'en', 'zh-Hans', 'zh-Hant') max_height: 限制视频最大高度 (如 1080, 720) proxy: 代理地址 (如 'http://127.0.0.1:10809') retries: 重试次数 merge_manually: 是否手动合并（下载分开的视频和音频后手动合并） """ # 确保输出目录存在 Path(output_dir).mkdir(parents=True, exist_ok=True) # 基础配置 ydl_opts = { 'outtmpl': os.path.join(output_dir, '%(title)s.%(ext)s'), 'ignoreerrors': True, # 忽略单个视频的错误，继续下载列表中的其他视频 'nooverwrites': True, # 跳过已下载的文件 'retries': retries, # 全局重试次数 'fragment_retries': retries, # 分片重试次数 'socket_timeout': 30, # 网络超时 'progress_hooks': [progress_hook], # 进度回调 } # 代理配置 if proxy: ydl_opts['proxy'] = proxy # 播放列表处理 if playlist: ydl_opts['noplaylist'] = False # 下载整个播放列表 else: ydl_opts['noplaylist'] = True # 只下载单个视频 # 字幕配置 if embed_subs: ydl_opts['writesubtitles'] = True ydl_opts['writeautomaticsub'] = True # 也下载自动生成的字幕 ydl_opts['subtitleslangs'] = [subs_lang] ydl_opts['embedsubs'] = True ydl_opts['embedmetadata'] = True # 嵌入元数据（标题、封面等） # 画质限制 if max_height: # 例如：'bestvideo[height<=1080]+bestaudio/best[height<=1080]' format_spec = f'bestvideo[height<={max_height}]+bestaudio/best[height<={max_height}]' # 音频/视频模式 if audio_only: print_info("音频模式已启用，将下载最佳音质并转换为 {} 格式 ({}kbps)".format(audio_format, audio_quality)) ydl_opts['format'] = 'bestaudio/best' ydl_opts['postprocessors'] = [{ 'key': 'FFmpegExtractAudio', 'preferredcodec': audio_format, 'preferredquality': audio_quality, }] # 对于音频，修改输出模板为 .audio 前缀（可选） ydl_opts['outtmpl'] = os.path.join(output_dir, '%(title)s.%(ext)s') else: if merge_manually: # 手动合并模式：分别下载视频和音频，不自动合并 print_info("手动合并模式：将分别下载视频和音频流") ydl_opts['format'] = 'bestvideo+bestaudio' # 不设置 merge_output_format 和 postprocessors else: print_info(f"视频模式，画质规格: {format_spec}") ydl_opts['format'] = format_spec ydl_opts['merge_output_format'] = 'mp4' # 最终合并为 MP4 # 确保视频格式为 MP4（后处理） ydl_opts['postprocessors'] = [{ 'key': 'FFmpegVideoConvertor', 'preferedformat': 'mp4', }] # 开始下载 try: with yt_dlp.YoutubeDL(ydl_opts) as ydl: print_info(f"正在解析 URL: {url}") # 先获取视频信息（可选，用于显示标题） info = ydl.extract_info(url, download=False) video_title = None if 'entries' in info: # 播放列表 count = len(info['entries']) if info['entries'] else 0 print_info(f"检测到播放列表: {info.get('title', 'Unknown')} (共 {count} 个视频)") if count == 0: print_warning("播放列表为空，退出。") return # 如果是播放列表，获取第一个视频的标题作为参考 if info['entries'] and info['entries'][0]: video_title = info['entries'][0].get('title', 'Unknown') else: # 单个视频 video_title = info.get('title', 'Unknown') print_info(f"视频标题: {video_title}") # 实际下载 ydl.download([url]) print_success("下载完成！") # 如果是手动合并模式，尝试合并分开的视频和音频 if merge_manually and not audio_only: print_info("尝试查找并合并分开的视频和音频文件...") # 等待文件写入完成 import time time.sleep(2) # 查找视频和音频文件 video_file, audio_file = find_video_audio_files(output_dir) if video_file and audio_file: # 生成输出文件名 if video_title: # 清理标题中的非法字符 safe_title = re.sub(r'[\\/*?:"<>|]', "", video_title) output_file = os.path.join(output_dir, f"{safe_title}_merged.mp4") else: output_file = os.path.join(output_dir, "merged_output.mp4") # 执行合并 merge_video_audio_with_ffmpeg(video_file, audio_file, output_file) else: print_warning("未找到需要合并的视频和音频文件") print_info(f"请手动合并: {output_dir} 目录下的视频和音频文件") print_success("所有任务执行完毕！") except Exception as e: print_error(f"下载失败: {str(e)}") sys.exit(1) def main(): parser = argparse.ArgumentParser( description="YouTube 下载工具 - 支持视频/音频、播放列表、画质选择", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例: %(prog)s "https://youtu.be/xxxxx" # 下载单个视频（最佳画质） %(prog)s "https://youtu.be/xxxxx" -a # 仅下载音频为 MP3 %(prog)s "播放列表URL" -p # 下载整个播放列表 %(prog)s "视频URL" -q 720 # 限制最大高度720p %(prog)s "视频URL" -o "./my_videos" -f "best[height<=480]" # 自定义输出和格式 %(prog)s "视频URL" --proxy "http://127.0.0.1:10809" # 使用代理 %(prog)s "视频URL" --merge-manual # 手动合并模式（下载分开的视频和音频） """ ) parser.add_argument("url", help="YouTube 视频或播放列表的 URL") parser.add_argument("-o", "--output", default="Downloads", help="输出目录 (默认: Downloads)") parser.add_argument("-f", "--format", default="bestvideo+bestaudio/best", help="yt-dlp 格式规格 (默认: bestvideo+bestaudio/best)") parser.add_argument("-a", "--audio", action="store_true", help="仅下载音频模式") parser.add_argument("--audio-format", default="mp3", choices=["mp3", "m4a", "opus", "aac", "flac", "wav"], help="音频格式 (默认: mp3)") parser.add_argument("--audio-quality", default="192", help="音频比特率 kbps (默认: 192)") parser.add_argument("-p", "--playlist", action="store_true", help="强制下载整个播放列表 (即使 URL 是单个视频)") parser.add_argument("-s", "--subs", action="store_true", help="嵌入字幕 (英文字幕)") parser.add_argument("--subs-lang", default="en", help="字幕语言代码 (默认: en，中文可用 zh-Hans 或 zh-Hant)") parser.add_argument("-q", "--max-height", type=int, metavar="HEIGHT", help="限制视频最大高度 (如 1080, 720, 480)") parser.add_argument("--proxy", help="HTTP/HTTPS 代理，如 http://127.0.0.1:10809") parser.add_argument("--retries", type=int, default=10, help="下载重试次数 (默认: 10)") parser.add_argument("--merge-manual", action="store_true", help="手动合并模式：下载分开的视频和音频文件，然后使用 FFmpeg 合并") args = parser.parse_args() # 如果只下载音频，自动将 playlist 设为 False 更合理？但是用户仍可手动加 -p 来下载播放列表的音频 if args.audio and args.playlist: print_info("下载播放列表的音频版本...") download_media( url=args.url, output_dir=args.output, format_spec=args.format, audio_only=args.audio, audio_format=args.audio_format, audio_quality=args.audio_quality, playlist=args.playlist, embed_subs=args.subs, subs_lang=args.subs_lang, max_height=args.max_height, proxy=args.proxy, retries=args.retries, merge_manually=args.merge_manual, ) if __name__ == "__main__": # 检查是否安装了 yt-dlp try: import yt_dlp except ImportError: print_error("未安装 yt-dlp，请先运行: pip install yt-dlp") sys.exit(1) # 简单提醒 FFmpeg（不强制，但如果没有某些功能会受限） print_info("确保已安装 FFmpeg (视频合并/音频转换需要) 可从 https://ffmpeg.org/ 下载并添加到 PATH") main()
