简介面向bilibili直播弹幕与礼物信息采集的Python爬虫项目基于selenium库驱动真实浏览器适用于需要从动态网页中自动化提取互动数据的数据分析师、爬虫初学者与直播运营人员。相比依赖静态结构的传统爬虫它通过模拟点击、滚动等操作解决JS动态渲染内容难以抓取的问题为研究观众实时反馈和主播粉丝互动提供了可行方案。压缩包整体仅29KB共5个文件包括核心爬虫脚本.py、中文使用说明.txt、项目详解.md、许可证license及弹幕去重效果图.png代码与文档配套结构清晰便于对照研读。已有233人浏览学习对轻量级爬虫实战具有参考价值。从脚本和说明中读者可以了解selenium启动浏览器、进入直播页面、滚动加载弹幕、解析HTML提取礼物信息并存储数据的关键流程同时学习到轮询等待、内容去重以及规避反爬拦截等实用细节。该示例不仅适用于bilibili也可迁移至其他动态加载站点是一份简单完整的动态网页爬虫入门参考。1. Selenium在bilibili直播爬虫里的真实位置接到这类需求第一个念头往往是“抓HTTP接口就行”真对着bilibili直播间翻一遍DevTools才会发现弹幕和礼物信息的主要通道是一条WebSocket长连接普通Ajax接口只给聚合数据拿不到按秒推的明细。Selenium在这个场景里的位置不是“模拟人点按钮”而是解决启动阶段的问题——把直播间SPA完整渲染出来取出room_id、cookie、弹幕网关host和token再交给真正的WebSocket客户端收流。两个环节分开后断线重连、消息背压、去重入库才挂得上钩。下面的方案适配bilibili直播web端覆盖弹幕和礼物两类数据读完你可以得到一条从直播间URL到结构化流水线的链路改三个配置参数就能切换监控目标适合做实时大屏、弹幕舆情或直播数据仓库。2. 用Selenium抓直播间参数room_id、cookie与弹幕token一条龙2.1 为什么弹幕和礼物信息不适合直接轮询DOM如果只要当前可视区的弹幕文本轮询DOM完全够用一旦落到“爬取弹幕和礼物信息”的量级轮询会有三个绕不开的问题。第一弹幕面板是虚拟滚动容器滚出去的节点会被回收轮询间隔内出现的弹幕只要没被捕捉到就永久丢失事后没有补偿手段。第二礼物消息在页面上停留只有几秒样式和普通弹幕混在一起靠文本匹配分拣SEND_GIFT错误率非常高。第三高频读取DOM会造成持续的渲染压力直播间本身又是高帧率应用Selenium长时间跑下来内存占用会明显上涨。所以常见的做法是把Selenium的职责收敛到“让页面变成能提取参数的状态”实时数据采集交给独立连接这也是这类实时采集工程与一次性网页抓取最重要的分水岭。2.2 最小启动代码Selenium等待直播间初始化完成import json from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC def create_room_driver(room_url: str) - webdriver.Chrome: options webdriver.ChromeOptions() options.add_argument(--window-size1600,900) # 去掉“Chrome 正在受到自动测试软件控制”的提示条避免页面多埋点 options.add_experimental_option(excludeSwitches, [enable-automation]) driver webdriver.Chrome(optionsoptions) driver.get(room_url) # 弹幕面板渲染出来说明直播间SPA主流程已结束 WebDriverWait(driver, 20).until( EC.presence_of_element_located( (By.XPATH, //*[contains(class,chat)]) ) ) return driver def extract_initial_state(driver: webdriver.Chrome) - dict: state driver.execute_script( return window.__INITIAL_STATE__ || window.__NEPTUNE_IS_MY_WAIFU__ || null ) if not state: raise RuntimeError(页面state未挂载大概率是直播间改版或组件懒加载) return json.loads(json.dumps(state))等待条件里我用的这句XPath是通用写法匹配任意class含chat的节点实际页面上可能叫别的名字改版后会直接抛TimeoutException这时不用改代码结构把等待条件换成当前直播间任意一个稳定容器即可。excludeSwitches并不承担反检测职责它只是去掉Chrome的自动化提示条让页面前端少打一些自动化相关埋点间接提高参数提取的稳定性。window.__INITIAL_STATE__是bilibili直播间页面上挂的初始状态对象不同产品线字段名会变所以保留一个Object.keys(window).filter(k k.startsWith(__))的调试入口比硬编码一个对象名更稳妥。2.3 从window状态里抠出room_id再换弹幕tokenstate extract_initial_state(driver) print(list(state.keys())[:20]) # 改版时先看一级键名 room_id state[roomInfo][roomId] up_uid state[userInfo][baseInfo][uid] cookie_str ; .join( f{c[name]}{c[value]} for c in driver.get_cookies() )这一步的要点是room_id和URL里的房间号不一定一致短位房间号在页面内部会映射到真实的room_id弹幕网关认证只认真实room_id所以不能拿URL硬拼。cookie由driver直接序列化出来省去了手动从DevTools复制再粘贴到requests头里的过程。拿到这些参数后请求B站的弹幕配置接口换tokenimport requests def fetch_danmu_config(room_id: int, cookie_str: str) - dict: resp requests.get( https://api.live.bilibili.com/xlive/web-room/v1/index/getDanmuInfo, params{id: room_id, type: 0}, headers{ cookie: cookie_str, referer: fhttps://live.bilibili.com/{room_id}, }, timeout10, ) payload resp.json() if payload[code] ! 0: raise RuntimeError(f弹幕配置接口返回错误: {payload.get(message)}) host payload[data][host_list][0] return { host: host[host], port: host.get(wss_port, 443), token: payload[data][token], }弹幕配置接口返回的host_list是一组网关地址多房间并发采集时可以按room_id做散列把连接分散到不同host上避免全部挤到第一个节点。接口的code必须显式判断登录态失效时很多接口是HTTP 200但code非0只检查状态码会把错误一路带到WebSocket阶段。配置项取值路径用途room_idstate.roomInfo.roomIdWebSocket认证入参up_uidstate.userInfo.baseInfo.uid标记直播间归属cookiedriver.get_cookies()请求弹幕配置接口的鉴权hostgetDanmuInfo.data.host_list[].host弹幕网关域名tokengetDanmuInfo.data.token认证包key字段3. WebSocket弹幕接入与二进制协议解析爬虫后半段才是肉3.1 认证包、心跳包与服务端推送的关系WebSocket连上弹幕网关后不是马上收数据必须按次序做三件事。第一发送认证包内容是JSON形态的认证信息核心字段是roomid、uid、platform固定为web、type固定为2、key使用上一节换到的token认证包的操作码是7。第二立即开始周期心跳操作码为2body为空间隔建议30秒服务端对空闲连接有回收机制不活跃时间超过阈值会被静默关闭onClose事件可能不触发表现就是什么错都不报但数据再也不来。第三服务端推送的操作码是5所有弹幕和礼物消息都包在5号帧里。认证失败最常见的现象不是连接建立失败而是握手成功后几秒内被断开或者立刻收到一条cmd为system_msg的系统通知优先检查room_id与token是否来自同一场直播。3.2 16字节包头与zlib/brotli解压的实现bilibili弹幕WebSocket的包结构是16字节定长包头加变长body包头全是大端字节序。前4字节是包总长包总长自身算在内接着4字节是头部长度固定16再2字节是协议版本1表示JSON明文2表示zlib压缩3表示brotli压缩再4字节是操作码2为心跳3为心跳响应5为推送7为客户端认证最后4字节是递增序号用来检测丢帧。压缩版本解压后的body内部可能又是一组完整包头因此解析必须写成循环或递归层层剥壳。op值方向含义2客户端→服务端心跳3服务端→客户端心跳响应5服务端→客户端业务消息推送7客户端→服务端认证import struct import zlib HEADER struct.Struct(IHHII) # 总长/头长/协议版本/操作码/序号 def wrap_packet(op: int, body: bytes) - bytes: return HEADER.pack(16 len(body), 16, 1, op, 1) body def unpack_loop(data: bytes): out [] offset 0 while offset 16 len(data): total_len, header_len, ver, op, seq HEADER.unpack_from(data, offset) if offset total_len len(data): break body data[offset header_len: offset total_len] if op 5: if ver 2: out.extend(unpack_loop(zlib.decompress(body))) elif ver 3: import brotli out.extend(unpack_loop(brotli.decompress(body))) else: out.append(body.decode(utf-8)) offset total_len return out网上不少教程只解ver1大规模落库后会漏掉压缩帧里包裹的多条消息。哪怕当前房间弹幕量小网关也可能按时间窗口把多条消息合并压缩成一个大包不处理zlib/brotli分支基本等于只采到了真实数据的一部分。brotli是第三方库跑之前要pip install brotli未安装时直接ImportError代码里把它放在分支内部是为了让未压缩场景不依赖这个库。seq递增连续性可以做丢帧检测相邻两条seq空隙大于1说明中间帧被网关丢弃继续读会一直缺数据我的做法是直接断开重连。3.3 备选方案让Selenium 4的CDP直接截WebSocket帧不想维护独立连接的话还有一条更贴合“基于Selenium”的路径给Chrome启用DevTools协议订阅Network.webSocketFrameReceived事件。Selenium 4的Python绑定提供add_cdp_listener事件对象里的response.payloadData就是原始帧字节直接丢给unpack_loop即可连token和心跳逻辑都省了。代价是采集生命周期绑定在driver和页面存活上页面卡死或网络抽风时连接跟着断大弹幕量下CDP事件回调会成为瓶颈更适合调试和小规模抓取。生产环境我还是建议走独立连接两条路并存的价值主要在快速验证协议字段。from selenium.webdriver.common.devtools.v121.network import WebSocketFrameReceived # v121 只是示例实际版本以本地selenium导出模块为准 frames: list[bytes] [] def on_ws_frame(event: WebSocketFrameReceived): frames.extend(unpack_loop(event.response.payloadData)) driver.execute_cdp_cmd(Network.enable, {}) driver.add_cdp_listener(Network.webSocketFrameReceived, on_ws_frame)这里有个异步坑必须提醒add_cdp_listener的回调是同步的直接在回调里写数据库会把事件循环堵死弹幕一多帧就排不上队。正确做法是把unpack_loop的输出塞进asyncio.Queue让单独一个消费协程去写库和下一章的消费者模型对接。4. 礼物信息捕获与并发写库把SEND_GIFT从弹幕流里拆出来4.1 按cmd字段分拣弹幕与礼物消息经过unpack_loop后每条消息是一个字典判断消息类型靠最外层cmd。弹幕消息的cmd以DANMU_MSG开头存在主播中奖、点赞等变体用startswith判断更稳礼物消息的cmd是精确的SEND_GIFT。为了把两类数据落到不同表里定义两个轻量记录结构再写一个转换函数import time from dataclasses import dataclass dataclass class DanmuRecord: room_id: int content: str uid: int uname: str ts: float dataclass class GiftRecord: tid: str room_id: int uname: str gift_name: str num: int price: int ts: float def parse_notify(msg: dict, room_id: int): cmd msg.get(cmd, ) if cmd.startswith(DANMU_MSG): info msg[info] return DanmuRecord( room_idroom_id, contentinfo[1], uidinfo[2][0], unameinfo[2][1], tstime.time(), ) if cmd SEND_GIFT: d msg[data] return GiftRecord( tidd[tid], room_idroom_id, unamed[uname], gift_named[giftName], numd[num], priced[price], tstime.time(), ) return None实际抓包里的msg字段比示例多很多建议在parse_notify入口把msg整体json.dumps进日志等确认哪些字段真正在用再关掉详细日志。这样改版时不用重新抓包直接看日志就能对字段映射。礼物字段里的tid是服务端生成的流水号同一笔礼物广播tid唯一是后面去重的主键。price单位是金瓜子写库前保留原始字段的值后续做金额汇总时按汇率换算才不会丢失精度。4.2 队列、Redis去重与批量入库短时间内弹幕量往往大于数据库写入吞吐直接逐条插入会反复建连。常用做法是固定一个asyncio.Queue做削峰主循环只入队消费协程批量攒批。去重策略分两套礼物的tid天然唯一用Redis的SETNX做精确去重弹幕没有全局唯一id就用房间号时间戳用户id内容哈希做一个近似键。这样能挡掉重连后的重复推送但不算业务级幂等批量写库前还应该在数据库表里加唯一索引做最终兜底。import asyncio import hashlib import redis.asyncio as aioredis r aioredis.Redis(decode_responsesTrue) queue: asyncio.Queue asyncio.Queue(maxsize2000) BATCH 100 def dedup_key(rec) - str: if isinstance(rec, GiftRecord): return fgift:{rec.tid} # 精确去重 return hashlib.md5( f{rec.room_id}:{rec.ts}:{rec.uid}:{rec.content}.encode() ).hexdigest() async def consume(): buf [] while True: rec await queue.get() key dedup_key(rec) ok await r.set(key, 1, nxTrue, ex3600) if not ok: continue buf.append(rec) if len(buf) BATCH: await flush_records(buf) # 替换成对应ORM的批量insert buf.clear()flush_records不写成具体ORM是因为不同项目的存储形态差别很大有直连MySQL的、有写ClickHouse的、有推Kafka的。只要把batch里的记录转成表结构对应的tuple批量executemany或批量写入即可。队列上限2000意味着积压超过这个数时生产方的put会等待用背压替代丢弃直播间瞬间弹幕暴涨时数据不丢、只是采集延迟这是实时采集里比盲目堆并发更稳妥的做法。4.3 常驻进程参数对照表参数建议值调参方向说明队列上限2000改大容忍更高瞬时并发代价是内存峰值上升消费批量100条/批改大减少写库次数单条失败回滚窗口变大去重TTL弹幕60s礼物3600s弹幕挡重连重复礼物覆盖完整直播时长心跳间隔30s小于网关无通信断连阈值太久会掉线Redis连接池20并发高时优先加连接池而不是拉大批量这几项里最容易忽略的是去重TTL。礼物去重窗口设成一小时不是怕重复而是同一笔礼物的广播在服务端可能因重连场景再次下发tid是同一串一小时窗口能保证一次性消费。弹幕的窗口设短因为服务端补推的弹幕通常只在几十秒内到达60秒足够设太长反而可能误杀真正同hash的新弹幕。5. 上线前必调的三个参数与自测方法5.1 无头模式、心跳间隔和批量大小怎么定无头模式最容易踩坑。--headlessnew在最新版Chrome下能正常渲染弹幕面板但部分直播间的页面会读navigator.webdriver这类自动化标记轻则不出弹幕节点重则视频流黑屏建议先在图形模式跑通链路再加无头回归。心跳间隔按前文固定30秒重连退避3秒起、指数增长到30秒封顶避免网关抖动时所有连接同时重连。批量大小100条/批是个中庸值单条失败影响范围小写库延迟也可接受。这三个参数用环境变量注入别写死在源码常量里BATCH_SIZE100 HEARTBEAT30 python -m stream_crawler --room 65.2 冒烟验证发一条测试弹幕看链路链路是否通了最直接的验证是往直播间发一条测试弹幕并确认它出现在消费端日志里。这条测试弹幕同时验证了Selenium启动、state解析、WebSocket认证、unpack_loop解压、parse_notify字段映射、Redis去重、批量写库七层链路比任何单点测试都有效。from selenium.webdriver.common.action_chains import ActionChains def send_test_danmu(driver, textselenium链路测试): try: box driver.find_element(By.CSS_SELECTOR, input[placeholder*弹幕]) ActionChains(driver).move_to_element(box).click().perform() box.send_keys(text) btn driver.find_element(By.CSS_SELECTOR, button[class*send]) btn.click() except Exception: driver.save_screenshot(debug_room.png) raise发送后回到消费侧过滤关键词tail -f crawler.log | grep selenium链路测试日志里出现这条弹幕就算链路闭环没出现就按顺序查四件事driver是否还持有直播间元素、WebSocket连接是否在重连后丢认证、unpack_loop是否走到ver3分支但brotli没装、parse_notify是否因cmd变体返回None。如果页面输入框的class改版上面的选择器会失效把find_element改成XPath匹配包含“发送”文本的按钮即可截图里能看到输入框但找不到发送按钮就说明问题在按钮选择器而不是链路本身。本文还有配套的精品资源点击获取