B站直播弹幕获取 - 用python写一个B站弹幕姬吧 前言关于这个小项目的由来。最开始是想要利用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-34-56-78-1112-1516-说明数据包长度数据包头部长度协议版本操作类型数据包头部长度数据包内容不过这些内容我们并不都需要用到。下面上代码，具体说明在注释中写了：# 将数据包传入： 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)好啦，去直播间发条弹幕看看效果：注意使用的时候加上限制不要对机器人的回复再去回复就行了。。。后记嗯。。。暂时就先写这么多吧，还有什么要补充的以后再说吧。。源码什么的之后再传吧。。

【笔记/学习】c++实现b站弹幕姬 差不多是为了后续某个功能插件的开发，于是开了这么个坑。之后还可以学习下相关知识，同时由于考试腾不出太多时间学习新知识所以拿旧项目顶一下，于是就有了这篇文章。。Step1.查找b站弹幕的http请求随便点开一个b站的直播间，打开f12，点击网络，刷新下，找有没有弹幕的相关请求包。之后可以发现“msg”这个包点开，从内容看，应该就是我们要找的弹幕包了（这里先略去具体的分析过程）然后就是http请求包的具体分析了。从浏览器中可以看到请求的网址，消息头和相关参数，下一步我们就要用c++去模拟请求了。step2.WinInet库的简单使用虽然用c++模拟请求时可以直接用底层的socket去发送请求，但为了方便，所以还是去直接使用相关库了。wininet库有点类似python的request库，这里就简单介绍使用wininet库去请求了。首先是使用wininet库必须包含的头文件：#include <Windows.h> #include <wininet.h> #pragma comment(lib,"wininet.lib")这里本来想贴一个之前学习时对我帮助挺大的一个网站的，结果找不到了，只能凭着自己的记忆写了。。首先，我们看一眼刚才的请求包，得知请求的网址是http://api.live.bilibili.com/ajax/msg，接下来我们用wininet的函数将网址分解。这里简单贴一段示例代码，看完应该就知道这个函数怎么用了：展开查看void CrackUrl() { URL_COMPONENTS uc; char Scheme[1000]; char HostName[1000]; char UserName[1000]; char Password[1000]; char UrlPath[1000]; char ExtraInfo[1000]; uc.dwStructSize = sizeof(uc); uc.lpszScheme = Scheme; uc.lpszHostName = HostName; uc.lpszUserName = UserName; uc.lpszPassword = Password; uc.lpszUrlPath = UrlPath; uc.lpszExtraInfo = ExtraInfo; uc.dwSchemeLength = 1000; uc.dwHostNameLength = 1000; uc.dwUserNameLength = 1000; uc.dwPasswordLength = 1000; uc.dwUrlPathLength = 1000; uc.dwExtraInfoLength = 1000; InternetCrackUrl("http://hoge:henyo@www.cool.ne.jp:8080/masapico/api_sample.index", 0, 0, &uc); printf("scheme: '%s'\n", uc.lpszScheme); printf("host name: '%s'\n", uc.lpszHostName); printf("port: %d\n", uc.nPort); printf("user name: '%s'\n", uc.lpszUserName); printf("password: '%s'\n", uc.lpszPassword); printf("url path: '%s'\n", uc.lpszUrlPath); printf("extra info: '%s'\n", uc.lpszExtraInfo); printf("scheme type: "); switch(uc.nScheme) { case INTERNET_SCHEME_PARTIAL: printf("partial.\n"); break; case INTERNET_SCHEME_UNKNOWN: printf("unknown.\n"); break; case INTERNET_SCHEME_DEFAULT: printf("default.\n"); break; case INTERNET_SCHEME_FTP: printf("FTP.\n"); break; case INTERNET_SCHEME_GOPHER: printf("GOPHER.\n"); break; case INTERNET_SCHEME_HTTP: printf("HTTP.\n"); break; case INTERNET_SCHEME_HTTPS: printf("HTTPS.\n"); break; case INTERNET_SCHEME_FILE: printf("FILE.\n"); break; case INTERNET_SCHEME_NEWS: printf("NEWS.\n"); break; case INTERNET_SCHEME_MAILTO: printf("MAILTO.\n"); break; default: printf("%d\n", uc.nScheme); } }然后是代码：#define URL_STRING L"http://api.live.bilibili.com/ajax/msg"//Bilive API TCHAR szHostName[128]; TCHAR szUrlPath[256]; URL_COMPONENTS crackedURL = { 0 }; crackedURL.dwStructSize = sizeof(URL_COMPONENTS); crackedURL.lpszHostName = szHostName; crackedURL.dwHostNameLength = ARRAYSIZE(szHostName); crackedURL.lpszUrlPath = szUrlPath; crackedURL.dwUrlPathLength = ARRAYSIZE(szUrlPath); InternetCrackUrl(URL_STRING, (DWORD)URL_STRING, 0, &crackedURL);之后是和服务器建立连接：HINTERNET hInternet = InternetOpen(L"Mozilla/5.0 (Windows NT 6.1) AppleWebKit/535.1 (KHTML, like Gecko) Chrome/15.0.849.0 Safari/535.1", INTERNET_OPEN_TYPE_DIRECT, NULL, NULL, 0); if (hInternet == NULL) return -1; HINTERNET hHttpSession = InternetConnect(hInternet, crackedURL.lpszHostName, crackedURL.nPort, NULL, NULL, INTERNET_SERVICE_HTTP, 0, 0); if (hHttpSession == NULL) { InternetCloseHandle(hInternet); std::cout << GetLastError(); return -2; } HINTERNET hHttpRequest = HttpOpenRequest(hHttpSession, L"POST", crackedURL.lpszUrlPath, NULL, L"", NULL, 0, 0); if (hHttpRequest == NULL) { InternetCloseHandle(hHttpSession); InternetCloseHandle(hInternet); return -3; }这三个函数从名字应该就能基本理解它们的作用了，我自己也不是特别精通就不讲了，感觉学过socket编程的话应该不难理解emm。。然后就是向服务器发送请求了。#define _HTTP_ARAC L"Content-Type: application/x-www-form-urlencoded\r\n" char _HTTP_POST[] = "roomid=5322&csrf_token=&csrf=&visit_id=";//roomid parameters. DWORD dwRetCode = 0; DWORD dwSizeOfRq = sizeof(DWORD); if (!HttpSendRequest(hHttpRequest, _HTTP_ARAC, 0, _HTTP_POST, sizeof(_HTTP_POST)) || !HttpQueryInfo(hHttpRequest, HTTP_QUERY_STATUS_CODE | HTTP_QUERY_FLAG_NUMBER, &dwRetCode, &dwSizeOfRq, NULL) || dwRetCode >= 400) { InternetCloseHandle(hHttpRequest); InternetCloseHandle(hHttpSession); InternetCloseHandle(hInternet); return -4; }经过实验，发现在基本的请求头的基础上必须要加上content-type，服务器才能正确的返回json数据，参数方面只需要roomid，其他参数都留空就可以正常使用。HttpSendRequest（）就是发送http请求的函数了，应该不难看懂吧。最后就是接收服务器返回的数据了#define READ_BUFFER_SIZE 1024 std::string strRet = ""; BOOL bRet = FALSE; char szBuffer[READ_BUFFER_SIZE + 1] = { 0 }; DWORD dwReadSize = READ_BUFFER_SIZE; while (true){ bRet = InternetReadFile(hHttpRequest, szBuffer, READ_BUFFER_SIZE, &dwReadSize); if (!bRet || (0 == dwReadSize)){ break; } szBuffer[dwReadSize] = '\0'; strRet.append(szBuffer); }代码应该不难看懂吧，不过最初写这段代码的时候我被坑过。虽然InternetReadFile（）可以指定自己要接收的字节数，但实际每次接收的数据不一定是你写的字节数，最后总会有几个字节是乱码。所以必须用到函数给出的接收到的字节数，在后面手动加一个'\0'才行。这样，我们就得到了想要的数据并储存到string里了。step3.项目的c++实现这里json数据的解析我用的是CJsonObject，用法参考我上篇转载的博文。这里有个需要注意的地方：我们每次的请求实际上返回的是最近的10条弹幕的数据，而我们要的是持续的弹幕姬，所以我的做法是循环进行请求，每次请求后与上次的请求进行比较，打印出不同的数据，来达到想要的效果。typedef struct { int uid; std::string name; std::string time; std::string text; } DM_DATA; DM_DATA old_list[10] = { 0 }; DM_DATA new_list[10] = { 0 };这里简单定义一个结构体，并认为：假如弹幕的发送者uid，发送时间，发送内容都相同的话，就认为这是同一条弹幕，就不打印。如果有人在一秒内发送多条相同的弹幕那我也没办法啦╮(￣▽￣)╭不过这种情况并不常见而且也没多大影响。neb::CJsonObject oJson(strRet); for (int i = 0; i < 10; i++) { oJson["data"]["room"][i].Get("uid", new_list[i].uid); if (new_list[i].uid == 0)break; oJson["data"]["room"][i].Get("nickname", new_list[i].name); oJson["data"]["room"][i].Get("timeline", new_list[i].time); oJson["data"]["room"][i].Get("text", new_list[i].text); } for (int j = 0; j < 10; j++) { int k = 1; for (int i = 0; i < 10; i++) { if (old_list[i].uid == new_list[j].uid && old_list[i].name == new_list[j].name && old_list[i].time == new_list[j].time && old_list[i].text == new_list[j].text)k = 0; } if (k)std::cout << "[" << new_list[j].name << "]" << new_list[j].text << std::endl; } for (int i = 0; i < 10; i++) { old_list[i].uid = new_list[i].uid; old_list[i].name = new_list[i].name; old_list[i].time = new_list[i].time; old_list[i].text = new_list[i].text; } Sleep(3000);neb::CJsonObject oJson(strRet);这条语句用来处理刚刚接收的json数据，然后下面的三个循环应该很好理解，第一个循环用来把接收到的数据存入新数组，第二个循环进行新数组与旧数组的比较，如不相同则打印，第三个循环把新数组的内容存入旧数组中。最后的Sleep(3000);用来防止请求过快被服务器ban。最后我们需要的就是循环了，由于有点偷懒的原因，所以最后就加一句goto label;，把label:放在HttpSendRequest 的前面就可以了。这样程序的设计就基本完成了。不过还需要注意的一点是，服务器返回的数据是utf8编码的，而大多数中文的windows默认是GBK编码，所以直接转换数据会乱码。最开始的时候我从网上copy了个utf8转gbk的函数，不过现在有更简单的方法，直接在代码的最前面加上一句system("chcp 65001");把控制台的编码换成utf8就ok了。最后放个我在github上的这个项目的旧版本吧：https://github.com/panedioic/CPPDanmaku参考资料：[[1]【python】b站直播弹幕获取][2]by 猫先生的早茶[[3]WinInet编程详解]4获取bilibili直播弹幕的WebSocket协议][3]by 炒鸡嗨客协管徐by skilledprogrammer[[4]C++实现Http Post请求][5]by DoubleLi[[5]WinInet使用详解][6]by analogous_love