项目概述

这是一个 B 站直播间弹幕联动系统，支持：

  • 实时监听直播间弹幕，识别指令关键词
  • 点歌功能（集成网易云音乐 pyncm）
  • 向指定用户批量发送私信
  • PySide6 GUI 界面

私信发送接口逆向

B 站私信发送使用 api.vc.bilibili.com（独立于主 API 域名）：

import uuid, json, time, requests

def send_message(cookie: str, receiver_uid: int, content: str) -> bool:
    bili_jct = extract_cookie(cookie, 'bili_jct')  # CSRF token
    sender_uid = get_self_uid(cookie)
    dev_id = str(uuid.uuid4()).upper()

    payload = {
        'msg[sender_uid]': sender_uid,
        'msg[receiver_id]': receiver_uid,
        'msg[receiver_type]': 1,   # 1=用户 2=粉丝团
        'msg[msg_type]': 1,        # 1=文字 2=图片
        'msg[dev_id]': dev_id,
        'msg[timestamp]': int(time.time()),
        'msg[content]': json.dumps({'content': content}),
        'csrf_token': bili_jct,
        'csrf': bili_jct,
        'build': 0,
        'mobi_app': 'web',
    }
    r = requests.post(
        'https://api.vc.bilibili.com/web_im/v1/web_im/send_msg',
        data=payload,
        headers={'Cookie': cookie, 'Referer': 'https://message.bilibili.com/'}
    )
    return r.json()['code'] == 0

弹幕监听

B 站直播弹幕使用 WebSocket 协议，连接 broadcastlv.chat.bilibili.com：

import asyncio, websockets, struct, json, zlib

DANMU_WS = 'wss://broadcastlv.chat.bilibili.com/sub'

async def listen_danmu(room_id: int, callback):
    async with websockets.connect(DANMU_WS) as ws:
        # 发送进入房间包
        await ws.send(build_join_packet(room_id))

        async for msg in ws:
            packets = parse_packet(msg)
            for pkt in packets:
                if pkt['op'] == 5:  # SEND_MSG_REPLY
                    data = json.loads(pkt['body'])
                    if data.get('cmd') == 'DANMU_MSG':
                        text = data['info'][1]
                        uid = data['info'][2][0]
                        await callback(uid, text)

弹幕数据包格式

Header (16 bytes):
  [0:4]  = 数据包总长度 (big-endian uint32)
  [4:6]  = 头部长度，固定 16
  [6:8]  = 协议版本 (0=JSON, 2=zlib压缩)
  [8:12] = 操作码 (2=心跳, 3=心跳回应, 5=通知, 7=进房, 8=进房回应)
  [12:16] = 序列号
Body: JSON 或 zlib(JSON)

点歌功能（pyncm 集成）

import pyncm
from pyncm import apis as ncm_apis

def search_and_play(keyword: str):
    # 搜索歌曲
    result = ncm_apis.cloudsearch.GetSearchResult(keyword, limit=1)
    songs = result['result']['songs']
    if not songs:
        return None
    song = songs[0]
    song_id = song['id']

    # 获取下载链接
    url_result = ncm_apis.track.GetTrackAudio([song_id], bitrate=320000)
    url = url_result['data'][0]['url']

    # pygame 播放
    pygame.mixer.music.load(url)
    pygame.mixer.music.play()
    return song['name']

弹幕规则引擎

通过正则规则匹配弹幕指令，触发对应动作：

[
  {"pattern": "点歌 (.+)",    "action": "song",    "group": 1},
  {"pattern": "私信我$",       "action": "sendmsg", "template": "您好！欢迎关注！"},
  {"pattern": "跳过$",         "action": "skip"}
]

规则配置存入 danmu_rules.json，运行时热重载，无需重启服务。