Python异步爬虫与yt-dlp实战：构建B站视频批量下载工具

1. 项目缘起与核心需求解析

最近在几个主流视频平台上闲逛，发现一个挺有意思的现象：一类被称为“宅舞”的短视频内容，更新频率高得惊人，而且热度持续不减。这些视频通常制作精良，舞者表现力强，背景音乐也多是时下流行的ACG或流行曲目，观赏性确实不错。作为一个对数据收集和自动化有点兴趣的从业者，我第一反应不是单纯地欣赏，而是琢磨：这么多高质量、持续更新的内容，如果手动一个个去下载保存，效率太低，也容易遗漏。能不能写个工具，把这些视频批量、自动地“搬”到本地硬盘里，建一个自己的专属资源库，想什么时候看就什么时候看，还能避免某天视频突然下架的风险？

这个想法听起来简单，但真要实现一个稳定、高效、且“讲武德”的批量下载方案，里面涉及的技术点和需要注意的边界问题还真不少。它绝不仅仅是找到一个下载链接然后开下那么简单。我们需要考虑如何在海量的视频列表中精准定位到目标内容（比如特定舞者、特定歌曲、特定时间段），如何应对平台反爬机制保护账号安全，如何设计合理的下载队列和错误重试机制确保成功率，以及最重要的——如何在满足个人学习与研究需求的同时，严格遵守平台规则和版权法律，避免对服务器造成不必要的压力。今天，我就把自己搭建这样一个自动化工具的思路、踩过的坑以及最终成型的方案，详细地分享出来。如果你也对Python网络爬虫、异步编程以及多媒体处理感兴趣，或者单纯想拥有一个自己的离线视频库，那这篇内容应该能给你提供一条清晰的路径。

2. 技术方案选型与整体架构设计

面对“批量下载热点视频”这个需求，我们首先要摒弃“找到一个万能下载器”的幻想。不同的平台（如B站、抖音、油管等）其视频流传输机制、页面结构、反爬策略截然不同，一套代码通吃所有平台是不现实的。因此，我们的方案必须是平台针对性的。这里我以某个国内主流视频平台（我们称为平台A）的“宅舞”分区为例进行拆解，其技术思路可以迁移到其他平台。

整个系统的核心目标可以分解为：发现目标 -> 解析信息 -> 获取链接 -> 下载媒体 -> 组织管理。基于这个流程，我设计了以下技术栈和架构：

2.1 核心工具链选型

• 编程语言：Python 3.8+。这是毋庸置疑的选择。其丰富的生态库（Requests, BeautifulSoup, Scrapy, Selenium, aiohttp等）为网络爬虫提供了全方位支持，异步编程库（asyncio, aiofiles）能极大提升IO密集型任务（如下载）的效率。
• 请求与解析库：
  • httpx/aiohttp： 用于发起异步HTTP请求，比传统的requests库在并发场景下性能高出一个数量级，是批量下载的利器。
  • BeautifulSoup4/lxml： 用于解析HTML页面，提取视频标题、UP主信息、AV/BV号、封面图链接等元数据。
  • Selenium/Playwright：备选方案。当目标数据是通过JavaScript动态渲染，直接HTTP请求获取的HTML中不包含时，就需要动用这类浏览器自动化工具。它们能模拟真实用户操作，获取渲染后的完整页面内容，但代价是资源消耗大、速度慢。原则是：能不用就不用，优先尝试分析XHR/Fetch请求。
• 下载与流处理库：
  • youtube-dl/yt-dlp：社区神器。这是一个命令行程序，但其Python接口非常强大。它内部集成了对上千个网站的视频流解析逻辑，能自动识别最佳画质、处理加密流（如m3u8）。我们的核心下载功能将围绕它或它的增强版yt-dlp来构建。
  • FFmpeg： 多媒体处理领域的“瑞士军刀”。当视频和音频流是分离的（常见于高清视频），我们需要用yt-dlp获取流地址后，调用FFmpeg进行合并。它通常作为系统级依赖存在。
