import yt_dlp
import os
import sys
import subprocess
def check_ffmpeg():
"""检查是否安装了ffmpeg"""
try:
# 尝试运行ffmpeg命令
with open(os.devnull, 'w') as devnull:
subprocess.check_call(['ffmpeg', '-h'], stdout=devnull, stderr=devnull)
return True
except (subprocess.SubprocessError, FileNotFoundError):
return False
def download_video(url, max_retries=3):
"""
下载 YouTube 视频,并转换为支持Twitter的MP4格式
参数:
url (str): 要下载的 YouTube 视频 URL
max_retries (int): 下载失败时的最大重试次数
"""
# 检查ffmpeg
if not check_ffmpeg():
print("警告: 未找到ffmpeg。无法转换为Twitter支持的格式。")
print("必须安装ffmpeg:
github.com/yt-dlp/yt-dlp#dep…")
print("程序将终止...")
return False
# 设置下载选项 - Twitter兼容格式
ydl_opts = {
# 保存在默认目录,使用视频标题作为文件名
'outtmpl': '%(title)s.%(ext)s',
# 显示进度条
'progress_hooks': [progress_hook],
# 选择Twitter兼容的格式
'format': 'bestvideo[ext=mp4][vcodec^=avc][height<=1080][filesize<300M] bestaudio[ext=m4a]/best[ext=mp4][vcodec^=avc][height<=1080][filesize<300M]',
# 如果没有完全匹配的格式,则转换为适合的格式
'postprocessors': [{
'key': 'FFmpegVideoConvertor',
'preferedformat': 'mp4',
}, {
# 确保使用H.264编码和AAC音频
'key': 'FFmpegVideoRemuxer',
'preferedformat': 'mp4',
}],
# 视频编码设置
'postprocessor_args': [
'-vcodec', 'libx264', '-acodec', 'aac',
'-pix_fmt', 'yuv420p', '-movflags', ' faststart',
# 确保视频比特率不超过5Mbps
'-b:v', '5M', '-maxrate', '5M', '-bufsize', '10M',
# 音频比特率
'-b:a', '128k'
],
# 启用HTTP源重试
'retries': 10,
# 启用片段重试
'fragment_retries': 10,
# 设置超时
'socket_timeout': 30,
# 在出错时继续
'ignoreerrors': False,
# 使用更多的格式提取器
'extractor_retries': 5,
# 使用外部下载器进行更稳定的下载
'external_downloader': 'aria2c' if is_aria2c_installed() else None,
# 设置缓冲区大小
'buffersize': 1024 * 1024, # 1MB
}
# 尝试下载,如果失败就重试
retries = 0
while retries <= max_retries:
try:
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(url, download=False)
print(f"\n准备下载: {info.get('title', '未知标题')}")
# 开始下载
ydl.download([url])
print("\n视频下载完成并保存为Twitter兼容的MP4格式!")
print("该视频应符合Twitter上传要求(H.264编码,AAC音频,MP4容器)")
return True
except Exception as e:
retries = 1
if retries <= max_retries:
print(f"\n下载出错: {e}")
print(f"正在尝试重新下载 (尝试 {retries}/{max_retries})...")
else:
print(f"\n下载失败: {e}")
print("已达到最大重试次数,下载失败。")
return False
def is_aria2c_installed():
"""检查是否安装了aria2c"""
try:
with open(os.devnull, 'w') as devnull:
subprocess.check_call(['aria2c', '--version'], stdout=devnull, stderr=devnull)
return True
except (subprocess.SubprocessError, FileNotFoundError):
return False
def progress_hook(d):
"""显示下载进度"""
if d['status'] == 'downloading':
percent = d.get('_percent_str', 'N/A')
speed = d.get('_speed_str', 'N/A')
eta = d.get('_eta_str', 'N/A')
downloaded = d.get('downloaded_bytes', 0)
total = d.get('total_bytes', 0) or d.get('total_bytes_estimate', 0)
if total > 0:
downloaded_mb = downloaded / (1024 * 1024)
total_mb = total / (1024 * 1024)
print(f'\r下载进度: {percent} ({downloaded_mb:.1f}MB/{total_mb:.1f}MB) 速度: {speed} 预计剩余时间: {eta}', end='')
else:
print(f'\r下载进度: {percent} 速度: {speed} 预计剩余时间: {eta}', end='')
elif d['status'] == 'finished':
print('\n下载完成,正在处理并转换为Twitter兼容格式...')
elif d['status'] == 'error':
print(f"\n下载出错: {d.get('error')}")
def main():
print("YouTube 视频下载器 - Twitter兼容版")
print("=" * 50)
print("此工具将下载视频并转换为Twitter支持的MP4格式")
print("要求:H.264编码,AAC音频,分辨率不超过1080p")
print("=" * 50)
# 检查ffmpeg
if not check_ffmpeg():
print("错误:必须安装ffmpeg才能使用此工具")
print("下载并安装ffmpeg:
github.com/yt-dlp/yt-dlp#dep…")
return
# 提示用户输入视频地址
url = input("请输入 YouTube 视频地址: ")
if not url:
print("错误:未输入视频地址!")
return
print(f"\n开始处理: {url}")
# 下载视频
download_video(url)
if __name__ == '__main__':
main()