blivedm 上手手记：把 B 站直播间变成你的实时数据源

blivedm 上手手记：把 B 站直播间变成你的实时数据源

【免费下载链接】blivedm获取bilibili直播弹幕，使用WebSocket协议，支持web端和B站直播开放平台两种接口项目地址: https://gitcode.com/gh_mirrors/bl/blivedm

做直播运营的人大概都经历过这样的时刻：一场三个小时的直播结束后，想复盘观众到底在聊什么、礼物集中在哪个时间段、哪位观众是真正的铁粉，结果只能对着回放手动翻弹幕，翻到眼冒金星。主播自己也想摸清观众喜好，却只有"礼物列表"和"弹幕滚屏"两扇窗，看不到全貌。

如果你正好需要解决"实时拿到 B 站直播间的弹幕、礼物、进房、上舰等数据"这件事，那么blivedm这个基于 WebSocket 的 Python 库就是为你准备的。它不刷屏、不轮询，把 B 站直播间的消息一条条推送到你的程序里，本文带你从零把它跑起来，并逐步搭出一个可落地的直播数据监控脚本。

动手之前，先看清一条直播消息里藏着什么

很多人以为"弹幕数据"就是一行文本加一个用户名，其实远不止如此。以 blivedm 解析后的DanmakuMessage为例，一条弹幕对象里除了msg（内容）和uname（用户名），还带上了这些字段：

• 发送者画像：uid、user_level（用户等级）、wealth_level（荣耀等级）、vip/svip（是否老爷）
• 勋章信息：medal_name和medal_level，即观众在哪个主播的直播间挂了什么牌子的粉丝勋章
• 弹幕属性：color（颜色）、font_size、dm_type（0 文本 / 1 表情 / 2 语音）
• 其他：timestamp、privilege_type（是否舰队成员）等

也就是说，你拿到的不是"一行字"，而是一份结构化的用户行为数据。有了用户等级和勋章信息，做"谁是核心观众"的分析就顺手多了。

类似地，礼物消息GiftMessage带gift_name、num、coin_type（金瓜子还是银瓜子）、total_coin（总瓜子数）；上舰消息能给出guard_level（1 总督、2 提督、3 舰长）；醒目留言SuperChatMessage直接给出人民币金额price。这些都是复盘直播收入的关键字段。

先把环境备好，让第一条弹幕跑起来

项目没有发布到 PyPI，所以正确的安装方式是先拉取源码再安装依赖。clone 之后进入目录，安装依赖包即可：

git clone https://gitcode.com/gh_mirrors/bl/blivedm cd blivedm pip install aiohttp Brotli pure-protobuf yarl

依赖只有四个：aiohttp（异步 HTTP 与 WebSocket 客户端）、Brotli（解压弹幕数据流）、pure-protobuf（解析进房等 protobuf 消息）、yarl（URL 解析）。Python 3.8 及以上都能跑。

连接一个直播间，核心代码其实只有四步：建客户端 → 挂处理器 → start → join。参考仓库根目录的sample.py，一个最小的监听脚本长这样：

import asyncio import blivedm async def main(): # 1. 用房间号创建客户端，房间号就是直播间 URL 里的那串数字 client = blivedm.BLiveClient(12235923) # 2. 挂上消息处理器 handler = MyHandler() client.set_handler(handler) # 3. 启动并保持运行 client.start() await client.join() # 继承 BaseHandler，重写 _on_xxx 回调即可接收对应消息 class MyHandler(blivedm.BaseHandler): def _on_danmaku(self, client, message): print(f'[{client.room_id}] {message.uname}：{message.msg}') if __name__ == '__main__': asyncio.run(main())

这段代码做了什么：BLiveClient负责与 B 站弹幕服务器建立 WebSocket 长连接并维持心跳；MyHandler是你的"消息接收器"——只要直播间有人发弹幕，_on_danmaku就会被调用，message是已经解析好的DanmakuMessage对象，直接取字段打印即可。

注意client.join()是等待客户端停止的协程，它会一直挂住，让脚本持续运行。想跑 5 秒后自动退出？参照sample.py的做法：await asyncio.sleep(5)之后调用client.stop()，再await client.join()收尾。

把回调用全：除了弹幕，还能收到什么

BaseHandler已经把 B 站 WebSocket 推送的各种命令分发成了一个个_on_xxx回调，你只需要按需重写。常用的一组：

