三步搭建B站直播数据采集工具：手把手实现弹幕实时监控与礼物统计

三步搭建B站直播数据采集工具手把手实现弹幕实时监控与礼物统计【免费下载链接】blivedm获取bilibili直播弹幕使用WebSocket协议支持web端和B站直播开放平台两种接口项目地址: https://gitcode.com/gh_mirrors/bl/blivedm凌晨一点你负责的直播间突然刷起了一波节奏运营群里却没人截图。等你第二天回看数据只能看到一堆冷冰冰的数字谁带起的节奏、送了什么礼物、弹幕里大家到底在讨论什么全都无从还原。如果有一台程序替你 24 小时盯着直播间把每一条弹幕、每一次送礼都实时记录下来问题是不是就解决了这正是blivedm这个开源库要干的事。它用 Python 获取 B 站直播弹幕底层走 WebSocket 协议同时支持网页端直连和 B 站直播开放平台两种接口。今天我就带你从零写一个能跑通的小工具先监听弹幕再把礼物数据收进来最后扩展到多个直播间同时采集全程大概三十分钟很适合第一次接触直播数据采集的朋友。先想清楚你要的是能用的数据还是稳定的数据动手之前先花两分钟搞明白 blivedm 提供的两条接入路线因为它们决定了你的项目能拿到什么。网页端直连对应BLiveClient。它模拟浏览器去连 B 站的弹幕服务器不需要任何申请只要知道直播间 ID 就能连。缺点是部分接口属于民间用法拿不到用户名全称不登录时用户名会打码、没有官方的 SLA 保障适合个人玩玩、快速验证想法。开放平台接口对应OpenLiveClient。需要你去 B 站开放平台申请开发者密钥、创建项目、拿到主播身份码门槛高一些但换来的是标准化数据格式和更稳定的事件推送比如直播开始/结束、点赞、进场这类 web 端不太容易拿到的消息。适合生产环境、商业项目。我的建议是先用网页端直连把逻辑跑通确认需求稳定后再考虑迁移到开放平台。两条路的回调写法非常接近迁移成本不高。第一步三分钟把环境搭好跑通第一个样例要求 Python 3.8 以上把仓库拉下来git clone https://gitcode.com/gh_mirrors/bl/blivedm cd blivedm pip install -r requirements.txt这个包没有发布到 PyPI直接从 PyPI 装会装到同名但内容不对的版本所以一定要从仓库安装或直接导入源码目录。安装完成后打开页面https://live.bilibili.com/123456URL 里那串数字就是直播间 ID拷贝出来备用。如果你希望收到的弹幕带上完整用户名和 UID再准备一个SESSDATA登录 B 站后从浏览器开发者工具的 Cookie 里复制SESSDATA字段的值即可。它相当于你的登录凭证只用来获取用户信息并不会拿来做别的事。import asyncio import aiohttp ROOM_ID 12235923 # 换成你要监听的直播间 async def main(): async with aiohttp.ClientSession() as session: client blivedm.BLiveClient(ROOM_ID, sessionsession) handler MyHandler() client.set_handler(handler) client.start() await client.join()这段代码先建了一个aiohttp会话再创建一个 web 端客户端把自定义的事件处理器挂上去然后启动并等待。client.join()会一直阻塞到客户端停止所以程序会持续运行下去把收到的消息一条条交给 handler。第二步让每一条弹幕都不被错过上面代码里的MyHandler还没定义。blivedm 的玩法是继承BaseHandler重写以_on_开头的回调方法库会在收到对应类型的消息时自动调用它们。WebSocket 的连接可以理解成一条一直通着电话的线路不是打一次电话、收一次回复就挂断而是连接保持不断开B 站那边每有新消息就顺着这条线实时传过来blivedm 负责把线路里的语音翻译成 Python 对象。import blivedm import blivedm.models.web as web_models class MyHandler(blivedm.BaseHandler): def _on_danmaku(self, client: blivedm.BLiveClient, message: web_models.DanmakuMessage): # 每条弹幕都带房间号、用户名和内容 print(f[{client.room_id}] {message.uname}{message.msg})DanmakuMessage里的字段相当丰富除了uname和msg还有user_level用户等级、medal_name粉丝勋章名、medal_level勋章等级等。你可以按需取用比如统计某个勋章等级以上用户的发言占比。第三步把礼物、上舰、醒目留言也一并收下弹幕只是直播数据的一小块。直播间里真正值钱的信息是钱——谁送了礼物、谁上了舰、谁砸了醒目留言。在同一个 handler 里继续加回调就行def _on_gift(self, client: blivedm.BLiveClient, message: web_models.GiftMessage): # 精确记录每次送礼含瓜子类型与总价值 print(f[{client.room_id}] {message.uname} 送 {message.gift_name} x{message.num} f({message.coin_type} x {message.total_coin})) def _on_user_toast_v2(self, client: blivedm.BLiveClient, message: web_models.UserToastV2Message): if message.source ! 2: print(f[{client.room_id}] {message.username} 上舰等级{message.guard_level}) def _on_super_chat(self, client: blivedm.BLiveClient, message: web_models.SuperChatMessage): print(f[{client.room_id}] 醒目留言 ¥{message.price} {message.uname}{message.message})到这里你其实已经拥有了一套基础但完整的直播数据采集脚本。不过数据只在终端里滚动等节目结束就没了。为了让数据真正留下来把它写进数据库才是正事。顺手升级多直播间并行监听 数据落库做运营分析的人往往不止盯一个房间而是同品类下好几个主播一起对比。blivedm 的客户端是互相独立的你只需把它们放进一个列表共享同一个aiohttp会话这很重要连接池能省下大量资源再统一交给asyncio.gather等它们结束async def monitor_rooms(room_ids: list, session: aiohttp.ClientSession): clients [blivedm.BLiveClient(rid, sessionsession) for rid in room_ids] handler MyHandler() for c in clients: c.set_handler(handler) c.start() try: await asyncio.gather(*(c.join() for c in clients)) finally: await asyncio.gather(*(c.stop_and_close() for c in clients))配合数据落库把弹幕和礼物存进 SQLite之后想怎么分析都行import sqlite3 from collections import Counter conn sqlite3.connect(live_stats.db) conn.execute(CREATE TABLE IF NOT EXISTS danmaku ( room_id INT, uname TEXT, msg TEXT, ts INT)) conn.execute(CREATE TABLE IF NOT EXISTS gifts ( room_id INT, uname TEXT, gift_name TEXT, num INT, total_coin INT)) hot_words Counter() # 弹幕热词统计 def _on_danmaku(self, client, message): words [w for w in message.msg.split() if len(w) 2] hot_words.update(words) conn.execute(INSERT INTO danmaku VALUES (?,?,?,?), (client.room_id, message.uname, message.msg, message.timestamp)) conn.commit() def _on_gift(self, client, message): conn.execute(INSERT INTO gifts VALUES (?,?,?,?,?), (client.room_id, message.uname, message.gift_name, message.num, message.total_coin)) conn.commit()到这里一个多房间弹幕热词 礼物统计的小工具就完整跑通了。你可以在节目结束时把hot_words.most_common(20)打印出来那就是当晚的话题风向标。新手最容易踩的四个坑坑一不填 SESSDATA用户名全是马赛克。不配置登录态也能连上但收到的uname会被打码uid变成 0勋章等用户维度信息也会缺失。如果你只需要有没有人在说话那无所谓要做用户级分析就老老实实配好 SESSDATA。另外复制 Cookie 时别勾选浏览器显示已解码的网址否则值可能被转义导致登录态失效。坑二担心断线。网络抖动、被服务器踢下线都是常态。好在 blivedm 内部有自动重连机制会在断开后按序重试弹幕服务器。如果你的业务要求在断线后做额外处理比如记录断线时长可以重写on_client_stopped回调在客户端停止时重新start()。坑三多房间同时采集机器扛不住。核心教训是多个客户端共用一个aiohttp.ClientSession让连接池和 Cookie 共享而不是每个客户端各建一个。另外回调是异步事件循环里跑的别在里面做慢速的同步 I/O比如每次都查一次数据库批量攒一攒再写性能会好很多。坑四误以为开放平台和 web 端回调一样。两者的消息模型不同web 端回调名是_on_danmaku、_on_gift开放平台对应_on_open_live_danmaku、_on_open_live_gift且开放平台用OpenLiveClient构造参数是access_key_id、access_key_secret、app_id、room_owner_auth_code。看官方样例open_live_sample.py能帮你少走弯路。数据到手之后还能往哪些方向延伸弹幕和礼物只是起点。直播数据采集的价值在于下游怎么用接消息队列把每一条弹幕推给 Kafka 或 RabbitMQ让消费端做实时流计算解耦采集和分析两个环节。弹幕情感分析用现成的情感词典或微调过的模型给弹幕打分识别出直播间里的情绪拐点。可视化大屏弹幕数量趋势、礼物收入排行、观众活跃时段用 ECharts 或 Grafana 渲染成图表运营一眼看懂。自动化互动监测到指定关键词如卡了求个房管时自动触发回复提醒减轻主播盯屏压力。这些方向每一块都够写一篇单独的文章你现在只需要把采集这层地基打好——blivedm 已经把最麻烦的协议、心跳、重连都封装好了剩下的就是你的想象力了。【免费下载链接】blivedm获取bilibili直播弹幕使用WebSocket协议支持web端和B站直播开放平台两种接口项目地址: https://gitcode.com/gh_mirrors/bl/blivedm创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考