pyqt5获取B站的直播弹幕

admin 管理员组

文章数量: 887006

pyqt5获取B站的直播弹幕

闲着没事做了一个B站的直播弹幕抓取很简单很粗糙的程序，并显示在GUI上，在这里记录一下，免得以后忘记开发过程了。

环境：

python：3.6

pyqt：5.11.3

requests：2.19.1

主要参照了，奈何搞不懂协程异步这类的概念，所以使用pyqt5自带的QWebSocket类，网上文档很少都是C++的程序，查看了两天的资料和源码终于搞出来了。

查看了前辈的代码捋一下他给出的思路：

1.先要获得B站直播间的房间号，下图的102就是房间短号，前辈用的aiohttp获取真正的房间号，我用的requests

2.用WebSockets连接B站并发送认证的参数

'wss://broadcastlv.chat.bilibili:2245/sub'

3.连接成功后30s发送一次心跳包，前辈用单独一个循环发送心跳包，我是用QThread分线程发送的

4.收到B站返回的数据，前辈用的单独一个循环轮询，我用的QWebSocket是事件触发的

原理什么的完全不懂，我是按照这个思路搞出来了，在前辈的代码基础上用pyqt5替换。

ssl认证参照了这个，原理不懂照葫芦画瓢

首先是win.py

