企业官网建设流程全解析

看直播的时候，弹幕总是一行一行飘过去，拦都拦不住。要是想把直播间里的弹幕攒下来做分析，比如看看主播在哪个节点弹幕最密集、观众都爱刷什么高频词，手头就需要一个稳定采集的 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 字节的排列顺序是这样的：

用生活类比说，这 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就是前面从弹幕信息接口拿到的 token，protover我建议填 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, ver=1): """ 构造 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}, timeout=5, ).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"}, timeout=5, ).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( f"wss://{host}:{port}/sub", timeout=10, enable_multithread=True, ) # 发送认证包 auth = { "uid": 0, "roomid": real_room_id, "protover": 2, "platform": "web", "type": 2, "key": token, } ws.send( pack_packet(json.dumps(auth), op=7, ver=0), opcode=websocket.ABNF.OPCODE_BINARY, ) # 后台线程发送心跳包 def heartbeat(): while True: try: ws.send( pack_packet("", op=2, ver=1), opcode=websocket.ABNF.OPCODE_BINARY, ) except Exception: break time.sleep(30) threading.Thread(target=heartbeat, daemon=True).start() print("已连接弹幕服务器，开始接收弹幕，Ctrl+C 退出") 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

这个时间戳是弹幕自带的时间，单位是秒，可以方便后续做时间序列分析。想停止采集就按Ctrl+C，程序会关闭 WebSocket 连接并退出。

我实际测试时用的就是一个普通直播间，从连接建立到收到第一条弹幕的延迟在 1 秒以内。不过要注意，必须保证直播间处于直播状态，如果主播已经下播，弹幕服务器自然不会有推送内容，脚本就会一直安静地挂在那里。这不是程序问题，而是没有数据源。

4. 运行效果与常见问题排查

4.1 高频问题速查表

我把自己跑这个项目时踩过的坑，以及很多开发者常遇到的问题整理成了一个速查表。遇到问题时先别怀疑人生，按表格里的思路排查，大概率几分钟就能解决。

表格里第 2 条是我最常遇到的情况。很多教程清一色把认证包的protover写成 3，然后让你装 brotli。如果环境里只有 zlib 可用，把这一个字段改成 2 就能解决问题。实测下来，zlib 解压速度足够应付普通直播间的弹幕量。

4.2 抓包调试和日志采样的经验

排查协议类问题，最有效的手段不是反复猜原因，而是抓原始数据。我建议在代码里临时加一段文件输出，把每次 WebSocket 收到的二进制原始数据存成十六进制，或者直接把 recev 到的数据长度打印出来。这一步能快速判断连接是否正常、服务器是否有推送。

比如你发现ws.recv()每隔 30 秒会返回一个很小的包，这是心跳回复，说明连接本身是健康的。如果一直没有 op=5 的包，可能是认证失败，也可能是直播间确实没有弹幕。此时再看服务器返回的第一个包是不是 op=8，确认认证成功，就能缩小问题范围。

线上调试时我习惯加一个计数器，统计每分钟收到的弹幕数量。直播间弹幕量突然下降时，这个数字能告诉你是真实流量少了，还是程序漏包了。不要裸跑一个后台进程就关机，日志和监控能省下大把排查时间。

5. 除了打印还能玩什么：扩展弹幕采集的边界

5.1 弹幕落盘和去重

控制台打印只能满足调试需求，真要拿数据做分析，还得把弹幕存下来。最简单的做法是在主进程中打开一个 CSV 文件，弹幕到达时用 Python 的csv模块写入一行。

import csv from datetime import datetime # 在 main 里打开文件 writer = csv.writer(open("danmu.csv", "w", newline="", encoding="utf-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 之后整个世界清净了。协议这种东西，只靠看文档总容易怀疑人生，直接抓包验证比你反复猜更高效。希望这份代码能让你少走一点弯路。