用Python爬虫+WebSocket抓取B站直播弹幕：协议拆解与CSV落盘实战

直播间的弹幕，看起来只是一串串飞快滚过的文字，但如果把它当成数据，能挖掘的信息量远比想象中大。弹幕发送频率反映直播间的活跃曲线，弹幕用词反映观众的情绪和关注点，甚至能提前感知到内容节奏的变化。这篇文章分享的是我最近打磨的一个 Python 爬虫小项目——抓取 B 站直播弹幕数据。和网上很多抓网页 HTML 的教程不同，这套代码直接对接直播间实际使用的 WebSocket 弹幕协议，拿到的是结构化 JSON 数据，延迟在秒级以内，而且附完整可运行代码。就算你之前没有系统写过爬虫，只要会一点 Python，把依赖装好、把房间号填上，就能把弹幕一条一条落盘成 CSV，后面做词频分析、情绪分析都很顺手。

先把丑话说在前面：本文的代码只用于学习协议和网络编程，请尊重目标平台的服务条款，不要拿抓到的数据做商业用途，更不要去干扰任何主播的正常直播。爬虫的底线是不影响目标服务，这一点后面我会专门展开。

1. 整体设计思路：弹幕数据怎么拿、能拿来做什么

1.1 弹幕数据的价值与使用场景

弹幕本质上是一种时序文本数据。每一条弹幕自带三个核心字段：发送者、内容、时间。把这三个维度组合起来，能做的事情非常多。

直播活跃度分析是最常见的用法。统计单位时间内的弹幕数量，可以画出一条直播间的热度曲线，甚至能定位到主播的哪个动作引发了弹幕高峰。比如游戏直播间里，弹幕峰值通常出现在团战、抽卡、翻车这些节点上，用数据还原现场节奏很有意思。

内容质量判断也很有价值。通过弹幕关键词的词频变化，可以看出观众对当前话题的关注度。比如新游戏试玩环节，弹幕里出现大量与游戏机制相关的词，说明观众沉浸度高；如果满屏都是"无聊""困了"，那这段内容大概率留不住人。

弹幕还是天然的中文口语语料。做情感分析、梗词挖掘、用户画像研究，弹幕的丰富程度远超普通评论。我在实际项目里的做法是先把弹幕存成 CSV 或 JSONL，再接 jieba 分词和 Counter 做词频统计，整个链路非常短，后面第 5 节会给完整代码。

1.2 为什么必须用 WebSocket 而不是 HTTP 轮询

第一次写直播弹幕爬虫的人，很自然地会想到用 requests 定时请求某个网页，然后从 HTML 里抠弹幕。这条路不是不能走，但走起来非常别扭。

直播弹幕是高频实时数据。如果 1 秒轮询一次，服务器压力大，自己的 IP 也容易触发风控；如果 5 秒轮询一次，弹幕延迟高到没法用，拿在手里已经不是"实时"了。更麻烦的是，网页 HTML 里能看到的是最近一小段弹幕历史，想连续采集，就得不断翻页或者找内部接口，效率和稳定性都很差。

WebSocket 的思路完全不一样。它建立一条长连接，服务器一有弹幕就往这条连接上推，客户端只需要挂在那里收。类比一下：HTTP 轮询像寄信，每封信问一次"有没有新弹幕"，对方再回一封信；WebSocket 像打电话，接通之后对方嘴巴一张你就听到一句，不用反复挂号。

B 站的直播弹幕服务正是走 WebSocket 推流。客户端先发一个认证包表明身份，之后服务端就会源源不断地把弹幕、礼物、进场通知等事件以二进制包的形式推过来。我们要做的核心事情就四件：连上、认证、按时发心跳、解析收到的二进制包。

1.3 技术栈全貌：requests + websocket-client + 自研协议层

整个项目由三部分组成，职责边界非常清楚。

第一层是房间信息获取。直播间 URL 里的那串数字通常是短房间号，不是弹幕服务真正用的真实房间号。需要先通过 HTTP 接口把短号解析成真实房间号，同时拿到弹幕服务器地址和认证 token。这一层用 requests 就够了。

