現在、Bilibili（B站）のライブ配信WebSocketから情報を取得するためのオープンソースプロジェクトは多数存在します。

しかし、実際に導入してみると、独自の要件に完全に適合しないケースが少なくありません。

そこで、自社のビジネス要件に最適化できるよう、独自の接続システムを構築することにしました。

本記事では、実装過程で得られた重要なポイントと注意事項をまとめました。

PHP用ライブラリ：composerでインストール可能、Bilibiliライブルームへの接続とデータ復号化に対応

自動応答ボット：ギフト感謝、定時広告、フォロー通知、自動返信機能を搭載、Dockerでのデプロイに対応

事前準備

リアルルームIDの取得

Web版のライブルームURLに含まれる部屋番号は短縮IDの可能性があり、必ずしも真のIDではありません。そのため、APIを呼び出して正確な部屋番号を確認することを推奨します。

リクエスト方式：GET

エンドポイント：https://api.live.bilibili.com/room/v1/Room/get_info

レスポンス例

詳細を表示

curl -G 'https://api.live.bilibili.com/room/v1/Room/get_info' \
--data-urlencode 'room_id=27668995'

{
    "code": 0,
    "msg": "ok",
    "data": {
        "uid": 3493124609411229,
        "room_id": 27668995,
        "short_id": 0,
        "attention": 13353,
        "online": 4173,
        "live_status": 1,
        "title": "不给糖就捣蛋",
        "area_name": "虚拟日常",
        "parent_area_name": "虚拟主播"
    }
}

認証トークンの取得

ライブルームの情報ストリーム接続に必要なサーバーアドレスと認証トークンを取得します。

重要： Bilibiliはプライバシーポリシーを更新しました。未ログインユーザーからの接続は約5分後に制限され、ユーザー名が「*」でマスクされ、ユーザーIDが0として表示されます。そのため、本APIを呼び出す際はcookieを送信する必要があります。

注意： WebSocket接続URLにはパス**/sub**が必要です。例：wss://tx-sh-live-comet-08.chat.bilibili.com:443/sub

リクエスト方式：GET

エンドポイント：https://api.live.bilibili.com/xlive/web-room/v1/index/getDanmuInfo

レスポンス例

詳細を表示

curl -G 'https://api.live.bilibili.com/xlive/web-room/v1/index/getDanmuInfo' \
--data-urlencode 'id=30118851'

{
    "code": 0,
    "message": "0",
    "data": {
        "token": "TrF6FaSlmxVBM4eBYGoaWPuZ-xVL-bhK80waLbGRfpj...",
        "host_list": [
            {
                "host": "tx-sh-live-comet-08.chat.bilibili.com",
                "port": 2243,
                "wss_port": 443,
                "ws_port": 2244
            },
            {
                "host": "tx-bj-live-comet-08.chat.bilibili.com",
                "port": 2243,
                "wss_port": 443,
                "ws_port": 2244
            }
        ]
    }
}

接続プロトコル

データパケットはWebSocketまたはTCP接続上で送信され、ヘッダー + ボディの形式をとります。

処理フロー：

サーバー接続 → 認証パケット送信 → 認証応答受信 → データ受信＆ハートビート送信（30秒間隔）

プロトコル形式：全フィールドはビッグエンディアン

• Packet Length：パケット全体の長さ（ヘッダー含む）
• Header Length：ヘッダー長（固定16バイト）
• Version：プロトコルバージョン
• Operation：操作コード
• Sequence ID：予約フィールド
• Body：メッセージ本体

Version 仕様： 0 - 非圧縮パケット 1 - ハートビート・認証パケット 2 - zlib圧縮 3 - brotli圧縮

Operation 仕様： 2 - クライアントハートビート送信 3 - サーバーハートビート応答 5 - サーバープッシュメッセージ 7 - クライアント認証パケット 8 - サーバー認証応答

注意：Bilibiliはセキュリティ強化を実施しています。接続時にはユーザーのcookieを必ず携帯してください。

認証パケットの構築

注意：認証パケットは接続成功後5秒以内に送信する必要があります。超過すると切断されます。

ボディのJSON構造：

protover 仕様： 2 - zlib圧縮 3 - brotli圧縮

パケット構造例：

00000000: 0000 0152 0010 0001 0000 0007 0000 0001
00000001: 7b22 7569 6422 3a34 3332 3530 3531 2c22  {"uid":4325051,"
00000002: 726f 6f6d 6964 223a 3331 3432 3735 3432  roomid":31427542
...

Pythonによる実装例：

import struct
import json
import zlib

def build_auth_packet(room_id: int, token: str, uid: int = 0, buvid: str = ""):
    """認証パケットを構築"""
    auth_body = {
        "uid": uid,
        "roomid": room_id,
        "protover": 2,
        "buvid": buvid,
        "platform": "web",
        "type": 2,
        "key": token
    }
    
    body_bytes = json.dumps(auth_body).encode('utf-8')
    header_length = 16
    packet_length = header_length + len(body_bytes)
    
    # ヘッダー構築（ビッグエンディアン）
    header = struct.pack(
        '>IHHII',
        packet_length,    # Packet Length
        header_length,    # Header Length
        1,                # Version
        7,                # Operation (Auth)
        1                 # Sequence ID
    )
    
    return header + body_bytes


def build_heartbeat_packet():
    """ハートビートパケットを構築"""
    body = b'[object Object]'
    header = struct.pack(
        '>IHHII',
        16 + len(body),
        16,
        1,
        2,
        1
    )
    return header + body


def parse_packet(data: bytes):
    """パケットを解析"""
    if len(data) < 16:
        return None
    
    header = struct.unpack('>IHHII', data[:16])
    packet_len = header[0]
    header_len = header[1]
    version = header[2]
    operation = header[3]
    
    body = data[header_len:packet_len]
    
    # 圧縮解除
    if version == 2:
        body = zlib.decompress(body)
    elif version == 3:
        import brotli
        body = brotli.decompress(body)
    
    return {
        "version": version,
        "operation": operation,
        "body": body
    }