class MyHandler(blivedm.BaseHandler): # 弹幕 def _on_danmaku(self, client, message): print(f'[{client.room_id}] {message.uname}：{message.msg}') # 礼物 def _on_gift(self, client, message): print(f'[{client.room_id}] {message.uname} 赠送{message.gift_name}x{message.num}' f'（{message.coin_type}瓜子x{message.total_coin}）') # 上舰（guard_level：1 总督 / 2 提督 / 3 舰长） def _on_user_toast_v2(self, client, message): print(f'[{client.room_id}] {message.username} 上舰，等级={message.guard_level}') # 醒目留言（直接带人民币金额） def _on_super_chat(self, client, message): print(f'[{client.room_id}] 醒目留言 ¥{message.price} {message.uname}：{message.message}') # 观众进房 / 关注等互动消息（msg_type 为 1 表示进入房间） def _on_interact_word_v2(self, client, message): if message.msg_type == 1: print(f'[{client.room_id}] {message.username} 进入房间') # 心跳包，可拿到当前人气值 def _on_heartbeat(self, client, message): print(f'[{client.room_id}] 当前人气 {message.popularity}')

这段代码做了什么：它把五类最常见的直播事件各自接到了处理函数上。尤其值得注意两点：_on_user_toast_v2里判断source != 2可以过滤掉"赠送的舰"而只留"付费的舰"，sample.py里就是这么做的；_on_heartbeat中message.popularity是服务器心跳回复里附带的人气值，虽然官方标注已废弃，但作为粗略的热度参考依然够用。

除了这些，还有_on_buy_guard（另一种上舰消息）、_on_super_chat_delete（醒目留言被删除）等回调。所有回调方法在blivedm/handlers.py里都能看到完整的清单和注释，翻一遍就知道全部能力边界。

一次盯五场直播：多房间并行监控

做竞品分析或批量监测时，一个客户端只盯一个房间显然不够。好消息是，多开客户端几乎零成本——每个BLiveClient都是独立的协程任务，共享一个aiohttp.ClientSession和同一个 handler 即可：

import asyncio import aiohttp import blivedm ROOM_IDS = [12235923, 14327465, 21396545] async def main(): # 共享一个 session，复用连接池，也方便统一注入 cookie session = aiohttp.ClientSession() handler = MyHandler() clients = [blivedm.BLiveClient(room_id, session=session) for room_id in ROOM_IDS] for client in clients: client.set_handler(handler) client.start() try: # 同时等待所有客户端 await asyncio.gather(*(client.join() for client in clients)) finally: # 统一收尾 await asyncio.gather(*(client.stop_and_close() for client in clients)) await session.close()

这段代码做了什么：列表推导式批量创建客户端，asyncio.gather让它们并行运行，任何一个房间的弹幕都会进入同一个MyHandler，而client.room_id帮你区分消息来自哪个房间。回调里所有 message 都是线程隔离的 dataclass 对象，直接累加统计即可，不用担心互相污染。

一个性能提示：set_handler的文档里写得很明白——消息处理与网络协程同在一个协程里跑，如果回调里做耗时操作会阻塞收消息。所以弹幕量大的直播间，建议回调只负责"入队"，把真正的统计、落库丢给线程池或消息队列，别在回调里写重逻辑。

聊一聊断线重连和优雅退出

直播 WebSocket 连接断断续续是常态，blivedm 在这块已经帮你兜了底：

• 自动重连：网络异常、认证失败时，客户端会自动重连；重连次数多了还会自动重新走一遍init_room()拿新的 token，防止旧 token 失效连不上。
• 可调的重连策略：set_reconnect_policy接收一个"输入重试次数、返回间隔秒数"的函数，你可以按需实现指数退避。
• 生命周期四件套：start()启动、stop()停止、join()等待停止完成、stop_and_close()停止并释放资源。注意stop_and_close()之后这个客户端就不能再用了，需要重新创建。
• 崩溃兜底：HandlerInterface.on_client_stopped(client, exception)会在客户端停止时被调用，exception非空就说明是异常退出——你可以在那里实现自己的告警或重启逻辑。

web 端直连 vs 开放平台接入，怎么选

blivedm 支持两条完全不同的接入通道，适用场景差别很大：