• 并发与任务调度：
  • asyncio： Python的原生异步IO框架。我们将利用它来协程化下载任务，实现数百个视频的同时下载，而不会开数百个线程导致系统崩溃。
  • aiofiles： 提供异步文件操作，避免在写入视频文件时阻塞整个事件循环。
• 数据存储与去重：
  • SQLite： 轻量级数据库。用于存储已成功下载的视频元数据（ID、标题、作者、发布时间、本地路径等），实现增量爬取，避免重复下载。
  • JSON/CSV： 用于简单配置和日志记录。

2.2 为什么选择 yt-dlp 而非直接抓取流链接？

这是一个关键决策。新手可能会尝试直接从网页源码或网络请求中寻找.mp4或.m3u8链接。但这存在几个大问题：

1. 复杂度高： 现代视频平台普遍使用动态、有时效性的签名来保护流媒体链接，直接解析难度极大，且规则经常变动。
2. 画质选择麻烦： 你需要自己解析出不同清晰度（1080P, 720P, 480P）对应的链接。
3. 音视频分离： 高清视频常将音视频流分开，你需要分别找到并下载，再用FFmpeg合并。yt-dlp完美解决了这些问题。它维护了一个庞大的“提取器”库，专门针对各个网站破解其视频信息获取逻辑。我们只需要告诉它视频的网页URL，它就能返回所有可用格式的流信息，并自动调用FFmpeg处理合并。这让我们能专注于“找视频列表”和“管理下载任务”，而不是“破解视频流”。

2.3 系统架构流程图（文字描述）

整个系统的工作流如下：

1. 种子输入： 提供一个或多个初始入口，如某个“宅舞”合集的链接、某个UP主的主页、或一个搜索关键词的结果页。
2. 列表爬取模块： 解析入口页面，通过翻页或滚动加载，获取所有目标视频的独立页面URL列表。这里需要处理分页逻辑。
3. 元数据提取模块： 对每个视频页面URL，提取核心元数据（标题、作者、发布时间、描述、标签等）。这些数据将存入SQLite数据库，并用于后续的文件命名和分类。
4. 下载任务生成模块： 检查数据库，过滤掉已成功下载的视频。为未下载的视频创建下载任务，包含其URL和元数据。
5. 异步下载引擎： 核心模块。使用asyncio创建多个并发任务，每个任务调用yt-dlp下载一个视频。需要集成进度回调、错误重试、速率限制等功能。
6. 本地存储与组织： 下载完成后，根据元数据（如作者、年份月份）自动组织文件夹结构，并将视频文件移动到对应位置，同时在数据库中更新状态。
7. 日志与监控： 整个过程中，所有操作（成功、失败、跳过）都需要记录到日志文件中，方便排查问题。

注意：法律与道德边界在开始之前，必须明确：此方案仅适用于个人学习、研究、欣赏，且不得用于任何商业用途。批量下载会对服务器造成压力，务必在代码中设置合理的请求间隔（如time.sleep(random.uniform(1, 3))），避免高频请求。绝对不要尝试绕过付费墙或下载明确声明禁止下载的版权内容。尊重创作者和平台规则是长久之计。

3. 核心模块拆解与实操要点

3.1 视频列表的高效发现与爬取

“宅舞”视频的发现渠道多种多样。我们的爬虫需要能适配不同的入口。

3.1.1 入口策略

1. UP主主页： 最直接的来源。构造URL模式如https://space.bilibili.com/{mid}/video。爬取时需要模拟滚动加载（分析XHR请求）或解析分页。
2. 频道或合集： 平台提供的官方或用户创建的合集页面，内容更垂直。
3. 搜索接口： 通过平台搜索API（通常需要分析其请求参数），用“宅舞”、“编舞”等关键词，结合时间、播放量排序来获取列表。这种方式获取的内容最广，但也最杂，需要后续过滤。
4. 排行榜单： 每日/每周热门榜，是发现“热点”视频的好地方。

