Python快速上手：用blivedm实时采集B站直播弹幕，搭一个自己的直播间监控脚本
【免费下载链接】blivedm获取bilibili直播弹幕，使用WebSocket协议，支持web端和B站直播开放平台两种接口项目地址: https://gitcode.com/gh_mirrors/bl/blivedm
你可能也遇到过这样的尴尬：喜欢的UP主深夜开播，等你看回放时才发现评论区已经刷了几千条弹幕，那些"关键信息"早就沉底了。与其事后翻找，不如让程序替你蹲守直播间——B站直播弹幕采集这件事，用 Python 生态里的 blivedm 库，十几行代码就能跑通。它基于 WebSocket 长连接，实时性一流，既支持普通 web 端直连，也支持 B 站直播开放平台接入，是搭建直播间互动监控、数据统计脚本的现成底座。
本篇文章会以"做一个能落盘保存弹幕的小工具"为主线，从装环境一路写到多直播间并行和开放平台升级，全程可直接复制运行。
第一步：3分钟完成环境准备与安装
blivedm 需要 Python 3.8 及以上版本，依赖 aiohttp、Brotli 等少量包，项目未发布到 PyPI，所以推荐用 Git 方式安装：
git clone https://gitcode.com/gh_mirrors/bl/blivedm cd blivedm pip install .安装完成后验证一下：
python -c "import blivedm; print(blivedm.__version__)"能打印出版本号（例如1.1.6），就说明环境就绪了。项目的两个官方示例在sample.py（web 端）和open_live_sample.py（开放平台），之后想查完整写法可以直接翻这两个文件。
第二步：搞懂 blivedm 的两个核心角色
在写代码之前，先认识两个搭档：
| 角色 | 类名 | 职责 |
|---|---|---|
| 客户端 | BLiveClient | 负责连上直播间、收消息、自动重连，一个实例管一个房间 |
| 处理器 | BaseHandler | 你继承它，重写_on_xxx回调，拿到解析好的消息对象 |
弹幕、礼物、上舰、醒目留言，都会被库解析成结构清晰的数据类（如DanmakuMessage），回调里直接用message.msg、message.uname这种字段即可，完全不用碰原始 JSON 和二进制分包。
另外有个细节值得注意：客户端和处理器跑在同一个协程里。回调里不要做耗时操作，需要存数据库、发 HTTP 请求时，用asyncio.create_task丢到后台，避免阻塞收消息。
第三步：写出第一个弹幕监控脚本
先来一个最小可用版本：把直播间的每一条弹幕实时打印出来。
import asyncio import blivedm import blivedm.models.web as web_models ROOM_ID = 12235923 # 取直播间URL里的数字，例如 live.bilibili.com/12235923 class DanmakuPrinter(blivedm.BaseHandler): """继承 BaseHandler，重写感兴趣的 _on_xxx 回调""" def _on_danmaku(self, client: blivedm.BLiveClient, message: web_models.DanmakuMessage): # 弹幕到达：打印房间号、用户名、内容 print(f'[{client.room_id}] {message.uname}: {message.msg}') async def main(): # 1. 创建客户端（一个房间一个） client = blivedm.BLiveClient(ROOM_ID) # 2. 挂上自定义处理器 handler = DanmakuPrinter() client.set_handler(handler) # 3. 启动并等待（内部自动处理认证、心跳、掉线重连） client.start() await client.join() if __name__ == '__main__': asyncio.run(main())跑起来后，房间里的滚动弹幕就会一条接一条出现在终端里。如果希望停止监控，调用client.stop()再await client.join()；彻底释放资源则用client.stop_and_close()。
顺手配置 SESSDATA，拿到完整用户名
不登录也能连，但收到的用户名会被打码、UID 会变成 0。想要完整用户信息，把已登录账号的SESSDATA塞进请求会话：
import aiohttp import http.cookies SESSDATA = '这里填你浏览器cookie里的SESSDATA' def build_session(): c = http.cookies.SimpleCookie() c['SESSDATA'] = SESSDATA c['SESSDATA']['domain'] = 'bilibili.com' session = aiohttp.ClientSession() session.cookie_jar.update_cookies(c) return session # 使用时：BLiveClient(ROOM_ID, session=build_session())从 Chrome 开发者工具复制 cookie 时，记得不要勾选"显示已解码的网址"，否则内容会被转义。
第四步：把礼物、上舰、醒目留言一起接住
弹幕只是开始。在同一个BaseHandler子类里继续重写其他回调，就能把直播间的"氪金流水"也收进来：
class LiveMonitor(blivedm.BaseHandler): # 普通弹幕 def _on_danmaku(self, client, message: web_models.DanmakuMessage): print(f'[弹幕] {message.uname}: {message.msg}') # 礼物 def _on_gift(self, client, message: web_models.GiftMessage): print(f'[礼物] {message.uname} 送出 {message.gift_name} x{message.num} ' f'（{message.coin_type}瓜子 x{message.total_coin}）') # 上舰（大航海） def _on_user_toast_v2(self, client, message: web_models.UserToastV2Message): print(f'[上舰] {message.username} 开通大航海，等级={message.guard_level}') # 醒目留言 def _on_super_chat(self, client, message: web_models.SuperChatMessage): print(f'[醒目留言] ¥{message.price} {message.uname}: {message.message}')每个回调收到的都是blivedm.models.web里的数据类，字段名直接查blivedm/models/web.py就能看到注释说明，非常省心。
第五步：实战升级，做一个弹幕落盘小工具
光打印不够"实战"，我们把上面的能力串起来，做一个带 SQLite 落盘的完整 Demo：弹幕、礼物、上舰全部存进本地数据库，方便事后查询统计。
import asyncio import sqlite3 import blivedm import blivedm.models.web as web_models ROOM_IDS = [12235923, 21449083, 23105590] # 支持一次监控多个房间 class SaveHandler(blivedm.BaseHandler): def __init__(self): self.conn = sqlite3.connect('live_events.db') self.conn.execute('''CREATE TABLE IF NOT EXISTS events( room_id INT, kind TEXT, uname TEXT, content TEXT, ts TEXT )''') def _save(self, room_id, kind, uname, content): self.conn.execute( 'INSERT INTO events VALUES (?,?,?,?,?)', (room_id, kind, uname, content, str(asyncio.get_event_loop().time())) ) self.conn.commit() def _on_danmaku(self, client, message: web_models.DanmakuMessage): self._save(client.room_id, 'danmaku', message.uname, message.msg) def _on_gift(self, client, message: web_models.GiftMessage): self._save(client.room_id, 'gift', message.uname, f'{message.gift_name} x{message.num}') def _on_super_chat(self, client, message: web_models.SuperChatMessage): self._save(client.room_id, 'super_chat', message.uname, f'¥{message.price} {message.message}') async def main(): handler = SaveHandler() clients = [blivedm.BLiveClient(room_id) for room_id in ROOM_IDS] for c in clients: c.set_handler(handler) c.start() # 多个客户端一起等待，任一个停止就结束 await asyncio.gather(*(c.join() for c in clients)) if __name__ == '__main__': asyncio.run(main())到这里，一个"多房间直播互动采集 + 本地存储"的小工具就成型了。之后想画弹幕趋势图、算礼物榜单，直接查这张events表即可。
第六步：更进一步，接入B站直播开放平台
web 端适合快速验证，但如果想要点赞、进房、开播/下播这类更完整的事件，以及更标准的数据格式，就应该走 B 站直播开放平台。blivedm 为此提供了OpenLiveClient：
import blivedm import blivedm.models.open_live as open_models client = blivedm.OpenLiveClient( access_key_id='你的ACCESS_KEY_ID', access_key_secret='你的ACCESS_KEY_SECRET', app_id=123456, # 开放平台创建的项目ID room_owner_auth_code='主播身份码', ) class OpenLiveHandler(blivedm.BaseHandler): def _on_open_live_danmaku(self, client, message: open_models.DanmakuMessage): print(f'[{message.room_id}] {message.uname}: {message.msg}') def _on_open_live_gift(self, client, message: open_models.GiftMessage): coin = '金瓜子' if message.paid else '银瓜子' print(f'[{message.room_id}] {message.uname} 送出 {message.gift_name} ' f'x{message.gift_num}（{coin} x{message.price * message.gift_num}）') def _on_open_live_like(self, client, message: open_models.LikeMessage): print(f'[{message.room_id}] {message.uname} 点赞了') def _on_open_live_enter_room(self, client, message: open_models.RoomEnterMessage): print(f'[{message.room_id}] {message.uname} 进入房间') def _on_open_live_start_live(self, client, message: open_models.LiveStartMessage): print(f'[{message.room_id}] 主播开播了：{message.title}')接入流程是：先在 B 站直播开放平台申请密钥、创建项目、拿到主播身份码，填进构造函数即可，客户端会自动处理鉴权签名和项目心跳。
常见问题速答
Q：不填 SESSDATA 有什么影响？A：能连能收，但弹幕里的用户名会变成打码形式，UID 为 0，礼物、上舰消息里的用户信息也会不完整。日常测试没问题，做数据统计建议填上。
Q：直播间房间 ID 在哪里找？A：打开直播间，浏览器地址栏live.bilibili.com/后面的那串数字就是房间 ID。短号、长号都可以，库会自动解析。
Q：网络波动掉线了怎么办？A：客户端内置自动重连，掉线后会按重试策略反复重连，无需自己写重连逻辑。如果想自定义重连间隔，可以用set_reconnect_policy传入自己的间隔策略函数。
Q：web 端和开放平台怎么选？A：快速验证、轻量监控选 web 端，零配置即可跑；需要点赞/开播事件、想要官方稳定接口和标准化数据的生产场景，选开放平台。两者处理器的回调方法是分开的，代码组织也互不干扰。
Q：消息量很大，回调里做耗时操作会怎样？A：会卡住收消息的协程，导致积压甚至掉线。请务必把数据库写入、网络请求等操作放进create_task或丢给线程池处理。
下一步做什么
到这里，你已经能用一个脚本实时监听任意 B 站直播间，并拿到弹幕、礼物、上舰、醒目留言等全量互动数据。接着可以朝这些方向继续发挥：
- 把数据接进 Kafka 或消息队列，做实时流处理；
- 结合弹幕做关键词统计、情感分析、抽奖指令识别；
- 在开播/下播事件里接通知，第一时间提醒自己；
- 用开放平台做主播自己的互动小工具，比如弹幕点歌、自动欢迎。
从"蹲直播翻弹幕"到"让程序帮你盯直播间"，差的只是一个 blivedm。现在就照上面的步骤跑起第一个脚本，把监控权交给代码吧！
【免费下载链接】blivedm获取bilibili直播弹幕，使用WebSocket协议，支持web端和B站直播开放平台两种接口项目地址: https://gitcode.com/gh_mirrors/bl/blivedm
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
