我明白你的问题。B站直播弹幕现在走的是WebSocket协议,F12里直接搜msg是找不到的。你得在Network标签里找WebSocket连接。
关键点就两步:
- 在F12的Network里刷新页面,筛选
WS或WebSocket,找到弹幕的WebSocket连接(通常是sub开头的)。
- 在
Messages标签里看实时数据流,弹幕内容就在里面,格式通常是JSON。
给你个能跑的Python例子,用websockets库。先装库:pip install websockets。
import asyncio
import websockets
import json
import zlib
async def listen_to_bilibili():
# 这是B站直播弹幕的WebSocket地址,room_id换成你要的房间号
room_id = 你的房间号
uri = f"wss://broadcastlv.chat.bilibili.com/sub"
async with websockets.connect(uri) as websocket:
# 第一步:发送认证包
auth_data = json.dumps({
"uid": 0,
"roomid": room_id,
"protover": 2,
"platform": "web",
"clientver": "1.14.3",
"type": 2
}).encode('utf-8')
# 包结构:头部(16字节) + 数据
header = bytearray(16)
# 数据长度
header[0:4] = (len(auth_data) + 16).to_bytes(4, 'big')
# 头部长度固定16
header[4:6] = (16).to_bytes(2, 'big')
# 协议版本,2表示用zlib压缩
header[6:8] = (2).to_bytes(2, 'big')
# 操作码,7表示认证
header[8:12] = (7).to_bytes(4, 'big')
await websocket.send(header + auth_data)
# 第二步:开始心跳,保持连接
heartbeat_data = b''
heartbeat_header = bytearray(16)
heartbeat_header[0:4] = (16).to_bytes(4, 'big')
heartbeat_header[4:6] = (16).to_bytes(2, 'big')
heartbeat_header[6:8] = (1).to_bytes(2, 'big')
heartbeat_header[8:12] = (2).to_bytes(4, 'big')
# 每30秒发一次心跳
async def send_heartbeat():
while True:
await asyncio.sleep(30)
await websocket.send(heartbeat_header + heartbeat_data)
asyncio.create_task(send_heartbeat())
# 第三步:循环接收消息
while True:
message = await websocket.recv()
# 解析消息头
packet_len = int.from_bytes(message[0:4], 'big')
header_len = int.from_bytes(message[4:6], 'big')
ver = int.from_bytes(message[6:8], 'big')
op = int.from_bytes(message[8:12], 'big')
# 处理数据部分
data = message[header_len:packet_len]
if op == 5: # 操作码5是实际消息
if ver == 2: # zlib压缩
data = zlib.decompress(data)
# 压缩包可能包含多个消息,需要循环处理
offset = 0
while offset < len(data):
sub_len = int.from_bytes(data[offset:offset+4], 'big')
sub_body = data[offset+16:offset+sub_len]
offset += sub_len
try:
msg_json = json.loads(sub_body.decode('utf-8', 'ignore'))
if msg_json['cmd'] == 'DANMU_MSG':
# 这里就是你要的弹幕
uname = msg_json['info'][2][1]
text = msg_json['info'][1]
print(f"{uname}: {text}")
except:
pass
else: # 未压缩
try:
msg_json = json.loads(data.decode('utf-8', 'ignore'))
if msg_json['cmd'] == 'DANMU_MSG':
uname = msg_json['info'][2][1]
text = msg_json['info'][1]
print(f"{uname}: {text}")
except:
pass
# 运行
asyncio.run(listen_to_bilibili())
把代码里的你的房间号换成实际房间ID(浏览器地址栏里的数字)。运行后就能看到实时弹幕了。
总结:B站弹幕走WebSocket,得按它的协议发认证包和心跳包才能收到数据。