实操要点：请求头与Cookie

import httpx headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Referer': 'https://www.bilibili.com/', # 关键，很多API会校验Referer 'Accept': 'application/json, text/plain, */*', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', } # 如果需要登录后才能访问的内容（如某些收藏夹），则需要加入Cookie。 # 获取Cookie：浏览器登录后，F12打开开发者工具，在Network标签下找一个请求，复制其Request Headers中的`Cookie`字段。 # 注意：Cookie是敏感信息，切勿泄露或上传到公开仓库。 cookies = {'SESSDATA': '你的SESSDATA值'} async with httpx.AsyncClient(headers=headers, cookies=cookies, timeout=30.0) as client: response = await client.get(url)

警告：使用Cookie意味着你的账号在进行操作。务必确保你的爬虫行为温和（低频率、有间隔），否则可能导致账号被临时封禁或限制功能。对于公开内容，尽量不使用Cookie。

3.1.2 解析列表与翻页

以B站UP主视频列表为例，其数据通常通过一个JSON API返回。我们需要分析网络请求。

1. 打开UP主视频页，F12进入“网络(Network)”选项卡，筛选XHR/Fetch请求。
2. 滚动页面，观察新出现的请求，找到一个返回视频列表数据的请求（响应体是JSON格式）。
3. 分析这个请求的URL、Query Parameters和Headers。
4. 在代码中模拟这个请求，解析返回的JSON数据，提取每个视频的bvid（或aid）和标题。

翻页逻辑通常包含在请求参数中，如pn（page number）和ps（page size）。通过循环递增pn即可获取所有页面，直到返回的数据列表为空。

3.2 元数据提取与数据库设计

获取到视频ID（如B站的BV号）列表后，我们需要获取每个视频的详细信息。

3.2.1 调用官方API最稳定可靠的方式是使用平台提供的公开API。例如，B站有https://api.bilibili.com/x/web-interface/view?bvid={bvid}。调用这个API可以获取到视频的完整信息，包括标题、作者、发布时间、描述、分区、标签、视频长度、各种清晰度的流信息等。

3.2.2 数据库表设计我们需要一个简单的数据库来记录状态，防止重复下载和便于管理。

-- 使用SQLite CREATE TABLE IF NOT EXISTS videos ( id INTEGER PRIMARY KEY AUTOINCREMENT, platform TEXT NOT NULL, -- 如 'bilibili' video_id TEXT NOT NULL UNIQUE, -- 如 'BV1xx411c7mh' title TEXT, author TEXT, publish_time INTEGER, -- 使用时间戳存储 description TEXT, tags TEXT, -- 可用逗号分隔存储 local_path TEXT, -- 本地存储路径 status TEXT DEFAULT 'pending', -- pending, downloading, completed, failed created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP );

在爬取到列表后，先将视频ID和基础信息插入数据库，状态设为pending。下载器只处理状态为pending的记录。

3.3 异步下载引擎的构建（核心中的核心）

这是整个项目技术含量最高的部分。我们将结合yt-dlp和asyncio来构建一个高性能下载器。

3.3.1 封装 yt-dlpyt-dlp本身是同步的。为了在异步环境中使用它，我们需要将其放入线程池中运行，避免阻塞事件循环。