第二层是弹幕长连接。用 websocket-client 库建立 WSS 连接，发送认证包，维持心跳，接收消息。这个库是同步写法，逻辑直观，新手照着抄不容易出问题。

第三层是报文解析层。自己实现 16 字节包头解析、zlib/brotli 解压、JSON 解码。这是整个项目最核心、也是最有学习价值的部分，理解了它，以后接任何基于自定义二进制协议的实时数据源都不慌。

如果你更习惯异步写法，可以把 websocket-client 换成 websockets 库加 asyncio，思路完全一样。本文不展开异步版本，先把同步版讲透。

2. 弹幕协议拆解：先看懂 16 字节包头再说爬虫

2.1 每个收到的二进制包里到底装了什么

先看最关键的内容：服务端推过来的不是一个干净的 JSON 字符串，而是一个二进制包。这个包的格式非常固定，Python 里用 struct 模块按大端序解包即可。

包头固定 16 字节，字段含义如下：

读出包头之后，从offset + header_len开始截取packet_len - header_len字节，就是包体。

举个例子，一条弹幕消息的 operation 是 5，version 可能是 0，此时包体直接就是 UTF-8 编码的 JSON，长这样：

{ "cmd": "DANMU_MSG", "info": [ [0, 1, 25, 12345, 1700000000000], "这条弹幕的内容", [123456, "取个名字", 0, 0, 0] ] }

info[1]是弹幕文本，info[2][1]是用户名。这两个字段就是最常取的数据。除了DANMU_MSG，还有SEND_GIFT（礼物）、INTERACT_WORD（进场）、WATCHED_CHANGE（看过人数变化）等命令，格式类似，解析逻辑可以复用。

2.2 认证包、心跳包、回包，三步建立稳定连接

连上 WebSocket 之后，第一件事不是干等弹幕，而是先发一个认证包。认证包的 operation = 7，包体是一个 JSON，关键字段是roomid（真实房间号）、protover（请求的协议版本）和key（从弹幕服务器接口拿到的 token）。

为什么要带 token？因为弹幕服务器不是大门敞开随便进的。直播间房管可以设置全禁言、粉丝发言门槛，后台也会对连接做合法性校验。带上了合法 token，服务端才会把这个连接当成一个正常的网页端观众连接，并返回认证结果：operation = 8 的回包，包体 JSON 里的code为 0 表示成功。

连接建立之后还需要定期发心跳包，operation = 2，包体为空。这个动作有两个作用：一是让服务端知道这个连接还活着，二是服务端会通过心跳回包（operation = 3）返回当前直播间的实时人气值。心跳间隔通常 30 秒发一次，太频繁会被当成异常连接，太久不发可能被服务端踢下线。

2.3 压缩、粘包与半包：好消息不是一个包只有一条弹幕

这里有个几乎每个人都会踩的坑：认证时如果请求protover=3，服务端后续推过来的大消息包会先用 brotli 算法压缩，包体解压之后并不是一条 JSON，而是好几个完整的"16 字节包头 + 包体"拼接在一起的消息串。

比如主播那边连续来了三条弹幕，服务端可能把它们压缩进同一个包推给我们。我们把这个包解压出来，又要再次按包头解析，拆成三条独立消息。同理，protover=2对应 zlib 压缩，protover=0是不压缩的纯 JSON。这也是为什么解析代码里需要递归处理：外层包是 op=5 且 version 是 2 或 3 时，先解压，再对解压后的字节流调用同样的parse_packets，对拆出来的内层包再做同样的判断。

另一个容易忽略的问题叫粘包和半包。虽然 WebSocket 协议本身有消息边界，但我们在on_message回调里拿到的 bytes 并不一定恰好是一个完整二进制包。TCP 层做过分包、重传之后，可能出现一次回调只给了半个包头，或者一次回调给了两个完整包。解决方法是在客户端内部维护一个 buffer，把每次收到的 bytes 追加进去，然后不停尝试从 buffer 头部解析完整包，解析不出来的剩余部分留到下次回调接着用。代码里我会用parse_packets返回"包列表 + 剩余字节"的方式处理。

