B站视频下载爬虫：从Wbi签名到DASH分片下载全链路解析

简介：这是一份面向Python初学者与爬虫实践者的轻量级B站小视频批量下载工具，解决用户手动逐个保存短视频效率低、缺乏进度反馈的痛点。资源为单文件Python脚本（task_4.py），压缩包仅1个ZIP文件、大小2KB，代码结构清晰：通过循环请求10页B站小视频排行榜JSON接口，解析标题与直链，调用requests流式下载并实时打印进度条，自动创建video目录、过滤非法文件名字符，且内置3–6秒随机延时以降低反爬风险。已有496人学习下载，读者可直接运行即用，快速掌握HTTP请求、JSON解析、文件流写入、异常容错及基础反爬策略等核心技能，是理解网页数据抓取与本地化存储流程的典型教学级案例。

1. B站视频下载爬虫（含进度）：不是“一键下载”，而是把登录态、分P逻辑、断点续传和进度可视化全链路拆开揉碎喂给你

你搜“B站视频下载”，首页弹出来的全是带GUI的exe工具，点几下就完事——但真用过的人心里都清楚：账号被顶号、下载中途卡死、多P视频漏掉中间几集、进度条永远停在99%……这些不是玄学，是底层没处理好 Cookie 刷新机制、分P接口幂等性、HTTP Range 头校验和本地文件写入原子性。这个「B站视频下载爬虫（含进度）」项目，本质是一套可调试、可打断、可复现的 Python 工程化下载方案：它不封装黑匣子，而是把requests.Session怎么保活、aid/bvid/cid三级映射怎么查、dash和flv清单怎么选、ffmpeg合并前怎么校验分片完整性、终端进度条怎么和线程安全的计数器联动，全部摊开在.py文件里。适合两类人：一是想搞懂 B 站 API 实际调用链路的爬虫学习者；二是需要嵌入自有工作流（比如课件自动归档、素材库定时同步）的工程使用者。它不承诺“免登录”，但把登录态失效后的重试路径写死了；它不跳过反爬，但把 UA、Referer、Wbi 签名、时间戳参数的生成逻辑全放源码里——你可以删掉一行就翻车，也能加一行就打日志。这才是真实生产环境该有的样子。

2. 核心模块解析：从 bvid 解析到 dash 清单获取，为什么必须自己手写get_play_info()而非调用第三方 SDK

B站视频下载最常被忽略的致命环节，是「如何稳定拿到真实播放地址」。很多脚本直接硬编码https://api.bilibili.com/x/player/playurl?bvid=...，结果上线三天就 403——因为 B 站早已对未携带wts（时间戳）、w_rid（Wbi 签名）、platform、qn等参数的请求做拦截。这个项目没走捷径，而是把官方 Web 端实际发出的请求完整复刻。我们先看最关键的get_play_info()函数设计逻辑。

2.1 Wbi 签名生成：绕过前端 JS 混淆，用 Python 精确还原签名算法

B站 Web 端所有带鉴权的接口，URL 中的w_rid参数都是通过一段固定 JS 代码生成的。这段 JS 会取 URL 查询参数字符串 + 一个硬编码 salt（如"6a57828f1e724c2d"），再经由md5和base64两次编码。项目中utils/wbi.py将其完全 Python 化：

# utils/wbi.py import hashlib import base64 import urllib.parse WBI_SALT = "6a57828f1e724c2d" def get_wbi_sign(params: dict) -> str: """生成 w_rid 签名，参数字典需已按 key 字典序排序""" query_str = urllib.parse.urlencode(params) mixed = query_str + WBI_SALT md5_hash = hashlib.md5(mixed.encode()).hexdigest() return base64.b64encode(md5_hash.encode()).decode()

注意：params字典必须严格按 key 升序排列（sorted(params.items())），否则签名失败。这是 B 站签名算法的硬性要求，也是多数仿写者翻车的第一步——他们直接urlencode(dict)，却忽略了 Python 字典无序性导致每次生成结果不同。

2.2 bvid → aid → cid 的三级解析链：为什么不能只靠aid？

早期 B 站接口支持aid（av号），但现在新视频默认只发bvid（BV号）。而playurl接口实际需要的是cid（内容ID），它对应具体一集（如正片、花絮、预告）。项目中core/resolve.py构建了完整解析链：

1. bvid_to_aid(bvid)：调用https://api.bilibili.com/x/web-interface/archive/stat?bvid={bvid}获取aid
2. aid_to_cids(aid)：调用https://api.bilibili.com/x/web-interface/view?aid={aid}获取data.pages[]，每个 page 是一个 cid + part 名称
3. cid_to_dash_url(cid, qn=120)：拼接playurl请求参数，注入w_rid、wts、platform=web、qn（清晰度）

