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.pyweb 端和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, sessionbuild_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 为此提供了OpenLiveClientimport 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_id123456, # 开放平台创建的项目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} fx{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传入自己的间隔策略函数。Qweb 端和开放平台怎么选A快速验证、轻量监控选 web 端零配置即可跑需要点赞/开播事件、想要官方稳定接口和标准化数据的生产场景选开放平台。两者处理器的回调方法是分开的代码组织也互不干扰。Q消息量很大回调里做耗时操作会怎样A会卡住收消息的协程导致积压甚至掉线。请务必把数据库写入、网络请求等操作放进create_task或丢给线程池处理。下一步做什么到这里你已经能用一个脚本实时监听任意 B 站直播间并拿到弹幕、礼物、上舰、醒目留言等全量互动数据。接着可以朝这些方向继续发挥把数据接进 Kafka 或消息队列做实时流处理结合弹幕做关键词统计、情感分析、抽奖指令识别在开播/下播事件里接通知第一时间提醒自己用开放平台做主播自己的互动小工具比如弹幕点歌、自动欢迎。从蹲直播翻弹幕到让程序帮你盯直播间差的只是一个 blivedm。现在就照上面的步骤跑起第一个脚本把监控权交给代码吧【免费下载链接】blivedm获取bilibili直播弹幕使用WebSocket协议支持web端和B站直播开放平台两种接口项目地址: https://gitcode.com/gh_mirrors/bl/blivedm创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Python快速上手：用blivedm实时采集B站直播弹幕，搭一个自己的直播间监控脚本
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.pyweb 端和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, sessionbuild_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 为此提供了OpenLiveClientimport 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_id123456, # 开放平台创建的项目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} fx{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传入自己的间隔策略函数。Qweb 端和开放平台怎么选A快速验证、轻量监控选 web 端零配置即可跑需要点赞/开播事件、想要官方稳定接口和标准化数据的生产场景选开放平台。两者处理器的回调方法是分开的代码组织也互不干扰。Q消息量很大回调里做耗时操作会怎样A会卡住收消息的协程导致积压甚至掉线。请务必把数据库写入、网络请求等操作放进create_task或丢给线程池处理。下一步做什么到这里你已经能用一个脚本实时监听任意 B 站直播间并拿到弹幕、礼物、上舰、醒目留言等全量互动数据。接着可以朝这些方向继续发挥把数据接进 Kafka 或消息队列做实时流处理结合弹幕做关键词统计、情感分析、抽奖指令识别在开播/下播事件里接通知第一时间提醒自己用开放平台做主播自己的互动小工具比如弹幕点歌、自动欢迎。从蹲直播翻弹幕到让程序帮你盯直播间差的只是一个 blivedm。现在就照上面的步骤跑起第一个脚本把监控权交给代码吧【免费下载链接】blivedm获取bilibili直播弹幕使用WebSocket协议支持web端和B站直播开放平台两种接口项目地址: https://gitcode.com/gh_mirrors/bl/blivedm创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考