2.4 protover 字段：一次选择影响整条数据链路

这个字段值得单独强调。认证包里的protover相当于告诉服务器"我能接收哪种压缩方式的数据"。填 3 表示支持 brotli，服务器会优先用 brotli 压缩；填 2 表示支持 zlib；填 0 表示不支持压缩，所有消息都平文传输。

平文传输看起来简单，但弹幕量大的直播间，一个包可能包含几十条弹幕的 JSON，不压缩会非常消耗带宽。所以实际写代码时我建议至少填 2，既能优雅处理 zlib，依赖也少（zlib 是 Python 标准库自带的）。如果你愿意多装一个 brotli，就直接填 3。我给的代码里两种压缩都兼容，后面可以看到对应的解压分支。

3. 完整可运行代码：从短房间号到 CSV 弹幕库

3.1 依赖安装与版本说明

先装依赖：

pip install websocket-client requests brotli

如果你在 Windows 上执行pip install brotli报错，可以试试把包名改成Brotli再装：

pip install Brotli

三个库的职责很清晰：requests 负责 HTTP 接口拿房间信息和 token；websocket-client 负责建立并维持 WSS 长连接；brotli 负责解压服务端推来的 brotli 数据。如果实在不想装 brotli，把认证包里的protover改成 2 就行，这样服务端只会用 zlib 压缩，Python 标准库自带 zlib，不需要额外安装。

我测试时的 Python 版本是 3.10，理论上 3.7 以上都能跑。

3.2 房间号解析与弹幕服务器配置获取

从直播间地址里看到的那串数字，多数时候是短房间号，不等于弹幕服务真正使用的真实房间号。用短房间号直接连接，认证包大概率会被拒绝。所以第一步先请求房间初始化接口：