关键点在于：同一bvid可能对应多个cid，且pages数组顺序不等于播放顺序（例如第1集可能在pages[2]）。项目在download.py中强制按page.page字段升序排列，而非数组索引：

# download.py pages = sorted(video_info["pages"], key=lambda x: x["page"]) for page in pages: cid = page["cid"] title = page["part"] # 后续下载逻辑...

这避免了“下载了花絮却漏掉正片”的血泪经验。

2.3 dash 清单结构解析：为什么优先选dash而非flv？

B站目前主推 DASH（Dynamic Adaptive Streaming over HTTP）协议，其优势在于：

• 音视频分离：audio和video分两个 URL，可独立选择码率；
• 支持SegmentBase：每个分片有明确duration和indexRange，便于断点续传；
• representation中codecs字段明确标识编码格式（如avc1.640032/mp4a.40.2），规避 ffmpeg 解码失败。

项目中parser/dash_parser.py提取逻辑如下：

# parser/dash_parser.py def parse_dash_manifest(dash_xml: str) -> dict: root = ET.fromstring(dash_xml) ns = {"mpd": "urn:mpeg:dash:schema:mpd:2011"} video_rep = root.find(".//mpd:Representation[@mimeType='video/mp4']", ns) audio_rep = root.find(".//mpd:Representation[@mimeType='audio/mp4']", ns) video_base = video_rep.find("mpd:BaseURL", ns).text audio_base = audio_rep.find("mpd:BaseURL", ns).text # 提取 init segment 和 media template init = video_rep.find("mpd:Initialization", ns).get("range") media_template = video_rep.find("mpd:SegmentTemplate", ns).get("media") return { "video_url": video_base, "audio_url": audio_base, "init_range": init, "media_template": media_template, }

提示：init_range是bytes=0-xxx形式，必须单独下载作为初始化头；media_template中的$Number$占位符需替换为实际分片序号（如0000000001），否则 ffmpeg 无法识别。

3. 下载执行层：多线程分片下载 + 进度共享内存 + 断点续传校验

下载不是简单requests.get().content写文件。B站 dash 分片动辄几百个，单线程慢、无进度、失败即重来。本项目采用「线程池 + 共享计数器 + 分片级校验」三重保障。

3.1 分片任务队列与线程安全计数器

downloader/downloader.py使用concurrent.futures.ThreadPoolExecutor管理下载线程，并通过threading.local()为每个线程绑定独立 session（避免 Cookie 冲突）：

# downloader/downloader.py import threading from concurrent.futures import ThreadPoolExecutor, as_completed _thread_local = threading.local() def get_session() -> requests.Session: if not hasattr(_thread_local, 'session'): _thread_local.session = requests.Session() # 注入通用 headers _thread_local.session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://www.bilibili.com/", }) return _thread_local.session def download_segment(url: str, filepath: Path, start_byte: int, end_byte: int) -> bool: session = get_session() headers = {"Range": f"bytes={start_byte}-{end_byte}"} try: resp = session.get(url, headers=headers, timeout=30) if resp.status_code == 206: # Partial Content with open(filepath, "rb+") as f: f.seek(start_byte) f.write(resp.content) return True except Exception as e: logger.error(f"分片 {url} 下载失败: {e}") return False

参数说明：start_byte和end_byte来自 dash manifest 中SegmentBase@initialization和SegmentList@duration计算得出，确保每个线程只写自己负责的字节区间，避免文件竞争。

3.2 进度共享内存：用multiprocessing.Manager().dict()实现跨进程可视进度

终端进度条需实时反映总完成量，但多线程无法直接共享tqdm对象。项目改用multiprocessing.Manager().dict()创建全局状态字典：

# downloader/progress.py from multiprocessing import Manager class DownloadProgress: def __init__(self): self.manager = Manager() self.state = self.manager.dict({ "total_segments": 0, "downloaded_segments": 0, "total_bytes": 0, "downloaded_bytes": 0, "current_file": "", }) def update(self, seg_count: int = 0, byte_count: int = 0, filename: str = ""): if seg_count: self.state["downloaded_segments"] += seg_count if byte_count: self.state["downloaded_bytes"] += byte_count if filename: self.state["current_file"] = filename

主线程启动tqdm时，传入self.progress.state并定期刷新：

# main.py pbar = tqdm(total=progress.state["total_segments"], desc="分片下载") while pbar.n < pbar.total: time.sleep(0.5) pbar.n = progress.state["downloaded_segments"] pbar.set_postfix({ "file": progress.state["current_file"][-20:], "bytes": f"{progress.state['downloaded_bytes'] / 1024 / 1024:.1f}MB" }) pbar.refresh()

这样即使线程池内部无感知，终端也能秒级更新。

3.3 断点续传校验：每个分片下载后立即比对Content-Range

