标签：websocket name data PyQt json jsonpath msg 弹幕

前言

从B站上获取直播弹幕的方式大体有两种，一种是通过调用下面这个接口，通过轮询获取

import requests
room_id = 123456 # 示例
url = 'https://api.live.bilibili.com/xlive/web-room/v1/dM/gethistory'
headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36 Edg/115.0.1901.200',
        'Host': 'api.live.bilibili.com'}

data = {'roomid': room_id,
        'csrf_token': '',
        'csrf': '',
        'visit_id': '',
        }
r = requests.get(url=url, headers=headers, data=data)

但是这种方法需要不断地向服务器发送请求，同时，获取到的弹幕数据不是即时的。此外，还有一个重大的缺点是每次获取到的数据需要做去重处理。当然，你可以简单
地通过python中的集合，进行去重，但是需要及时清理旧数据，防止内存泄漏。

所以就有了第二种方法，websocket。

对B站的直播进行简单的逆向，可以知道，B站直播间的弹幕推送正是基于websocket协议推送的。接下来只要找到B站的websocket服务器地址，数据包的格式，就可
以实时获取到B站的数据了。

WebSocket 爬取弹幕数据

获取wss服务器地址和token

首先要先获取真实房间号，可通过下面这个接口，用get方法获取，请求时要带上从URL中提取的房间号参数。

https://api.live.bilibili.com/room/v1/Room/room_init

获取wss服务器地址比较简单，只需通过对下面这个接口发送post请求，并加上请求头，就可以获取到对应的wss服务器地址。

https://api.live.bilibili.com/xlive/web-room/v1/index/getDanmuInfo

发送的数据格式为json，real_room_id为真实的房间号。

# post发送的json示例
real_room_id = 123456 # 示例
params = {"id": real_room_id,
          "type": 0}  # real_room_id 为int类型

这个接口返回的参数也是个json，可以通过jsonpath库快速获取到wss服务器地址和鉴权用的token, 这里用的服务器地址是

wss://broadcastlv.chat.bilibili.com/sub

发送鉴权包和心跳包

连接服务器时，要在短时间内向wss服务器发送一个鉴权包，得到服务器的认可后，每隔30秒向服务器发送一次心跳包，保持长连。

参考B站直播信息流，和逆向得到的信息。可以
知道B站鉴权包和心跳包的请求头是经过特殊封装的，以请求头中的Type（消息类型）进行判断操作类型。

请求头格式

name	value	type	mean
PackLength	数据包大小	uint32	封包总大小
HeaderLength	16	uint16	请求头大小（默认为16）
Protocol	1	uint16	协议类型，1为心跳及认证包不使用压缩
Type	7	uint32	封包类型，7为鉴权包，2为心跳包
Sequence	1	uint32	保留字段，可忽略，默认为1

这里可以用python的struct模块封装请求头，最后根据鉴权包和心跳包的json，封装成数据包发送给服务器，就能够持续地获取到弹幕数据了。值得注意的是，自2023
年9月开始，B站服务器升级，未登录的用户获取到的直播弹幕数据是脱敏的。也就是说获取到的用户名是打上*号的。解决方法是，发送鉴权包时，加入"buvid"参数和用
户UID。这里的"buvid"指的是用户cookie里的buvid3参数。不过，经过测试，12月10号后此方法好像行不通了。

处理弹幕数据

连接到服务器后，接收到的数据要先根据响应头的Type（封包类型）判断数据类型，再进行处理。一般的弹幕数据的Type为5，在此之下又有很多种消息类型。例如弹幕
消息、付费留言、礼物信息等。这些消息的区分需要依靠响应体（json）里的cmd参数进行判断，在用jsonpath提取。这里为了和前端的qt窗口联系，同时，为了更好
地格式化输出弹幕信息，我直接将获取到的json字符串当做信号发送。

def get_dm_msg(self, data):
    # 获取数据包长度，协议类型和操作类型
    packet_len = int(data[:4].hex(), 16) # 数据包长度
    proto = int(data[6:8].hex(), 16) # 协议
    op = int(data[8:12].hex(), 16) # 操作类型

    # 若数据包是连着的，则根据第一个数据包的长度进行切分
    if len(data) > packet_len:
        self.get_dm_msg(data[packet_len:])
        data = data[:packet_len]

    # 判断协议类型，若为2则用zlib解压
    if proto == 2:
        data = zlib.decompress(data[16:])
        self.get_dm_msg(data)
        return
    if op == 3:
        my_logger.info("HeartBeat")
    # 判断消息类型
    if op == 5:
        try:
            # 解析json
            content = json.loads(data[16:].decode())
            # 发送数据
            global_signal.MsgData.emit(content) # 发送信号
        except Exception as e:
            my_logger.error(f"[GETDATA ERROR]: {e}")

