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_type0 文本 / 1 表情 / 2 语音其他timestamp、privilege_type是否舰队成员等也就是说你拿到的不是一行字而是一份结构化的用户行为数据。有了用户等级和勋章信息做谁是核心观众的分析就顺手多了。类似地礼物消息GiftMessage带gift_name、num、coin_type金瓜子还是银瓜子、total_coin总瓜子数上舰消息能给出guard_level1 总督、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 消息、yarlURL 解析。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_level1 总督 / 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, sessionsession) 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 端直连BLiveClient开放平台OpenLiveClient接入门槛只需房间号零申请需在 B 站直播开放平台申请密钥、建项目、拿主播身份码消息类型弹幕、礼物、上舰、醒目留言、进房等额外支持点赞、直播开始/结束等事件数据可靠性依赖网页端接口字段可能随前端改版变动官方接口格式稳定、字段标准化适合场景快速验证、个人项目、监控任意直播间生产环境、需要完整事件类型的正式应用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 秒发一次如果长时间没收到服务器会判定项目失活并停止推送客户端内部会自动检测并重新开启项目——这也是它适合长期运行的原因之一。让数据落库把实时流存成可分析的记录纯打印没有积累下一步自然是落库。把上一节的回调稍加改动弹幕就能持续写入 SQLiteimport 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