HTTPRange请求返回的Content-Range响应头形如bytes 1024-2047/1048576，其中末尾是文件总大小。项目强制校验该值是否与 manifest 中声明一致：

# downloader/downloader.py def download_segment(...) -> bool: # ... 上面的请求逻辑 content_range = resp.headers.get("Content-Range", "") if not content_range.startswith(f"bytes {start_byte}-{end_byte}/"): logger.warning(f"分片 {url} Content-Range 不匹配，期望 {start_byte}-{end_byte}，实际 {content_range}") return False # 触发重试 # ... 写入文件

为什么重要？B站 CDN 有时会返回 200（全量）而非 206（分片），若不校验，会导致文件错位、ffmpeg 合并失败。这是线上实测踩出的核心坑之一。

4. 避坑指南：5 条血泪换来的避坑记录，每一条都对应一次凌晨三点的 debug

4.1 现象：下载到 99% 卡住，日志显示ConnectionResetError

原因：B站服务端对单 IP 短时高频请求限流，requests默认连接池复用 TCP 连接，导致后续请求被拒绝。
解决：在get_session()中显式关闭连接复用，并设置max_retries=1防止重试放大压力：

adapter = requests.adapters.HTTPAdapter( pool_connections=10, pool_maxsize=10, max_retries=urllib3.util.Retry( total=1, # 只重试1次 backoff_factor=0.3, allowed_methods={"GET", "HEAD"}, ) ) session.mount("http://", adapter) session.mount("https://", adapter)

4.2 现象：合并后的 MP4 无法播放，vlc 报错moov atom not found

原因：ffmpeg -i video.mp4 -i audio.mp4 -c copy output.mp4要求moov（元数据）在文件开头，但 dash 分片下载后直接拼接，moov在末尾。
解决：强制 ffmpeg 重排moov到开头，增加-movflags +faststart：

ffmpeg -i video.mp4 -i audio.mp4 -c copy -movflags +faststart output.mp4

注意：此操作会重写整个文件，耗时较长，但不可省略。

4.3 现象：多P视频下载顺序错乱，第3集内容被写入第1集文件

原因：pages数组中page.page字段为字符串（如"1"、"10"），直接sorted(pages)按字符串排序得"1", "10", "2"，而非"1", "2", "10"。
解决：排序时转为整型：

pages = sorted(video_info["pages"], key=lambda x: int(x["page"]))

4.4 现象：登录态失效后无限重试，日志刷屏{"code":-101,"message":"账号未登录"}

原因：get_play_info()未对code == -101做特殊处理，导致后续所有请求都失败。
解决：在core/api.py中统一拦截：

def call_api(url: str, params: dict) -> dict: resp = session.get(url, params=params) data = resp.json() if data.get("code") == -101: raise LoginRequiredError("Cookie 失效，请重新登录") return data

并在主流程中捕获该异常，触发登录流程。

4.5 现象：Windows 下路径含中文时报错UnicodeEncodeError: 'gbk' codec can't encode character

原因：Python 3.8+ 在 Windows 控制台默认用gbk编码，但tqdm日志含 Unicode 字符（如part名称）。
解决：启动时强制设置控制台编码为 UTF-8：

# main.py 开头 import sys if sys.platform == "win32": import os os.system("chcp 65001 > nul") # 切换 CMD 为 UTF-8

5. 合并与后处理：从 raw 分片到可播放 MP4，ffmpeg 参数精调与静音检测实战

下载完所有分片只是开始。video.m4s和audio.m4s是原始裸流，必须封装、校验、优化才能交付。本章聚焦三个真实场景：如何避免合并后音画不同步、如何剔除无声片段、如何批量重命名符合规范。

5.1 音画同步校验：用 ffprobe 提取 PTS 时间戳做差值分析

B站 dash 分片存在音视频起始 PTS（Presentation Time Stamp）偏移问题。直接ffmpeg -c copy合并可能导致开头几帧音画不同步。项目中postprocess/sync_checker.py用ffprobe提取首帧 PTS：

# postprocess/sync_checker.py import subprocess import json def get_first_pts(filepath: str) -> tuple[float, float]: cmd = [ "ffprobe", "-v", "quiet", "-show_entries", "frame=pkt_pts_time", "-select_streams", "v:0,a:0", "-of", "json", filepath ] result = subprocess.run(cmd, capture_output=True, text=True) data = json.loads(result.stdout) video_pts = float(data["frames"][0]["pkt_pts_time"]) audio_pts = float(data["frames"][1]["pkt_pts_time"]) return video_pts, audio_pts # 若差值 > 0.1s，则用 ffmpeg 重编码修正 if abs(video_pts - audio_pts) > 0.1: cmd = [ "ffmpeg", "-i", "video.m4s", "-i", "audio.m4s", "-itsoffset", str(audio_pts - video_pts), # 音频偏移 "-i", "audio.m4s", "-map", "0:v:0", "-map", "2:a:0", "-c:v", "copy", "-c:a", "aac", "-movflags", "+faststart", "output_fixed.mp4" ]

参数说明：-itsoffset是音频流相对于视频流的偏移量（秒），正值表示音频延后，负值表示音频提前。-map显式指定音视频来源流，避免 ffmpeg 自动选择错误。

5.2 静音片段剔除：用ffmpeg -af silencedetect批量扫描并裁剪

课程类视频常有片头片尾静音、讲师停顿间隙。项目提供--strip-silence选项，调用ffmpeg静音检测 +ffmpeg裁剪两步：

# 第一步：检测静音区间（输出到 silence.log） ffmpeg -i input.mp4 -af "silencedetect=noise=-50dB:d=0.5" -f null - 2> silence.log # 第二步：解析 log，生成裁剪命令（示例） # [silencedetect @ 0x...] silence_start: 0.234 # [silencedetect @ 0x...] silence_end: 3.567 | silence_duration: 3.333 # → 裁剪掉 [0,3.567]，保留 [3.567, end] ffmpeg -ss 3.567 -i input.mp4 -c copy -avoid_negative_ts make_zero output_clean.mp4

项目中postprocess/silence_remover.py封装了完整解析逻辑，支持阈值noise和最小静音时长d参数化。

5.3 批量重命名与元数据注入：让文件名自带标题、UP主、日期

最终交付文件需可读性强。项目postprocess/metadata_injector.py从 B站 API 获取title、owner.name、pubdate，注入 MP4 的title、artist、date标签：

# postprocess/metadata_injector.py from mutagen.mp4 import MP4, MP4Cover def inject_metadata(filepath: Path, title: str, author: str, pubdate: str): video = MP4(filepath) video["\xa9nam"] = title # title video["\xa9ART"] = author # artist video["\xa9day"] = pubdate # date # 封面图（如有） if cover_path.exists(): with open(cover_path, "rb") as f: video["covr"] = [MP4Cover(f.read(), imageformat=MP4Cover.FORMAT_JPEG)] video.save()

注意：mutagen库对 MP4 标签支持有限，"\xa9nam"是©nam的 UTF-8 编码，必须用此形式，否则写入失败。

6. 生产环境加固技巧：Cookie 自动续期、失败任务持久化、日志分级归档

最后这一章，讲的是真正跑在服务器上、无人值守、周更一次的落地细节。不是“能跑”，而是“敢放那儿不管”。

6.1 Cookie 自动续期：用browser_cookie3同步 Chrome 登录态

手动维护 Cookie 文件太脆弱。项目支持--use-chrome-cookies参数，调用browser_cookie3直接读取本地 Chrome 的CookiesSQLite 数据库：

# utils/chrome_cookie.py import browser_cookie3 import sqlite3 def get_bilibili_cookies() -> dict: cj = browser_cookie3.chrome(domain_name="bilibili.com") cookies = {} for cookie in cj: if cookie.name in ["SESSDATA", "bili_jct", "DedeUserID"]: cookies[cookie.name] = cookie.value return cookies

前提：运行机器需已登录 B站账号的 Chrome 浏览器，且未启用“退出时清除 Cookie”。这是目前最稳定的免交互登录方案。

6.2 失败任务持久化：JSONL 日志记录每个失败分片的完整上下文

当某一分片连续失败 3 次，项目不直接报错退出，而是写入failed_segments.jsonl，每行一个 JSON 对象：

{ "url": "https://upos-sz-mirrorali.bilivideo.com/...", "cid": 123456789, "segment_id": 42, "start_byte": 1048576, "end_byte": 2097151, "error": "ConnectionResetError", "timestamp": "2024-06-15T02:18:33.123Z" }

后续可用--retry-failed failed_segments.jsonl单独重试，避免整批重下。

6.3 日志分级归档：DEBUG 级别进文件，ERROR 级别发邮件（可选）

项目默认将INFO及以上日志写入logs/download_20240615.log，DEBUG级别额外写入logs/debug_20240615.log。同时支持配置 SMTP 发送ERROR：

# config.ini [mail] enabled = true smtp_server = smtp.example.com port = 587 username = bot@example.com password = app_password_here to = admin@example.com

logger模块中SMTPHandler自动监听ERROR级别事件，附带最近 10 行 DEBUG 日志供排查。

从那以后我每次部署到服务器，都强制走一遍--dry-run --log-level DEBUG验证 Cookie 和网络通路；每次新增 UP主订阅，必先--list-pages看pages解析是否正常；每次 ffmpeg 合并后，必用ffprobe -v quiet -show_entries format=duration input.mp4校验时长是否与 B站页面显示一致。这些动作现在已固化成 pre-commit hook，少一步，我就觉得心里没底。希望帮到你。

本文还有配套的精品资源，点击获取