处理弹幕数据方法样例

def __out_put(self, data):
    try:
        # 展示显示窗口
        self.show_w.show()
        # 获取消息类型
        cmd = jsonpath.jsonpath(data, "$.cmd")[0]

        # 弹幕消息
        if cmd == "DANMU_MSG":
            # 用户名
            user_name = jsonpath.jsonpath(data, "$.info[0]..user.base.name")[0]
            # 弹幕信息
            dm_msg = jsonpath.jsonpath(data, "$.info[1]")[0]
            if not self.__is_ban_msg(dm_msg):
                self.msg_show.append("<p><font color='white'>%s</font><font color='white'>: %s</font></p>" %
                                     (user_name, dm_msg))

        # 付费留言
        elif cmd == "SUPER_CHAT_MESSAGE":
            # 用户名
            user_name = jsonpath.jsonpath(data, "$.data.user_info.uname")[0]
            # 付费留言
            sc_msg = jsonpath.jsonpath(data, "$.data.message")[0]
            self.msg_show.append("<h1><font color='white'>[SC]%s</font><font color='white'>: %s</font></h1>" %
                                 (user_name, sc_msg))

        # 上舰通知
        elif cmd == "GUARD_BUY":
            # 用户名
            user_name = jsonpath.jsonpath(data, "$.data.username")[0]
            # 礼物名字
            gift_name = jsonpath.jsonpath(data, "$.data.gift_name")[0]
            self.msg_show.append("<font color='white'>[GUARD]%s</font><font color='white'> 购买了%s</font>" %
                                 (user_name, gift_name))

PyQt开发GUI

界面我选择使用PyQt开发。首先先对业务功能进行整理，再根据其对界面进行划分。因此分成了两个界面，一个设置窗口，负责连接到直播间，同时可以修改弹幕样式
或者增加屏蔽词。另一个是展示窗口，我选择使用QTextBrowser对弹幕信息进行展示，一是QTextBrowser相当于游览器里的网页，支持插入html，另外就是修改窗口
透明度和字体大小时比较简单，只需通过下面这一行代码就可以简单地修改展示窗口:

# 为控件设置半透明效果和默认字体大小
self.showMsg.setStyleSheet(f"background-color: rgba(128, 128, 128, {opacity});font-size: {font_size}px;")

设置窗口和展示窗口的设计

设置窗口
)
展示窗口
)

这里使用Qtdesinger可以很方便地制作出一个心仪的GUI，实现的效果如上图所示。

设置全局信号

信号与槽是PyQt开发的核心。为了方便个文件的调用，我将常用的信号放在staticData.py中，方便各窗体的调用和联系。

# 自定义信号
class Signals(QObject):
    # 操作类型信号
    Operation = pyqtSignal(str)
    # 弹幕数据信号
    MsgData = pyqtSignal(dict)

配置文件config

为了方便，且实时修改展示窗口的消息样式，我引入了配置文件config.json。类似与中间件的功能，后端通过WebSocket获取到原始的数据后，主程序要先读取
config.json文件里的内容，对数据进行处理，并格式化输出。另外，当设置窗口里的按钮被点击时，会先主程序发送一个字符串类型的信号（用于标识操作类型），
主程序会根据信号类型，对config.json文件进行修改。而每次弹幕消息要展示在窗口时，要先根据config.json文件里的内容进行修改。这样就实现了对弹幕消息的实时修改。

默认的config.json文件格式:

{
  "icon_path": "ui/television.ico",
  "tray_icon_path": "ui/tray_icon.png",
  "opacity": 0.2,
  "font_size": 20,
  "ban_words": [],
  "BiliSocket": {
    "headers": {
      "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0",
      "Origin": "https://live.bilibili.com",
      "Accept": "application/json",
      "Accept-Language": "zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2",
      "Content-Type": "application/json"
    },
    "cookies": "",
    "wss_addr": "wss://broadcastlv.chat.bilibili.com/sub",
    "UID": 0
  }
}

结语

本文所述的内容仅为简单地概括设计此项目的思路。要了解具体的实现方法，可以移步到我的仓库BiliDmCat，查看源代码

标签：websocket,name,data,PyQt,json,jsonpath,msg,弹幕
From： https://www.cnblogs.com/thetheOrange/p/17899681.html

基于PyQt和websocket，制作一个简单的BiliBili弹幕机（大体思路）

前言