下载模块是整个项目里第一个动手写的部分——YouTube、B 站、抖音、小红书、Twitter 的链接会一股脑丢进来,每个平台的接口、反爬策略、字幕格式都不一样。用户会把 YouTube、B 站、抖音、小红书、Twitter 的链接一股脑丢进来,每个平台的接口、反爬策略、字幕格式都不一样,一开始我甚至想过给每个平台写独立爬虫,后来想想维护成本直接劝退。最后定下的方案是:统一用 yt-dlp 做底层,外面包一层薄薄的适配层,把平台差异全部收敛到几个方法里。这套设计跑了半年,加新平台时基本不用动主流程。下面从 yt-dlp 是什么讲起,逐步拆到封装和踩坑。
一、yt-dlp 是什么
yt-dlp 是 youtube-dl 的活跃 fork,支持 1000+ 网站的视频/音频/字幕下载。这个数量级意味着绝大多数常见平台我都不用写任何解析逻辑,传个 URL 进去它自己会识别。支持列表大致覆盖:
- YouTube、B站、抖音、小红书、Twitter、TikTok、Instagram;
- Vimeo、Dailymotion、Twitch、Reddit;
- 各国家地区的视频网站。
相比 youtube-dl,选它有几个现实理由:
- 持续更新(适配平台反爬);
- 性能更好(多线程下载);
- 格式选择更灵活;
- 社区活跃。
youtube-dl 基本处于停止维护的状态,平台反爬一升级它就失灵;yt-dlp 的社区把更新节奏拉得很紧,出问题当天往往就有修好的版本。对我们这种拿它当基础设施的项目来说,”有人持续维护”本身就是选它的核心理由。
二、基础调用
先看最基础的用法。yt-dlp 的用法高度集中在 YoutubeDL 这个类上,配置、解析、下载都通过它完成:
import yt_dlp
ydl_opts = {
'format': 'bestvideo+bestaudio/best',
'outtmpl': '%(title)s.%(ext)s',
'noplaylist': True,
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(url, download=True)
extract_info 返回视频元信息(标题、时长、缩略图、可选格式等)。
我一开始没设 noplaylist,结果有次用户丢了个视频链接,它把整个播放列表都拉了下来,磁盘瞬间多出几十个文件。这两个选项是教训换来的。extract_info 的返回值包含标题、时长、缩略图、可用格式列表,元信息拿全了,前端才能先渲染卡片再下载。
三、平台封装设计
直接裸调 yt-dlp 的问题在于,每个平台的需求不一样:B 站要 Referer、抖音要移动端 UA、Twitter 可能要登录态。把这些差异散落在业务代码里会非常难维护,我选择封装一个 VideoDownloader 类,用 VideoInfo 数据结构统一对外输出:
# services/downloader.py
import yt_dlp
from typing import Optional
from dataclasses import dataclass
@dataclass
class VideoInfo:
url: str
platform: str
video_id: str
title: str
duration: int
thumbnail: str
uploader: str
formats: list # 可选格式
class VideoDownloader:
def __init__(self):
self._base_opts = {
'quiet': True,
'no_warnings': True,
'noplaylist': True,
}
async def get_info(self, url: str) -> VideoInfo:
"""异步获取视频元信息"""
loop = asyncio.get_event_loop()
info = await loop.run_in_executor(
None, self._extract_info, url
)
return self._parse_info(url, info)
def _extract_info(self, url: str) -> dict:
with yt_dlp.YoutubeDL(self._base_opts) as ydl:
return ydl.extract_info(url, download=False)
def _parse_info(self, url: str, info: dict) -> VideoInfo:
return VideoInfo(
url=url,
platform=self._detect_platform(url),
video_id=info.get('id', ''),
title=info.get('title', ''),
duration=info.get('duration', 0),
thumbnail=info.get('thumbnail', ''),
uploader=info.get('uploader', ''),
formats=info.get('formats', [])
)
def _detect_platform(self, url: str) -> str:
if 'youtube.com' in url or 'youtu.be' in url:
return 'youtube'
if 'bilibili.com' in url:
return 'bilibili'
if 'douyin.com' in url:
return 'douyin'
if 'xiaohongshu.com' in url:
return 'xiaohongshu'
if 'twitter.com' in url or 'x.com' in url:
return 'twitter'
if 'tiktok.com' in url:
return 'tiktok'
return 'unknown'
这段代码解决两件事:一是把 yt-dlp 的 dict 返回值转成结构化的 VideoInfo,上层代码再也不用面对一堆魔法键名;二是用 URL 关键词识别平台,后面各平台的专属参数就靠这个字段来选。整个下载器对外只暴露 get_info 和 download,业务层不关心底层是 yt-dlp 还是别的。
四、各平台适配
平台差异集中在三件事:格式选择、请求头、字幕语言。下面按平台拆开讲,参数看着多,其实每一行都能对上具体的坑。
1. YouTube
YouTube 的视频流和音频流是分开的,所以我限定 1080p 以内,同时把中英文字幕一起拉下来:
def _youtube_opts(self) -> dict:
return {
**self._base_opts,
'format': 'bestvideo[height<=1080]+bestaudio/best',
'writesubtitles': True,
'subtitleslangs': ['zh-Hans', 'en'],
'writethumbnail': True,
}
YouTube 字幕分”自动字幕”和”人工字幕”:
writesubtitles:只下人工字幕;writeautomaticsub:下自动字幕(机器生成);allsubtitles:所有语言。
我把字幕语言限制在简中和英文,理由很简单:AI 总结要用,中文优先,英文兜底,再多就是浪费存储。
2. B 站
B 站不加 Referer 直接请求会返回 403,这是容易踩的第一个坑:
def _bilibili_opts(self) -> dict:
return {
**self._base_opts,
'format': 'bestvideo+bestaudio/best',
'writesubtitles': True,
'subtitleslangs': ['zh-Hans'], # B 站字幕
'http_headers': {
'Referer': 'https://www.bilibili.com',
'User-Agent': 'Mozilla/5.0 ...'
}
}
B 站特色:
- 必须带 Referer 否则 403;
- 字幕分”官方字幕”和”AI 字幕”;
- 分 P 视频要选 P 数。
字幕这里还有个细节:B 站有些视频只有 AI 字幕,writesubtitles 只抓官方字幕,抓不到就得靠后面的 Whisper 兜底。
3. 抖音(无 Cookie)
抖音是国内反爬较严的平台之一,无 Cookie 下载要靠移动端 UA 伪装(详细方案我单独写了一篇):
def _douyin_opts(self) -> dict:
return {
**self._base_opts,
'format': 'best',
'http_headers': {
'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS ...) '
'AppleWebKit/605.1.15'
},
'extractor_args': {
'douyin': {
'app_name': 'douyin_web',
'aid': '6383'
}
}
}
抖音反爬严,详见下一节。
4. 小红书
小红书视频和图文笔记都能下,但图文笔记没有视频流,要转成视频得额外处理:
def _xiaohongshu_opts(self) -> dict:
return {
**self._base_opts,
'format': 'best',
'http_headers': {
'User-Agent': 'Mozilla/5.0 ...',
'Referer': 'https://www.xiaohongshu.com'
}
}
小红书视频和图文笔记都能下,但图文转视频需要额外处理。
5. Twitter/X
Twitter 的高清视频基本要求登录态,未登录只能拿到 syndication API 返回的低清版本:
def _twitter_opts(self) -> dict:
return {
**self._base_opts,
'format': 'best',
# 推特视频需要登录态才能下高分辨率
'extractor_args': {
'twitter': {
'api': 'syndication' # 用 syndication API 免登录
}
}
}
推特未登录只能下 syndication API 返回的版本(480p 左右),高清要登录。
这个限制我给产品侧也同步了:Twitter 默认给 480p,用户要高清就去传 Cookie。预期设对,运维才不会三天两头被投诉。
五、异步执行
yt-dlp 的 extract_info 是同步阻塞调用,直接丢进 FastAPI 事件循环会卡住所有请求。我把它丢到线程池里执行,用一个 ThreadPoolExecutor 隔离:
import asyncio
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=4)
async def get_info(self, url: str) -> VideoInfo:
loop = asyncio.get_event_loop()
# 同步逻辑放线程池
info = await loop.run_in_executor(executor, self._extract_sync, url)
return self._parse_info(url, info)
线程池大小我设了 4,够用又不会把网卡打满。如果图省事直接在 async 函数里同步调 extract_info,压测时其他请求会集体超时,这是必踩的坑。
六、下载进度回调
下载一个 1080p 视频要几十秒,前端必须能看到进度条,否则用户早跑了。yt-dlp 的 progress_hooks 提供了回调钩子:
progress_data = []
def progress_hook(d):
if d['status'] == 'downloading':
percent = d.get('_percent_str', '0%')
speed = d.get('_speed_str', '0B/s')
eta = d.get('_eta_str', '00:00')
# 推 SSE 给前端
sse_queue.put({
'event': 'progress',
'data': {'percent': percent, 'speed': speed, 'eta': eta}
})
elif d['status'] == 'finished':
sse_queue.put({'event': 'done', 'data': {'path': d['filename']}})
ydl_opts = {
'progress_hooks': [progress_hook],
# ...
}
回调里拿到百分比、速度、剩余时间三个值,塞进 SSE 队列推给前端。要注意的是回调在下载线程里执行,别在里边做重活,否则会拖慢下载本身。
七、错误处理
yt-dlp 的错误信息是长长一串字符串,直接抛给用户谁也看不懂。我把它翻译成几类明确的业务异常:
class DownloadError(Exception):
pass
class VideoUnavailable(DownloadError):
"""视频不可用(删除/私密)"""
class NetworkError(DownloadError):
"""网络错误"""
class UnsupportedPlatform(DownloadError):
"""不支持的平台"""
def _handle_error(self, e: yt_dlp.utils.DownloadError):
msg = str(e)
if 'Video unavailable' in msg:
raise VideoUnavailable(msg)
if 'Unable to extract' in msg:
raise UnsupportedPlatform(msg)
if 'HTTP Error 429' in msg:
raise NetworkError('Rate limited')
raise DownloadError(msg)
前端收到 VideoUnavailable 就提示”视频已删除或私密”,收到 NetworkError 就提示”稍后重试”,错误文案和用户行为一一对应,客服的压力小了很多。
八、Cookie 与登录态
有些平台(Twitter、B 站大会员)必须带登录态。yt-dlp 支持从文件读取 cookie:
# 从环境变量读 cookie
ydl_opts = {
'cookiefile': '/path/to/cookies.txt', # Netscape 格式
}
平台用 cookies.txt 文件管理登录态:
# 用浏览器扩展导出 cookies
# EditThisCookie / Get cookies.txt LOCALLY
敏感 cookie 加密存储。
这里有个格式坑:浏览器扩展导出时可能选 JSON 格式,yt-dlp 只认 Netscape 格式,格式错了静默失效,排查起来特别费劲。cookie 文件本身是敏感数据,我加密后存储,权限也收紧到只有下载进程能读。
九、字幕与元数据
字幕是 AI 总结的原料,这一节的配置直接决定总结质量:
ydl_opts = {
'writesubtitles': True,
'writeautomaticsub': True,
'subtitleslangs': ['zh-Hans', 'en', 'ja'],
'writethumbnail': True,
'writeinfojson': True, # 元数据 JSON
'postprocessors': [{
'key': 'FFmpegEmbedSubtitle', # 嵌入字幕到视频
}]
}
我把简中、英文、日文三种字幕都开了,覆盖大多数内容。元数据 JSON 也一并落盘,后面做搜索和推荐时能直接用。如果视频没有字幕文件,就走 Whisper 转录,这条兜底链路在第十一节会看到。
十、性能优化
下载慢和下载失败是两回事,性能优化我做了三处:
1. 多线程下载
ydl_opts = {
'concurrent_fragment_downloads': 4, # 4 线程
}
2. 格式选择
# 选 1080p 而非 4K(带宽优先)
'format': 'bestvideo[height<=1080]+bestaudio/best'
3. 缓存元信息
# 同一视频 1 小时内不重复解析
import functools
import time
@functools.lru_cache(maxsize=1000)
def _get_info_cached(url: str, _t: int = None):
if _t is None or time.time() - _t < 3600:
return self._extract_info(url)
这三处里,缓存元信息的收益直接:同一视频被重复解析是高频场景,lru_cache 配时间戳,1 小时内的重复请求直接命中,接口延迟从秒级降到毫秒级。格式上我坚持 1080p 上限,不是 4K 下不了,而是带宽和存储成本撑不住所有用户都下 4K。优化顺序上我建议先做缓存再做并发,前者零风险收益大。
十一、与 AI 总结的衔接
下载完不是终点,字幕还要喂给 AI。完整链路这样串:
async def download_and_summarize(self, url: str, user: User):
# 1) 下载视频
info = await self.get_info(url)
video_path = await self.download(url)
# 2) 提取字幕(优先)
subtitle = await self.extract_subtitle(url, info)
# 3) 无字幕则用 Whisper 转录
if not subtitle:
subtitle = await self.whisper_transcribe(video_path)
# 4) AI 总结
summary = await ai_service.summarize(subtitle)
mindmap = await ai_service.mindmap(summary)
return {
'info': info,
'subtitle': subtitle,
'summary': summary,
'mindmap': mindmap
}
这里最重要的决策是”字幕优先、Whisper 兜底”。有字幕的视频走字幕,成本低、速度快、准确率高;没字幕的才上 Whisper,因为转录耗时是字幕的几十倍。顺序反了,服务器资源和用户等待时间都会翻倍。
十二、踩过的坑
下载器是踩坑重灾区,我把排查顺序和教训都记了下来,新接手的人可以先按这个清单查:
- 先看 yt-dlp 版本是不是太旧,平台改接口后老版本必挂;
- 再看请求头(Referer、UA)够不够,403 十有八九是这里;
- 然后看格式选择有没有 fallback,
bestvideo+bestaudio不存在时直接崩; - 最后看是不是并发太高触发了平台限流。
- yt-dlp 频繁更新:平台改了反爬就要更新 yt-dlp。每月拉新版。
- 大视频下载超时:4K 视频 5GB,10Mbps 网络下 1 小时。设超时 + 断点续传。
- 格式选择冲突:
bestvideo+bestaudio不存在时(如只有 DASH 单一文件)会失败。加 fallbackbest。 - Subtitle 不存在:很多视频没字幕,必须有 Whisper 兜底。
- Cookie 过期:登录态失效要通知用户重新登录。
- 版权限制:YouTube 音乐 MV 受 DRM 保护,下不了。要捕获异常提示用户。
- 并发过多:同时下 10 个视频触发平台限流。用 Semaphore 限并发。
第一条最隐蔽:平台偷偷改接口后 yt-dlp 的解析逻辑就过时了,报错信息跟用户行为完全对不上,我排查了一整天才发现是版本问题。从那以后我把 yt-dlp 升级做成每月例行任务,并在 requirements.txt 里锁住版本下限。
十三、与商业 SaaS 差异
用户常问”为什么不自用个商业下载工具”。我做了张对比表,把两种路子的账算清楚:
| 维度 | yt-dlp 自建 | 商业下载工具 |
|---|---|---|
| 价格 | 免费 | 订阅 |
| 多平台 | 1000+ | 几个 |
| 反爬更新 | 自己跟 | 平台维护 |
| 部署灵活 | 自部署 | 第三方依赖 |
| 维护成本 | 高 | 低 |
结论一句话:自建买的是平台覆盖和可定制性,代价是反爬和版本更新都得自己盯。对我们这种要把下载接进自己业务流的产品,自建是唯一能跑通的路。
常见问题(FAQ)
Q1:能下付费视频吗?
不能。yt-dlp 不破解付费墙。
Q2:能下 YouTube 4K 吗?
能,但带宽要求高。
Q3:为什么抖音有时下不了?
抖音反爬严,需要 Cookie 或 IP 轮换。