#win.py
#  -*- coding:utf-8 -*-import json
import struct
import sys,time
import requestsfrom device import *
from PyQt5.QtWidgets import *
from collections import namedtuple
from enum import IntEnum
from PyQt5.QtCore import QUrl
from PyQt5.QtNetwork import QSsl,QSslSocket
from PyQt5.QtWebSockets import QWebSocketROOM_INIT_URL = ''
WEBSOCKET_URL = 'wss://broadcastlv.chat.bilibili:2245/sub'HEADER_STRUCT = struct.Struct('>I2H2I')
HeaderTuple = namedtuple('HeaderTuple', ('total_len', 'header_len', 'proto_ver', 'operation', 'sequence'))class Operation(IntEnum):SEND_HEARTBEAT = 2POPULARITY = 3COMMAND = 5AUTH = 7RECV_HEARTBEAT = 8class win(QWidget):def __init__(self):super(win, self).__init__(parent=None)self.dataRecvws = QWebSocket()self.config = self.dataRecvws.sslConfiguration()#ssl认证之类的self.config.setPeerVerifyMode(QSslSocket.VerifyNone)self.config.setProtocol(QSsl.TlsV1SslV3)self.dataRecvws.setSslConfiguration(self.config)self.dataRecvws.disconnected.connect(self.onDisconnect)#断开连接# self.dataRecvws.textMessageReceived.connect(self.onTextReceived)self.dataRecvws.binaryMessageReceived.connect(self.onBinaryReceived)#数据接收self.dataRecvws.connected.connect(self.onConnected)#连接self.th = heart_beat()self.resize(300,400)self.setWindowFlags(Qt.WindowTitleHint|Qt.WindowCloseButtonHint)#Qt.CustomizeWindowHint|self.room = QLineEdit('102')self.room.setPlaceholderText('房间号')self.sure = QPushButton('确定')lab0 = QLabel('人数')self.num = QLineEdit('0')self.num.setFixedWidth(50)self.num.setReadOnly(True)self.text = QTextEdit()self.text.setReadOnly(True)#self.text.document().setMaximumBlockCount(100)self.sure.clicked.connect(self.set_open)hbox1 = QHBoxLayout()hbox1.addWidget(self.room)hbox1.addWidget(self.sure)hbox1.addWidget(lab0)hbox1.addWidget(self.num)vbox1 = QVBoxLayout(self)vbox1.addLayout(hbox1)vbox1.addWidget(self.text)def send_auth(self, uid=0, room_id=0):auth_params = {'uid': uid,'roomid': room_id,'protover': 1,'platform': ' web','clientver': '1.4.0'}ret = self.dataRecvws.sendBinaryMessage(self.make_packet(auth_params, Operation.AUTH))print(ret)def onConnected(self):#print("DataReveive websocket is already connect!")self.text.insertHtml("<p style='color:red;'>---------------Connect!---------------</p>")self.text.append('')self.connectStatus = Trueself.send_auth(0, self.get_room_id(int(self.room.text())))self.th.set_heart(self.dataRecvws, self.make_packet({}, Operation.SEND_HEARTBEAT))self.th.start()def onBinaryReceived(self, message):print("------------------data2----------------")# print(message)offset = 0while offset < len(message):try:header = HeaderTuple(*HEADER_STRUCT.unpack_from(message, offset))except struct.error:breakif header.operation == Operation.POPULARITY:popularity = int.from_bytes(message[offset + HEADER_STRUCT.size:offset + HEADER_STRUCT.size + 4], 'big')# print('popularity',popularity)self.get_popularity(popularity)elif header.operation == Operation.COMMAND:body = message[offset + HEADER_STRUCT.size: offset + header.total_len]body = json.loads(str(body,encoding='utf-8'))self.handle_command(body)elif header.operation == Operation.RECV_HEARTBEAT:self.dataRecvws.sendBinaryMessage(self.make_packet({}, Operation.SEND_HEARTBEAT))else:body = message[offset + HEADER_STRUCT.size: offset + header.total_len]print('未知包类型：', header, body, file=sys.stderr)offset += header.total_lendef set_open(self):if self.room.text():print('set Open!')self.dataRecvws.open(QUrl(WEBSOCKET_URL))def onDisconnect(self):#print("DataReceive websocket is disconnected!!!")self.text.insertHtml("<p style='color:red;'>---------------Disconnect!---------------</p>")self.text.append('')self.th.exit()self.reconnect()# self.m_timer.start(3000)def reconnect(self):self.text.insertHtml("<p style='color:red;'>---------------Reconnect!---------------</p>")self.text.append('')self.dataRecvws.abort()self.set_open()def get_popularity(self, popularity):print(u'人数',popularity)self.num.setText(str(popularity))def get_danmaku(self, content, user_name):self.text.insertHtml("<p style='color:black;'>"+'【'+user_name+'】:'+content+"</p>")self.text.append('')#self.text.append()#self.text.append(user_name + ':' + content)def handle_command(self, command):if isinstance(command, list):for one_command in command:self.handle_command(one_command)returncmd = command['cmd']# print(command)if cmd == 'DANMU_MSG':  # 收到弹幕self.get_danmaku(command['info'][1], command['info'][2][1])elif cmd == 'SEND_GIFT':  # 送礼物passelif cmd == 'WELCOME':  # 欢迎passelif cmd == 'WELCOME_GUARD':  # 欢迎房管passelif cmd == 'SYS_MSG':  # 系统消息passelif cmd == 'PREPARING':  # 房主准备中passelif cmd == 'LIVE':  # 直播开始passelif cmd == 'WISH_BOTTLE':  # 许愿瓶？passelse:print('未知命令：', command, file=sys.stderr)def make_packet(self, data, operation):body = json.dumps(data).encode('utf-8')header = HEADER_STRUCT.pack(HEADER_STRUCT.size + len(body),HEADER_STRUCT.size,1,operation,1)return header + body#request获得房间号def get_room_id(self, short_id=6782619):params = {'id': short_id}s = requests.Session()r = s.get(ROOM_INIT_URL, params=params)# print('r_status_code', r.status_code)data = r.json()# print("data['code']", data['code'])if data['code'] == 0:room_id = data['data']['room_id']print('room_id', room_id)return room_idif __name__ == '__main__':app = QApplication(sys.argv)ar = win()ar.show()sys.exit(app.exec_())

然后是分线程device.py

#device.py
# -*- coding:utf-8 -*-from PyQt5.QtCore import QThreadclass heart_beat(QThread):def __init__(self):super(heart_beat, self).__init__(parent=None)self.dataRecvws = Noneself.heart = Noneself.stop = 0def run(self):print('start thread!')while self.dataRecvws and self.heart:self.dataRecvws.sendBinaryMessage(self.heart)time.sleep(30)def set_heart(self,rev,heart):self.dataRecvws = revself.heart = heart

效果大概是这样的，输入房间短号后按下确定按钮开始连接。

本文标签： pyqt5获取B站的直播弹幕