前言
关于这个小项目的由来。
最开始是想要利用b站的弹幕进行一些互动之类的。原本也有想过可以利用现有的弹幕姬做个插件来解决的，但无奈不会C#，所以只能自己研究b站的弹幕协议。
后来有写过一个C++版本的，不过有一些小问题，这在后文中会提到。
开码
一丶利用 POST 方式获取 B 站直播弹幕
参考：【python】b站直播弹幕获取
首先，随便打开一个b站的直播页面，按F12打开控制台，点进“网络（Network）”标签，刷新一下，然后审计一下里面的内容，可以找到“gethistory”这个文件里面就是我们要的弹幕了。
实际上，仔细观察便不难发现，请求 gethistory 的时候返回的是请求时最近的10条历史弹幕，不过根据这些就可以写出来一个简易的弹幕姬了。具体做法就是每隔一定的时间请求一次，然后与上次的请求做对比。不同的部分就是这段时间新发的弹幕了，这样就可以对弹幕进行一些操作了。
我们点进“headers”标签：
有了这些我们就可以开写一个弹幕姬了。
虽然headers很乱，不过实际上我们在请求弹幕的时候并不需要这么多headers，具体哪些headers是必要的可以用实验试出来，不过具体过程和结果我就直接略去了。最后的代码可以参考：
B站直播弹幕爬取
或我自己写的C++版本：
【笔记/学习】c++实现b站弹幕姬
（代码有点长而且不是本文的重点这里就不放了）
注：之前的时候获取弹幕的URL是：https://api.live.bilibili.com/ajax/msg，不过我写这篇文再去复现的时候发现这个URL已经没了，经过观察发现变成了 https://api.live.bilibili.com/xlive/web-room/v1/dM/gethistory 截至写文时二者都能用。不过这些都不是重点了。
二丶利用 WebSocket 获取 B 站弹幕
前文利用 POST 的方式获取B站弹幕。这种方法虽然简单，但也有不方便的地方。我设的时间间隔为3s，但如果3s内发送的弹幕数量超过了10条，这种方法就会丢失一部分的弹幕，而如果简单的减小时间间隔，不仅会占用更多的网络资源，如果太过频繁的话还可能会被封IP。
而 HTTP 请求的这种缺陷也正好就是 WebSocket 的出现所为了解决的问题。事实上，我们在看B站直播的时候正是通过 WebSocket 的方式与服务器通信的。
让我们继续打开 F12 ：
这个 sub 就是与弹幕服务器通信的 WebSocket 啦。
点进 Message 标签，会看见一大堆东西。不过我们并不需要自己去研究这个通信协议，在 Github 上已经有了B站的API可以直接使用。
弹幕WS协议
API文档使用 JavaScript 写的，不过这并不妨碍我们移植一个 Python 版本的。
由 API 可知，我们与服务器进行通信所发送的数据大多是 json 的数据，偶尔还会有 zlib 数据。所以我们自然需要导入这两个包。我们与服务器使用 WebSocket 进行通信，但原生 Python 并不能直接发送 WebSocket，我们自然也不可能使用 socket 去造轮子。不过好在已经有很多好用的 WebSocket 的库可供我们使用了。
目前常用的 WebSocket 库有： websocket-client, websockets, aiowebsocket 三个。其中 websocket-client 是同步的，因为我们在收弹幕的同时还得要发送心跳包才能不被服务器断开连接，使用异步io会方便一些。所以不用他。另外两个我也都试过。感觉上 aiowebsocket 更稳定一些，所以这里我们使用这个库。
安装：
pip install aiowebsocket除了 aiowebsocket 要安装外，其他的库都是 python 自带的，直接导入就行了。自然不要忘了用了异步操作要加上 asyncio 库哦。
import asyncio
import zlib
from aiowebsocket.converses import AioWebSocket
import json之后我们写好入口函数：
if __name__ == '__main__':
remote = 'wss://broadcastlv.chat.bilibili.com:2245/sub'
try:
asyncio.get_event_loop().run_until_complete(startup(remote))
except KeyboardInterrupt as exc:
pring('Quit.')remote 自然就是API中弹幕服务器的地址了。然后是startup()
roomid = '5322'
data_raw='000000{headerLen}0010000100000007000000017b22726f6f6d6964223a{roomid}7d'
data_raw=data_raw.format(headerLen=hex(27+len(roomid))[2:], roomid=''.join(map(lambda x:hex(ord(x))[2:],list(roomid))))
async def startup(url):
async with AioWebSocket(url) as aws:
converse = aws.manipulator
await converse.send(bytes.fromhex(data_raw))
tasks=[receDM(converse), sendHeartBeat(converse)]
await asyncio.wait(tasks)在连接到弹幕服务器后必须先发一个数据包写出进入的房间，否则连接会被断开。这里我是直接从浏览器抄的。数据包中的包长度必须要正确，所以这里要计算一下包长度。
然后这里先把接受弹幕的 receDM() 和发送心跳包的 sendHeartBeat 先写好。接下来是这两个函数：
hb = '00000010001000010000000200000001'
async def sendHeartBeat(websocket):
while True:
await asyncio.sleep(30)
await websocket.send(bytes.fromhex(hb))
print('[Notice] Sent HeartBeat.')
async def receDM(websocket):
while True:
recv_text = await websocket.receive()
printDM(recv_text)B站的弹幕服务器是如果70秒没有心跳就断开连接，这里是30s发送一次。因为只需要发一个没有内容的数据包就行了，所以这里也是直接从浏览器抄的。。
对于接收到的数据包的处理比较复杂，这里我们单独写一个函数来处理它。
首先，由API我们可以看到每个数据包的头部是怎样的：
| 位置 | 0-3 | 4-5 | 6-7 | 8-11 | 12-15 | 16- |
|---|---|---|---|---|---|---|
| 说明 | 数据包长度 | 数据包头部长度 | 协议版本 | 操作类型 | 数据包头部长度 | 数据包内容 |
不过这些内容我们并不都需要用到。
下面上代码，具体说明在注释中写了：
# 将数据包传入：
def printDM(data):
# 获取数据包的长度，版本和操作类型
packetLen = int(data[:4].hex(),16)
ver = int(data[6:8].hex(),16)
op = int(data[8:12].hex(),16)
# 有的时候可能会两个数据包连在一起发过来，所以利用前面的数据包长度判断，
if(len(data)>packetLen):
printDM(data[packetLen:])
data=data[:packetLen]
# 有时会发送过来 zlib 压缩的数据包，这个时候要去解压。
if(ver == 2):
data = zlib.decompress(data)
printDM(data)
print('db3')
return
# ver 为1的时候为进入房间后或心跳包服务器的回应。op 为3的时候为房间的人气值。
if(ver == 1):
if(op == 3):
print('[RENQI] {}'.format(int(data[16:].hex(),16)))
return
# ver 不为2也不为1目前就只能是0了，也就是普通的 json 数据。
# op 为5意味着这是通知消息，cmd 基本就那几个了。
if(op==5):
try:
jd = json.loads(data[16:].decode('utf-8', errors='ignore'))
if(jd['cmd']=='DANMU_MSG'):
print('[DANMU] ', jd['info'][2][1], ': ', jd['info'][1])
elif(jd['cmd']=='SEND_GIFT'):
print('[GITT]',jd['data']['uname'], ' ', jd['data']['action'], ' ', jd['data']['num'], 'x', jd['data']['giftName'])
elif(jd['cmd']=='LIVE'):
print('[Notice] LIVE Start!')
elif(jd['cmd']=='PREPARING'):
print('[Notice] LIVE Ended!')
else:
print('[OTHER] ', jd['cmd'])
except Exception as e:
pass这样，一个简单的弹幕姬就完成了！
三丶试试搞一些其他的事情吧！
至此，我们已经有了一个可以在控制台输出弹幕内容的弹幕姬了。不过，这并不是结束，有了这个我们就可以利用弹幕搞事情了（
先放个简单的功能：
把弹幕保存至本地
先到如下时间：
import time然后我们把前面的print()函数改掉：
def log(typ, body):
with open('D:/danmu.txt','a') as fd:
fd.write(time.strftime("[%H:%M:%S] ", time.localtime()))
fd.write(body+'\n')这样就可以爬取弹幕到本地了。
利用SAPI朗读弹幕
利用 SAPI 朗读需要导入相应的包：
import win32com.client然后改写log函数：
def log(typ, body):
speak = win32com.client.Dispatch("SAPI.SpVoice")
#创建发声对象
speak.Speak(body)
#使用发生对象读取文字
with open('D:/danmu.txt','a') as fd:
fd.write(time.strftime("[%H:%M:%S] ", time.localtime()))
fd.write(body+'\n')来使python自动朗读弹幕。
一般 Windows 都可以直接使用，不能用的话再上网查吧。。
利用聊天机器人实现自动聊天
首先打开b站一个直播间，发条弹幕截下包：
照着参考，可以大致写出一份发送弹幕的python脚本
import requests
import time
form_data = {
'color': '65532',
'fontsize': '25',
'mode': '1',
'msg': 'test',
'rnd': int(time.time()),
'roomid': '1136753',
'csrf_token': 'cce335cbfa5bfd292a049b813175bd12',
'csrf': 'cce335cbfa5bfd292a049b813175bd12'
}
# 设置cookie值帮助我们在发送弹幕的时候，服务器识别我们的身份
cookie = { '你的cookie（上图红色部分）' }
res = requests.post('https://api.live.bilibili.com/msg/send', cookies=cookie, data=form_data)
print (res.status_code)上面的 csrf 和 csrf_token 在使用的时候最好也换成自己的。
然后可以去注册图灵机器人/思知机器人等API（不推荐图灵，之前还好，后来一去看感觉有点贵），申请到 appid。
具体可以参考我以前写的这篇文章
最后代码差不多是这样：
def talk(msg):
form_data = {
'color': '65532',
'fontsize': '25',
'mode': '1',
'msg': 'test',
'rnd': int(time.time()),
'roomid': roomid,
'csrf_token': 'cce335cbfa5bfd292a049b813175bd12',
'csrf': 'cce335cbfa5bfd292a049b813175bd12'
}
cookie = { '你的cookie（上图红色部分）' }
payload = {
'appid': '你的appid',
'userid': '1234',
}
payload['spoken'] = msg
res1 = requests.get("https://api.ownthink.com/bot", params=payload)
form_data['msg'] = res1.json()['data']['info']['text']
res2 = requests.post('https://api.live.bilibili.com/msg/send', cookies=cookie, data=form_data)好啦，去直播间发条弹幕看看效果：
注意使用的时候加上限制不要对机器人的回复再去回复就行了
。。。
后记
嗯。。。暂时就先写这么多吧，还有什么要补充的以后再说吧。。
源码什么的之后再传吧。。
评论 (0)