import asyncio import yt_dlp from concurrent.futures import ThreadPoolExecutor class AsyncDownloader: def __init__(self, max_workers=5): self.executor = ThreadPoolExecutor(max_workers=max_workers) self.loop = asyncio.get_event_loop() async def download_video(self, url, output_template): """异步下载单个视频""" ydl_opts = { 'outtmpl': output_template, # 如 'downloads/%(uploader)s/%(title)s.%(ext)s' 'format': 'bestvideo[height<=1080]+bestaudio/best[height<=1080]', # 选择1080P及以下的最佳组合 'merge_output_format': 'mp4', 'quiet': False, 'no_warnings': False, 'progress_hooks': [self._progress_hook], # 进度回调 'socket_timeout': 30, 'retries': 3, 'fragment_retries': 3, } # 将同步的ydl.download()放到线程池中执行 try: with yt_dlp.YoutubeDL(ydl_opts) as ydl: # ydl.download([url]) 是阻塞的 await self.loop.run_in_executor(self.executor, ydl.download, [url]) return True, None except Exception as e: return False, str(e) def _progress_hook(self, d): """进度回调函数，可以在这里打印进度或更新数据库状态""" if d['status'] == 'downloading': # 示例：打印进度 if '_percent_str' in d: print(f"\r下载进度: {d['_percent_str']}", end='') elif d['status'] == 'finished': print(f"\n下载完成，文件位于: {d['filename']}")

关键参数解释：

• outtmpl: 输出模板。这是yt-dlp的强大功能，可以使用%(title)s,%(uploader)s,%(upload_date)s等变量自动组织文件夹和文件名。强烈建议使用，避免手动拼接路径。
• format: 格式选择器。bestvideo[height<=1080]+bestaudio/best[height<=1080]表示：优先选择视频高度不超过1080的最佳视频流+最佳音频流进行合并；如果不行，则直接选择高度不超过1080的最佳单文件。这确保了下载的是1080P及以下最高质量。
• merge_output_format: 合并后的格式，通常设为mp4，兼容性好。

3.3.2 实现并发下载与队列管理我们不能一次性发起几百个下载任务，那样会撑爆网络和磁盘IO。需要一个生产者-消费者模型。

import asyncio import aiosqlite from collections import deque class DownloadManager: def __init__(self, db_path, downloader, max_concurrent=3): self.db_path = db_path self.downloader = downloader self.semaphore = asyncio.Semaphore(max_concurrent) # 控制最大并发数 self.task_queue = asyncio.Queue() async def producer(self): """从数据库读取pending状态的任务，放入队列""" async with aiosqlite.connect(self.db_path) as db: async with db.execute("SELECT video_id, title, author FROM videos WHERE status='pending'") as cursor: async for row in cursor: video_id, title, author = row url = f"https://www.bilibili.com/video/{video_id}" await self.task_queue.put((url, video_id, title, author)) # 放入结束信号 for _ in range(self.max_workers): await self.task_queue.put(None) async def consumer(self): """从队列取任务并下载""" while True: item = await self.task_queue.get() if item is None: self.task_queue.task_done() break url, video_id, title, author = item async with self.semaphore: # 控制并发 output_tmpl = f'downloads/{author}/%(title)s.%(ext)s' success, error = await self.downloader.download_video(url, output_tmpl) async with aiosqlite.connect(self.db_path) as db: status = 'completed' if success else 'failed' await db.execute( "UPDATE videos SET status=?, local_path=? WHERE video_id=?", (status, output_tmpl, video_id) # 注意：实际路径需从hook或返回值获取 ) await db.commit() self.task_queue.task_done() async def run(self): """启动下载管理器""" producer_task = asyncio.create_task(self.producer()) consumer_tasks = [asyncio.create_task(self.consumer()) for _ in range(5)] # 5个消费者 await self.task_queue.join() producer_task.cancel() for task in consumer_tasks: task.cancel()

这个管理器确保了最多只有max_concurrent个视频在同时下载，并且能优雅地处理所有任务。

4. 环境搭建与完整流程实现

4.1 项目环境准备

1. 创建项目目录：

   mkdir video_archiver && cd video_archiver python -m venv venv # 创建虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate

2. 安装依赖：

   pip install httpx beautifulsoup4 yt-dlp aiosqlite # 系统级安装FFmpeg (这是必须的) # Ubuntu/Debian: sudo apt install ffmpeg # macOS: brew install ffmpeg # Windows: 从官网下载编译好的二进制文件，将bin目录加入系统PATH。

3. 初始化数据库： 可以使用一个单独的Python脚本init_db.py来创建表结构。

4.2 完整脚本示例：一个简化的B站UP主视频下载器

下面是一个整合了上述核心概念的简化版脚本，用于下载某个UP主的所有视频。

# main.py import asyncio import aiosqlite import httpx from yt_dlp import YoutubeDL from concurrent.futures import ThreadPoolExecutor import time import random # 配置 UP_MID = '12345678' # 替换为目标UP主的MID DB_PATH = 'videos.db' DOWNLOAD_DIR = './downloads' MAX_CONCURRENT_DOWNLOADS = 2 # 并发数，切勿设置过高 REQUEST_DELAY = (1, 3) # 请求间隔秒数，模拟人类操作 class BilibiliCrawler: def __init__(self): self.headers = { 'User-Agent': 'Mozilla/5.0...', 'Referer': 'https://www.bilibili.com/', } self.api_template = 'https://api.bilibili.com/x/space/wbi/arc/search?mid={mid}&pn={pn}&ps=30&order=pubdate' async def fetch_video_list(self, mid): """获取UP主所有视频的bvid列表""" video_list = [] pn = 1 async with httpx.AsyncClient(headers=self.headers, timeout=30) as client: while True: url = self.api_template.format(mid=mid, pn=pn) try: resp = await client.get(url) resp.raise_for_status() data = resp.json() if data['code'] != 0: print(f"API错误: {data['message']}") break vlist = data['data']['list']['vlist'] if not vlist: break # 没有更多视频了 for video in vlist: video_list.append({ 'bvid': video['bvid'], 'title': video['title'], 'author': video['author'], 'created': video['created'], # 时间戳 }) print(f"已获取第{pn}页，共{len(vlist)}个视频。") pn += 1 await asyncio.sleep(random.uniform(*REQUEST_DELAY)) # 礼貌性延迟 except Exception as e: print(f"获取第{pn}页失败: {e}") break return video_list class AsyncVideoDownloader: def __init__(self, max_workers=2): self.executor = ThreadPoolExecutor(max_workers=max_workers) self.loop = asyncio.get_event_loop() async def download(self, bvid, title, author): """下载单个视频""" url = f'https://www.bilibili.com/video/{bvid}' # 使用作者名和标题来组织路径，yt-dlp会自动清理文件名中的非法字符 output_template = f'{DOWNLOAD_DIR}/{author}/%(title)s.%(ext)s' ydl_opts = { 'outtmpl': output_template, 'format': 'bestvideo[height<=1080]+bestaudio/best[height<=1080]', 'merge_output_format': 'mp4', 'quiet': True, 'no_warnings': True, 'socket_timeout': 30, 'retries': 5, 'fragment_retries': 5, 'ignoreerrors': True, # 忽略单个视频的错误，继续下一个 } try: def sync_download(): with YoutubeDL(ydl_opts) as ydl: ydl.download([url]) await self.loop.run_in_executor(self.executor, sync_download) return True, None except Exception as e: return False, str(e) async def main(): # 1. 初始化数据库 async with aiosqlite.connect(DB_PATH) as db: await db.execute('''CREATE TABLE IF NOT EXISTS videos (bvid TEXT PRIMARY KEY, title TEXT, author TEXT, created INTEGER, status TEXT DEFAULT 'pending')''') await db.commit() # 2. 爬取视频列表 print("开始爬取UP主视频列表...") crawler = BilibiliCrawler() videos = await crawler.fetch_video_list(UP_MID) print(f"共发现 {len(videos)} 个视频。") # 3. 将新视频插入数据库 async with aiosqlite.connect(DB_PATH) as db: for v in videos: try: await db.execute( "INSERT OR IGNORE INTO videos (bvid, title, author, created) VALUES (?, ?, ?, ?)", (v['bvid'], v['title'], v['author'], v['created']) ) except Exception as e: print(f"插入数据库失败 {v['bvid']}: {e}") await db.commit() print("视频列表已更新至数据库。") # 4. 下载 pending 状态的视频 downloader = AsyncVideoDownloader() semaphore = asyncio.Semaphore(MAX_CONCURRENT_DOWNLOADS) async def download_task(bvid, title, author): async with semaphore: print(f"开始下载: {title}") success, error = await downloader.download(bvid, title, author) async with aiosqlite.connect(DB_PATH) as db: status = 'completed' if success else 'failed' await db.execute("UPDATE videos SET status=? WHERE bvid=?", (status, bvid)) await db.commit() if success: print(f"下载成功: {title}") else: print(f"下载失败 {title}: {error}") # 获取所有待下载任务 async with aiosqlite.connect(DB_PATH) as db: async with db.execute("SELECT bvid, title, author FROM videos WHERE status='pending'") as cursor: pending_tasks = [row for row in await cursor.fetchall()] print(f"有 {len(pending_tasks)} 个视频待下载。") # 创建并运行所有下载任务 tasks = [download_task(bvid, title, author) for bvid, title, author in pending_tasks] await asyncio.gather(*tasks, return_exceptions=True) print("所有任务处理完毕。") if __name__ == '__main__': asyncio.run(main())

运行这个脚本：

1. 将UP_MID替换为你想要下载的UP主的实际MID（在UP主主页URL中找到）。
2. 确保FFmpeg已安装并可在命令行中调用。
3. 在项目目录下执行python main.py。
4. 程序会先爬取列表、更新数据库，然后开始并发下载。下载的文件会保存在./downloads/UP主名称/目录下。

5. 常见问题、排查技巧与进阶优化

在实际操作中，你一定会遇到各种各样的问题。下面是我在多次运行中总结的“避坑指南”。

5.1 常见错误与解决方案

5.2 进阶优化与功能扩展

1. 增量爬取与定时任务：

   • 将主脚本封装成函数，然后使用系统的定时任务（如Linux的cron，Windows的任务计划程序）每天定时运行。
   • 脚本每次运行都从数据库pending状态开始，实现增量下载。可以在爬取列表时，与数据库对比created时间，只添加新视频。

2. 更精细的过滤：

   • 在爬取列表后、插入数据库前，根据标题、标签、描述进行关键词过滤（如只保留包含“宅舞”、“舞蹈”的视频）。
   • 根据发布时间过滤，只下载最近N天的视频。

3. 元数据丰富与封面下载：

   • yt-dlp提供了--write-thumbnail选项可以下载封面。
   • 可以将视频的描述、标签、弹幕/评论数等信息也存入数据库，方便后期检索和管理。

4. 图形化界面与状态监控：

   • 使用tkinter或PyQt构建一个简单的GUI，用于输入UP主ID、选择下载目录、查看下载队列和进度。
   • 将日志输出到文件，并提供一个实时查看日志的窗口。

5. 分布式与持久化队列：

   • 对于海量下载任务，可以考虑使用Redis作为任务队列，配合Celery实现分布式下载，提升可靠性和扩展性。

6. 尊重与合规的终极策略：

   • 在代码中强制加入随机延迟（time.sleep(random.uniform(2, 5))）。
   • 设置合理的每日/每小时下载数量上限。
   • 最好使用一个独立的、不常用的账号进行此类操作。
   • 最终下载的资源，请在本地妥善保存，勿进行二次分发。

最后一点个人心得：构建这样一个系统，最大的收获不是那几百GB的视频文件，而是在解决一个个具体问题（如反爬、异步调度、错误处理）的过程中，对Python生态和网络协议理解的加深。它像是一个微型的系统工程，涵盖了数据获取、处理、存储、调度的完整链条。开始动手做吧，从最简单的单线程下载一个视频开始，逐步迭代，你会遇到问题，然后解决问题，这才是学习的乐趣所在。