import requests def fetch_real_room_id(short_id: int) -> int: """把直播 URL 里的短房间号，解析成真实房间号""" resp = requests.get( 'https://api.live.bilibili.com/room/v1/Room/room_init', params={'id': short_id}, headers={ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) ' 'AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36' }, timeout=10, ) j = resp.json() if j.get('code') != 0: raise RuntimeError(f'房间信息获取失败: code={j.get("code")} msg={j.get("msg")}') return j['data']['room_id']

拿到真实房间号后，再去请求弹幕服务器配置接口，一次性把 token 和 WebSocket 地址都拿回来：

def fetch_danmu_config(real_room_id: int): """获取弹幕服务器地址和认证 token""" 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 (Windows NT 10.0; Win64; x64) ' 'AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36' }, timeout=10, ) j = resp.json() if j.get('code') != 0: raise RuntimeError(f'弹幕配置获取失败: code={j.get("code")} msg={j.get("msg")}') data = j['data'] host = data['host_list'][0] ws_url = f"wss://{host['host']}:{host['wss_port']}/ws/v1/" return data['token'], ws_url

有个细节值得记住：接口返回的host_list是一组服务器地址，取第一个通常就够用。万一第一个连不上，可以按顺序往下尝试，这是最简单的容错策略。

3.3 协议层实现：组包、拆包、解压、解码

接下来是协议层代码，这是全项目最有价值的部分。建议对着第 2 节的字段表一行一行看：

import json import struct import zlib try: import brotli except ImportError: brotli = None HEADER_LEN = 16 OP_HEARTBEAT = 2 # 心跳 OP_HEARTBEAT_REPLY = 3 # 心跳回包（人气值） OP_MESSAGE = 5 # 服务端推送的消息 OP_AUTH = 7 # 认证 OP_AUTH_REPLY = 8 # 认证回包 VER_JSON = 0 VER_POP = 1 VER_ZLIB = 2 VER_BROTLI = 3 def build_packet(body: bytes, version: int = 1, operation: int = OP_AUTH, sequence: int = 1) -> bytes: """构造一个完整的弹幕协议包：16 字节包头 + 包体""" length = HEADER_LEN + len(body) header = struct.pack('>IHHII', length, HEADER_LEN, version, operation, sequence) return header + body def parse_packets(buf: bytes): """从字节流里尽量切出完整包，返回 (包列表, 剩余未处理字节)""" packets = [] offset = 0 while offset + HEADER_LEN <= len(buf): length, header_len, version, operation, sequence = struct.unpack( '>IHHII', buf[offset: offset + HEADER_LEN] ) if length < HEADER_LEN or length > len(buf) - offset: # 数据还没到齐，留到下一次回调 break body = buf[offset + header_len: offset + length] packets.append((version, operation, body, sequence)) offset += length return packets, buf[offset:] def unpack_body(body: bytes, version: int) -> bytes: """按版本对包体解压""" if version == VER_ZLIB: return zlib.decompress(body) if version == VER_BROTLI: if brotli is None: raise RuntimeError('缺少 brotli 库，请先 pip install Brotli') return brotli.decompress(body) raise ValueError(f'未知压缩版本: {version}') def decode_body(body: bytes) -> dict: """解析包体 JSON，兼容服务端偶尔出现的转义引号""" text = body.decode('utf-8', errors='ignore').strip() if not text: return {} try: return json.loads(text) except json.JSONDecodeError: return json.loads(text.replace('\\"', '"'))

注意decode_body最后一行：B 站弹幕 JSON 偶尔会出现整段字符串被转义的情况，直接json.loads会报错。用replace('\\"', '"')把反斜杠引号还原成普通引号，基本都能救回来。这个坑很隐蔽，不处理的话弹幕一多程序就会时不时崩一下。

3.4 DanmakuClient 主类：连接、认证、心跳、路由

有了协议层之后，主类就水到渠成：

import threading import time import websocket class DanmakuClient: def __init__(self, short_id: int, on_danmaku=None): self.short_id = short_id self.on_danmaku = on_danmaku self._buf = b'' self._ws = None self._running = False self._real_room_id = None self._token = None def _handle_open(self, ws): auth_pkt = build_packet( json.dumps({ 'uid': 0, 'roomid': self._real_room_id, 'protover': 3, 'platform': 'web', 'type': 2, 'key': self._token, }, separators=(',', ':')).encode('utf-8'), version=1, operation=OP_AUTH, ) ws.send(auth_pkt, opcode=websocket.ABNF.OPCODE_BINARY) print('[连接] 认证包已发送') def _handle_message(self, ws, message): if not isinstance(message, bytes): return # 粘包 / 半包处理：追加到缓冲区，尽量切出完整包 self._buf += message packets, self._buf = parse_packets(self._buf) for ver, op, body, seq in packets: if op == OP_MESSAGE and ver in (VER_ZLIB, VER_BROTLI): # 压缩包解压后可能包含多个内层包 raw = unpack_body(body, ver) inner_packets, _ = parse_packets(raw) for iv, iop, ibody, iseq in inner_packets: self._route(iop, iv, ibody) else: self._route(op, ver, body) def _route(self, op, ver, body): if op == OP_AUTH_REPLY: code = decode_body(body).get('code') print('[认证] 服务端返回 code =', code) if code != 0: print('[认证] 失败，请检查 token 与真实房间号') elif op == OP_HEARTBEAT_REPLY: # 人气值可能是 4 字节整数，也可能是 JSON if len(body) == 4: popularity = struct.unpack('>I', body)[0] else: popularity = decode_body(body).get('popularity', 0) print(f'[心跳] 当前人气值: {popularity}') elif op == OP_MESSAGE: data = decode_body(body) cmd = data.get('cmd', '') if cmd == 'DANMU_MSG': text = data['info'][1] user = data['info'][2][1] if self.on_danmaku: self.on_danmaku(user, text) def _heartbeat_loop(self): while self._running: try: self._ws.send( build_packet(b'', version=1, operation=OP_HEARTBEAT), opcode=websocket.ABNF.OPCODE_BINARY, ) except Exception as exc: print('[心跳] 发送失败:', exc) time.sleep(30) def start(self): self._real_room_id = fetch_real_room_id(self.short_id) self._token, ws_url = fetch_danmu_config(self._real_room_id) print(f'[初始化] 真实房间号 {self._real_room_id}') print(f'[初始化] 弹幕服务器 {ws_url}') self._ws = websocket.WebSocketApp( ws_url, on_open=self._handle_open, on_message=self._handle_message, on_error=lambda ws, err: print('[错误]', err), on_close=lambda ws, *args: print('[关闭] 连接已断开'), header={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'}, ) self._running = True threading.Thread(target=self._heartbeat_loop, daemon=True).start() self._ws.run_forever() def stop(self): self._running = False if self._ws: self._ws.close()

这里有个特别容易踩的点：认证包本身也要用build_packet组包，而且包头里的 version 字段填 1 而不是 3。因为认证是一次请求，不是服务端主动推送的消息，走的是普通不压缩通道。如果认证包的 version 填成 3，服务端会按压缩数据处理你的认证请求，大概率直接失败。

3.5 主程序与 CSV 落盘：跑起来看到弹幕

弹幕拿到手，光打印到控制台是不够的，落盘成文件才是数据分析的第一步。这里我用 CSV 追加写的方式，每收到一条弹幕就写一行并 flush 一次，避免程序意外退出丢失数据：

import csv import os import time class CsvRecorder: def __init__(self, path): self.path = path if not os.path.exists(path): with open(path, 'w', newline='', encoding='utf-8-sig') as f: csv.writer(f).writerow(['时间', '用户', '弹幕内容']) self._file = open(path, 'a', newline='', encoding='utf-8-sig') self._writer = csv.writer(self._file) def write(self, user, text): self._writer.writerow([ time.strftime('%Y-%m-%d %H:%M:%S'), user, text.replace('\n', ' ').strip(), ]) self._file.flush() def close(self): self._file.close() if __name__ == '__main__': recorder = CsvRecorder('danmaku.csv') def handle_danmaku(user, text): print(f'{user}: {text}') recorder.write(user, text) client = DanmakuClient( short_id=123456, # 改成你要抓的直播间短号 on_danmaku=handle_danmaku, ) try: client.start() except KeyboardInterrupt: print('\n手动停止，正在关闭连接...') client.stop() recorder.close()

运行后如果看到类似下面的输出，就说明整条链路通了：

[初始化] 真实房间号 228888 [初始化] 弹幕服务器 wss://dm.live.bilibili.com:443/ws/v1/ [连接] 认证包已发送 [认证] 服务端返回 code = 0 [心跳] 当前人气值: 10234 用户A: 哈哈哈 用户B: 主播今天状态不错

CSV 文件的编码一定要用utf-8-sig，不要用普通的utf-8。utf-8-sig会在文件开头写入 BOM，常见的表格软件打开才不会乱码，数据后续导入 pandas 也完全兼容。这个小细节能省掉很多不必要的折腾。

4. 实战排坑：认证失败、风控、粘包这些坎怎么过

4.1 高频问题速查表

我把自己和身边朋友踩过的坑整理成了一张表，建议先收藏，遇到问题直接对照：

4.2 一次认证失败的完整排查过程

我第一次跑通这套代码，卡在认证上整整一个下午。现象是：连接能建立，认证包发出去了，但服务端回包 code 一直不是 0。排查过程很有参考价值，我按顺序做了一遍。

第一步，检查房间号。直接把直播间 URL 里的那串数字拿去请求 room_init，发现返回 code 是 0，说明短房间号没问题。但当时没意识到短房间号和真实房间号是有区别的，直接拿短号去连接，这是第一个坑。

第二步，检查 token。打印 getDanmuInfo 的返回，发现 token 字段其实拿到了，但问题出在请求头上。没加 User-Agent 的时候，这个接口直接返回风控错误；加了之后才正常。这类接口对无头请求非常敏感，这是第二个坑。

第三步，检查认证包格式。我用调试工具把发送的字节流打印出来，和协议字段表比对，发现结构体的字节序写错了。小端序解出来的包长是个天文数字，服务端自然认不出来。把所有struct.pack都改成大端序>之后，认证立刻通过。这是第三个坑。

这三个坑分别对应三类最常见的问题：数据不对、请求头不对、二进制格式不对。遇到认证失败，按这个顺序查，基本都能解决。

4.3 爬虫的合规边界与自约束

弹幕协议是平台给网页端用户设计的，我们用代码模拟网页端连接，本质上相当于"替一个观众挂着"。这意味着要主动给自己加约束。

不要同时开几十个连接去刷同一个直播间，那会明显影响平台服务，也违背了做技术分享的本意。不要拿弹幕数据做商业产品，比如实时监控某个主播的负面弹幕并对外售卖。不要绕过直播间本身的限制，比如房管设置的粉丝发言门槛、特殊用户组禁言。抓下来的数据，建议只用于个人学习和学术研究，用完之后及时删除。

我在代码里特意让每个客户端只维持一条连接、心跳间隔 30 秒，不为别的，就是让这个工具对目标平台足够"轻"。做爬虫的人，技术能力越强，越应该知道什么不能碰。

5. 抓下来之后怎么玩：词频统计与实时监控扩展

5.1 弹幕词频统计：CSV 接 jieba 一行搞定

弹幕落盘之后，第一个最简单的分析是词频统计。装一个 jieba：

pip install jieba

然后写一个几行的脚本：

import csv from collections import Counter import jieba def top_words(path, n=20): counter = Counter() with open(path, encoding='utf-8-sig') as f: for row in csv.DictReader(f): content = row['弹幕内容'] for word in jieba.cut(content): word = word.strip() if len(word) >= 2 and not word.isdigit(): counter[word] += 1 return counter.most_common(n) if __name__ == '__main__': for word, cnt in top_words('danmaku.csv'): print(word, cnt)

跑完之后你会发现，一场直播的高频词往往能直接说出内容主题。比如游戏直播可能高频出现角色名、技能名，聊天类直播可能高频出现观众之间的互动用语。这个结果再配上时间维度，还能看出话题的演化过程：前面还在聊 A，后面全在聊 B，转折点往往对应直播里的某个事件。

5.2 更进阶的玩法：实时面板与语义标签

如果你不想等到直播结束才做分析，可以把程序改成异步版本，用 websockets 库把弹幕消息实时推进消息队列，后端接 Redis 做计数，前端用可视化框架做成实时弹幕大屏。弹幕数量、人气值、高频词、礼物事件都可以在一个面板上看到。

再进一步，可以用大语言模型对弹幕做实时聚类和摘要。比如每 5 分钟把这段时间的弹幕批量喂给模型，让它总结出"观众此刻最关心的三个话题"。直播运营者拿到这个信息，可以及时调整内容节奏。不过这部分属于后话，前提是把本文的数据采集链路跑稳定。

5.3 代码维护建议：接口会变，留好扩展点

B 站这类平台的接口时不时会调整，协议也可能升级。给代码留好扩展点很重要。

我的建议是：把协议解析层、HTTP 接口层、数据处理层拆成三个模块，各自独立。协议解析只依赖字节流输入，不关心数据从哪来；HTTP 接口层只负责返回 token、房间号、服务器地址；数据处理层只接收解析好的弹幕对象。这样任何一个接口变动，只需要改对应的一层，不会牵一发动全身。

另外，每次运行前可以先打印关键参数，比如真实房间号、token 前几位、连接的服务器地址。发现异常时，这些日志能帮你快速定位是哪个环节出了问题。我在第 3 节的代码里已经尽量保留了这些打印，实际使用中你也可以根据自己的需要加密或者去掉敏感信息。

最后再分享一个我自己的使用习惯。跑这种采集程序，我从不在一台日常开发电脑上挂着收一整天，而是放到一台低配服务器上，配合进程守护工具做守护，弹幕直接落盘到远程磁盘，第二天再集中做统计。这样既不占用办公电脑，也能保证数据连续不中断。协议解析代码本身并不难，难的是把这条链路稳定地跑起来，并且对每一条异常都有清晰的排查思路。希望这套代码和这些经验，能帮你少踩几个坑。