web 端是最快的路径，跑通原型用它；如果你的应用要长期稳定跑、还想要"开始直播 / 结束直播 / 点赞"这类事件，就走开放平台。后者的接入代码在仓库根目录的open_live_sample.py里，核心骨架如下：

import asyncio import blivedm async def main(): # 换成你在开放平台申请到的真实凭证 client = blivedm.OpenLiveClient( access_key_id='你的 ACCESS_KEY_ID', access_key_secret='你的 ACCESS_KEY_SECRET', app_id=你的_APP_ID, room_owner_auth_code='主播身份码', ) client.set_handler(MyHandler()) client.start() await client.join() class MyHandler(blivedm.BaseHandler): def _on_open_live_danmaku(self, client, message): print(f'[{message.room_id}] {message.uname}：{message.msg}') def _on_open_live_gift(self, client, message): coin_type = '金瓜子' if message.paid else '银瓜子' print(f'[{message.room_id}] {message.uname} 赠送{message.gift_name}x{message.gift_num}（{coin_type}）') def _on_open_live_like(self, client, message): print(f'[{message.room_id}] {message.uname} 点赞了') def _on_open_live_start_live(self, client, message): print(f'[{message.room_id}] 开播了')

这段代码做了什么：OpenLiveClient在连接前会先向开放平台发起"开启项目"请求，拿到场次 ID 和 WebSocket 地址，然后像 web 端一样建立长连接；回调方法名以_on_open_live_开头，与 web 端的_on_区分开，两者可以同时使用。注意开放平台还有一层项目心跳（默认 20 秒发一次），如果长时间没收到，服务器会判定项目失活并停止推送，客户端内部会自动检测并重新开启项目——这也是它适合长期运行的原因之一。

让数据落库：把实时流存成可分析的记录

纯打印没有积累，下一步自然是落库。把上一节的回调稍加改动，弹幕就能持续写入 SQLite：

import sqlite3 from datetime import datetime # 初始化一张弹幕表 conn = sqlite3.connect('live_data.db') conn.execute('''CREATE TABLE IF NOT EXISTS danmaku ( room_id INTEGER, uname TEXT, msg TEXT, user_level INTEGER, medal_name TEXT, medal_level INTEGER, ts INTEGER )''') conn.commit() class MyHandler(blivedm.BaseHandler): def _on_danmaku(self, client, message): conn.execute( 'INSERT INTO danmaku VALUES (?, ?, ?, ?, ?, ?, ?)', (client.room_id, message.uname, message.msg, message.user_level, message.medal_name, message.medal_level, message.timestamp), ) conn.commit()

这段代码做了什么：每收到一条弹幕就插入一行，medal_name和medal_level记录了观众挂的粉丝勋章，user_level记录了用户等级。存上几天后你会发现，这些字段足够支撑不少有意思的分析：

• 弹幕情感分析：用情感词典或现成模型给msg打分，画出"整场直播的情绪曲线"，看哪个环节观众反应最热烈
• 热门话题识别：对弹幕做分词 + 词频统计，一键找出当天的热梗
• 核心观众圈层：按medal_level、user_level、送礼频次排序，识别高价值粉丝
• 礼物收入复盘：礼物表和上舰表按timestamp聚合，就能看到"哪十分钟收入最高"

社区里甚至有人基于本库直接做了一个叫blivechat的弹幕互动工具（README 里可找到它的说明），说明这套数据流的想象空间远比"看弹幕"大得多。

从第一行代码到你的专属直播监控

回顾一下我们走过的路：先看清一条直播消息里能挖出多少字段，然后用四行代码接住了第一条弹幕，再把礼物、上舰、醒目留言、进房全部接入回调，紧接着实现了多直播间并行监控和断线自愈，最后对比了 web 端与开放平台两条接入通道，并把数据沉淀进了 SQLite。

到这里，一个能实时采集、持久化、可分析的 B 站直播数据管道已经在你手里了。接下来想把它做成什么，完全取决于你：挂一个机器人播报大额礼物、写一个自动统计下播收益的小工具、还是攒一个季度数据做观众画像？仓库里的sample.py和open_live_sample.py就是你的起跑线，clone 下来改一改，今晚就能在你自己关注的直播间看到第一条被代码捕获的弹幕。

【免费下载链接】blivedm获取bilibili直播弹幕，使用WebSocket协议，支持web端和B站直播开放平台两种接口项目地址: https://gitcode.com/gh_mirrors/bl/blivedm