看直播的时候弹幕总是一行一行飘过去拦都拦不住。要是想把直播间里的弹幕攒下来做分析比如看看主播在哪个节点弹幕最密集、观众都爱刷什么高频词手头就需要一个稳定采集的 Python 爬虫。最近我就折腾了一个小项目专门爬取 B 站直播弹幕数据完整代码放在下面跑起来能实时收到弹幕流也能很方便改造成 CSV 落盘。这个项目适合两类人一是想拿真实数据练手的 Python 爬虫初学者想做数据分析和可视化的开发者。顺着这套逻辑走一遍你会发现直播弹幕的采集并没有想象中那么神秘难点反而藏在协议包头那几个字节里。1. 弹幕从哪儿来先把采集思路理清楚1.1 直播弹幕不是“网页里的静态文本”很多人第一次接触爬虫习惯是打开网页看 Network 面板把 XHR 请求一个个翻出来然后模拟 GET 请求拿 JSON。但直播弹幕和普通接口数据不一样它不会安安静静躺在某个响应里等你取用。观众每一次发送弹幕后台服务器都会通过 WebSocket 长连接把消息实时推送到直播间所有在线用户的客户端里。也就是说采集弹幕的正确姿势不是“轮询接口”而是“接住推送”。这种推送机制很像微信收消息别人发一条你收一条消息到达没有主动拉取的动作只要连接不断消息就会源源不断过来。爬虫要做的就是模拟一个正在看直播的客户端和弹幕服务器建立 WebSocket 连接然后持续接收推送数据并解析出弹幕文本。所以项目的第一步不是写解析函数而是先把“客户端到底连了哪个服务器带什么暗号进去”这个问题搞清楚。B 站直播的前端页面里公开的接口已经把这些信息暴露出来了我们只需要正常访问接口拿参数再用 WebSocket 把连接搭起来。整个过程不涉及任何绕过或私有协议属于正常的公开数据访问范畴个人学习和小规模跑数据是完全够用的。1.2 两个关键接口加一条长连接实际动手之前我整理了整条链路上必须用到的三样东西一个真实房间号、一个弹幕服务器地址、一个连接认证 token。这三个信息分别藏在两个公开接口里拿到之后才能建立 WebSocket 连接。第一个接口是直播间初始化接口用来把用户最容易记的短房间号比如6转换成直播间实时数据的真实房间号room_id。这个转换很有必要因为很多业务接口只认真实房间号。 短号是主播对外展示用的真实号才是服务器内部的数据索引。请求room_init接口后返回 JSON 里有个data.room_id这就是后面所有请求需要的房号。第二个接口是弹幕信息接口用来获取可用弹幕服务器列表和当前连接需要的 token。 请求这个接口后返回 JSON 里的data.host_list会给出几台弹幕服务器的域名、端口和 WSS 端口data.token则是一串加密 token。这串 token 相当于进入直播间的临时凭证WebSocket 连接建立后第一件事就是把它打包发给服务器完成认证。拿完这两个接口的信息剩下的工作就是写一个 WebSocket 客户端主动连上wss://弹幕服务器地址/sub这个长连接地址先发认证包再定时发心跳包同时持续接收服务器推送的二进制消息。很多教程会把注意力放在解析弹幕 JSON 上但我自己的实操经验是真正容易卡壳的地方是协议包头根本没拼对。关于这一点下一节详细展开。2. 协议细节16 字节二进制包头才是真正的关键2.1 手写二进制包头别被 struct 吓退B 站直播弹幕协议用的是二进制封包所有数据都以二进制帧的方式在 WebSocket 里传输。每个封包头部固定占 16 字节后面跟的是业务数据体。如果头部字节拼错服务器要么直接断开连接要么不回任何消息很多新手在这个问题上浪费了大量时间。头部 16 字节的排列顺序是这样的偏移字节数格式含义04int32封包总长度包头长度加包体长度42int16头长度固定位 1662int16协议版本0 或 1 表示不压缩2 表示 zlib 压缩3 表示 brotli 压缩84int32操作码决定这个包是心跳、认证还是弹幕124int32序列号通常填 1用生活类比说这 16 字节就像快递面单总长度告诉你包裹多大头长度告诉你面单占多少协议版本告诉你里面的货物是散装还是压缩件操作码告诉快递员这应该送到哪个部门。缺任何一项包裹都没法正常流转。Python 里构造这个头部非常简单用struct.pack(IHHII, total_len, header_len, protover, op, sequence)即可。表示大端序B 站协议里所有整数都按网络字节序排列这点不要搞混。I对应 4 字节无符号整数H对应 2 字节无符号整数正好对应头部五个字段。我在第一次实现时犯过一个低级错误把struct.pack里的格式字符串写成了IHHII也就是小端序。结果服务器半天没反应抓包后仔细对比才发现字节顺序完全不同。所以这里特别提醒看到协议文档里写“网络字节序”直接闭眼用就对了。2.2 认证包和心跳包的配置逻辑WebSocket 连接建立之后客户端要做的第一件事是发认证包。认证包的操作码是 7协议版本一般填 0包体是一个 JSON 字符串核心字段包括uid、roomid、protover、platform、type、key。roomid是真实房间号key就是前面从弹幕信息接口拿到的 tokenprotover我建议填 2这样服务器会优先用 zlib 压缩弹幕数据减少网络传输量同时避免必须安装 brotli 这个额外依赖。认证包发完之后客户端还不能干等必须开启一个后台线程每 30 秒发送一次心跳包。心跳包操作码是 2包体为空功能就是告诉服务器“我还活着”。如果长时间不发心跳服务器会认为客户端已经掉线直接把连接断开。这个机制和办公室打卡类似不打卡就默认你没来上班过一会儿工位就会被回收。心跳间隔我实测过30 秒是稳定值。有段时间为省流量把间隔拉长到 60 秒结果部分服务器在第 50 秒左右就断开了。后来我直接把心跳固定为 30 秒跑了 3 个小时没有断过一次。还有一点经验如果直播间长时间没人发弹幕服务器也会偶尔推一些进入房间之类的系统消息所以不要因为一段时间没有弹幕就以为程序挂了心跳正常且连接未关闭就是没问题的状态。3. 完整可运行代码从安装到跑通3.1 环境准备与依赖安装这个项目用到的 Python 库不多核心依赖有三个requests用来请求两个公开接口websocket-client用来建立 WebSocket 长连接brotli用来解压极端情况下的 brotli 压缩包。Python 版本建议 3.8 及以上太老的版本对websocket-client的支持不够友好。安装命令我直接放在下面国内网络环境下也可以加-i指定镜像源但这个不是必须操作。pip install requests websocket-client brotli如果运行过程中发现提示缺少websocket模块多半是安装成了websocket这个旧包注意正确包名是websocket-client。老项目里有时候会看到import websocket它真正对应的 pip 包名就是websocket-client不是websocket。代码里我只用websocket.create_connection建立同步连接再用一个独立线程发心跳整体逻辑比WebSocketApp的异步回调更直观适合新手一步步调试。想做长时间稳定采集的可以把核心逻辑包进类里但我们需要先跑通最小版本再谈工程化。3.2 完整代码 B 站直播弹幕采集脚本 用法python bilibili_danmu.py 直播间短号 import json import struct import threading import time import zlib import sys import requests import websocket def pack_packet(body, op, ver1): 构造 B 站直播弹幕协议封包 body: 包体字符串会自动编码为 utf-8 op: 操作码2 心跳7 认证 ver: 协议版本认证包用 0普通包用 1 if isinstance(body, str): body body.encode(utf-8) header struct.pack(IHHII, 16 len(body), 16, ver, op, 1) return header body def unpack_packets(data): 解析一段可能包含多个封包的二进制数据 返回列表元素为 (op, ver, body) packets [] while len(data) 16: total_len, header_len, ver, op, seq struct.unpack(IHHII, data[:16]) if total_len len(data): break body data[header_len:total_len] packets.append((op, ver, body)) data data[total_len:] return packets def handle_packet(op, body): 处理单个业务包 op 5 时body 是弹幕或房间消息的 JSON 数据 if op ! 5: return try: obj json.loads(body) items obj if isinstance(obj, list) else [obj] for item in items: cmd item.get(cmd, ) if not cmd.startswith(DANMU_MSG): continue info item[info] msg info[1] user info[2][1] ts info[0][0] print(f[{ts}] {user}: {msg}) except Exception: pass def main(): if len(sys.argv) 2: print(用法: python bilibili_danmu.py 直播间短号) return short_id sys.argv[1] # 第一步短房间号转真实房间号 init_resp requests.get( https://api.live.bilibili.com/room/v1/Room/room_init, params{id: short_id}, timeout5, ).json() if init_resp[code] ! 0: raise RuntimeError(f房间号无效: {init_resp}) real_room_id init_resp[data][room_id] # 第二步获取弹幕服务器地址和 token danmu_resp requests.get( https://api.live.bilibili.com/xlive/web-room/v1/index/getDanmuInfo, params{id: real_room_id, type: 0}, headers{User-Agent: Mozilla/5.0}, timeout5, ).json() if danmu_resp[code] ! 0: raise RuntimeError(f获取弹幕信息失败: {danmu_resp}) data danmu_resp[data] host_info data[host_list][0] host host_info[host] port host_info[wss_port] token data[token] print(f真实房间号: {real_room_id}, 弹幕服务器: {host}:{port}) # 第三步建立 WebSocket 连接 ws websocket.create_connection( fwss://{host}:{port}/sub, timeout10, enable_multithreadTrue, ) # 发送认证包 auth { uid: 0, roomid: real_room_id, protover: 2, platform: web, type: 2, key: token, } ws.send( pack_packet(json.dumps(auth), op7, ver0), opcodewebsocket.ABNF.OPCODE_BINARY, ) # 后台线程发送心跳包 def heartbeat(): while True: try: ws.send( pack_packet(, op2, ver1), opcodewebsocket.ABNF.OPCODE_BINARY, ) except Exception: break time.sleep(30) threading.Thread(targetheartbeat, daemonTrue).start() print(已连接弹幕服务器开始接收弹幕CtrlC 退出) try: while True: raw ws.recv() if not raw: continue packets unpack_packets(raw) for op, ver, body in packets: if ver in (2, 3): try: if ver 2: decompressed zlib.decompress(body) else: import brotli decompressed brotli.decompress(body) except Exception: continue for sub_op, _, sub_body in unpack_packets(decompressed): handle_packet(sub_op, sub_body) else: handle_packet(op, body) except KeyboardInterrupt: print(\n停止采集) finally: ws.close() if __name__ __main__: main()上面这段代码里我把整个采集流程分成了五步短号转真实号、获取弹幕服务器信息、建立 WebSocket、发送认证包、循环接收弹幕。代码里最值得关注的是unpack_packets函数它循环剥离二进制头部直到解析完当前帧里的每一个封包。因为 WebSocket 的一次 recv 可能同时返回多个连续封包如果只取第一个高频弹幕场景下漏数据是必然的。3.3 运行方式和预期效果把代码保存成bilibili_danmu.py在命令行里执行下面的命令注意把最后的数字换成你想爬的直播间短号python bilibili_danmu.py 6如果参数正确脚本会先输出真实房间号和弹幕服务器地址然后停在那里等待实时弹幕。你在直播间里发一条弹幕脚本控制台里几乎同步就会出现一条消息格式类似这样[1680000000] 小明: 666这个时间戳是弹幕自带的时间单位是秒可以方便后续做时间序列分析。想停止采集就按CtrlC程序会关闭 WebSocket 连接并退出。我实际测试时用的就是一个普通直播间从连接建立到收到第一条弹幕的延迟在 1 秒以内。不过要注意必须保证直播间处于直播状态如果主播已经下播弹幕服务器自然不会有推送内容脚本就会一直安静地挂在那里。这不是程序问题而是没有数据源。4. 运行效果与常见问题排查4.1 高频问题速查表我把自己跑这个项目时踩过的坑以及很多开发者常遇到的问题整理成了一个速查表。遇到问题时先别怀疑人生按表格里的思路排查大概率几分钟就能解决。现象可能原因解决办法连接后完全收不到消息直播间未开播或认证包里的 token 已过期检查直播间状态重新调用接口拿新 token收到消息但弹幕是乱码数据是 brotli 压缩但本地未安装 brotli执行pip install brotli或把认证包的 protover 改为 2连接一会儿就被断开心跳间隔太长服务器判定客户端离线把心跳线程间隔改为 30 秒报错ImportError: No module named websocket安装包名写错使用pip install websocket-client不是websocket获取弹幕信息接口返回非 0 code请求缺少 User-Agent 或参数不对加上浏览器 User-Agent确认真实房间号正确解析 JSON 时报错收到的是系统消息不是标准弹幕包忽略异常或先判断 cmd 是否以DANMU_MSG开头表格里第 2 条是我最常遇到的情况。很多教程清一色把认证包的protover写成 3然后让你装 brotli。如果环境里只有 zlib 可用把这一个字段改成 2 就能解决问题。实测下来zlib 解压速度足够应付普通直播间的弹幕量。4.2 抓包调试和日志采样的经验排查协议类问题最有效的手段不是反复猜原因而是抓原始数据。我建议在代码里临时加一段文件输出把每次 WebSocket 收到的二进制原始数据存成十六进制或者直接把 recev 到的数据长度打印出来。这一步能快速判断连接是否正常、服务器是否有推送。比如你发现ws.recv()每隔 30 秒会返回一个很小的包这是心跳回复说明连接本身是健康的。如果一直没有 op5 的包可能是认证失败也可能是直播间确实没有弹幕。此时再看服务器返回的第一个包是不是 op8确认认证成功就能缩小问题范围。线上调试时我习惯加一个计数器统计每分钟收到的弹幕数量。直播间弹幕量突然下降时这个数字能告诉你是真实流量少了还是程序漏包了。不要裸跑一个后台进程就关机日志和监控能省下大把排查时间。5. 除了打印还能玩什么扩展弹幕采集的边界5.1 弹幕落盘和去重控制台打印只能满足调试需求真要拿数据做分析还得把弹幕存下来。最简单的做法是在主进程中打开一个 CSV 文件弹幕到达时用 Python 的csv模块写入一行。import csv from datetime import datetime # 在 main 里打开文件 writer csv.writer(open(danmu.csv, w, newline, encodingutf-8)) writer.writerow([timestamp, user, content]) # handle_packet 里再加入一行写入 writer.writerow([datetime.now().isoformat(), user, msg])需要注意直播弹幕可能有重复尤其是同一用户在同一秒连发的多条消息。如果你做的是词频分析去重一般不是重点但如果做用户行为序列分析最好把用户 ID 也存下来配合弹幕内容和时间戳能还原出很多有趣的行为模式。弹幕数据里其实还藏着用户 ID、粉丝牌等级、弹幕颜色等信息切分info数组时多打印几层你会发现数据结构比想象的丰富得多。5.2 稳定性和合规建议程序跑久了难免会遇到网络抖动和服务器主动断开。想做成长期任务至少要加两件事一是断线自动重连把连接和认证的逻辑包成一个函数断线后等待几秒再调用一次二是异常捕获要覆盖整个 while 循环否则某一条脏数据就会把整个进程带崩。我见过有人采集 4 小时后因为一个解析异常直接退出前面的数据全白跑。这里也要提醒一句爬虫本质是对公开数据的访问但任何平台的重连接频率和请求次数都应该保持克制。我这个脚本只在连接建立时请求两次公开接口此后所有数据都靠 WebSocket 推送不会对服务器造成额外压力。如果你要大规模并发采集多个直播间请评估好自己的访问频率尊重平台规则把项目控制在个人学习和研究的合理范围内。最后再分享一个我踩过的小坑一开始我把认证包里的protover写成 1结果服务器就是不回弹幕换成 2 之后整个世界清净了。协议这种东西只靠看文档总容易怀疑人生直接抓包验证比你反复猜更高效。希望这份代码能让你少走一点弯路。