fork xfgryujk/blivedm

blivedm/__init__.py

@@ -0,0 +1,3 @@
+# -*- coding: utf-8 -*-
+from .handlers import *
+from .clients import *

blivedm/clients/__init__.py

@@ -0,0 +1,3 @@
+# -*- coding: utf-8 -*-
+from .web import *
+from .open_live import *

blivedm/clients/open_live.py

@@ -0,0 +1,293 @@
+# -*- coding: utf-8 -*-
+import asyncio
+import datetime
+import hashlib
+import hmac
+import json
+import logging
+import uuid
+from typing import *
+
+import aiohttp
+
+from . import ws_base
+
+__all__ = (
+    'OpenLiveClient',
+)
+
+logger = logging.getLogger('blivedm')
+
+START_URL = 'https://live-open.biliapi.com/v2/app/start'
+HEARTBEAT_URL = 'https://live-open.biliapi.com/v2/app/heartbeat'
+END_URL = 'https://live-open.biliapi.com/v2/app/end'
+
+
+class OpenLiveClient(ws_base.WebSocketClientBase):
+    """
+    开放平台客户端
+
+    文档参考：https://open-live.bilibili.com/document/
+
+    :param access_key_id: 在开放平台申请的access_key_id
+    :param access_key_secret: 在开放平台申请的access_key_secret
+    :param app_id: 在开放平台创建的项目ID
+    :param room_owner_auth_code: 主播身份码
+    :param session: cookie、连接池
+    :param heartbeat_interval: 发送连接心跳包的间隔时间（秒）
+    :param game_heartbeat_interval: 发送项目心跳包的间隔时间（秒）
+    """
+
+    def __init__(
+        self,
+        access_key_id: str,
+        access_key_secret: str,
+        app_id: int,
+        room_owner_auth_code: str,
+        *,
+        session: Optional[aiohttp.ClientSession] = None,
+        heartbeat_interval=30,
+        game_heartbeat_interval=20,
+    ):
+        super().__init__(session, heartbeat_interval)
+
+        self._access_key_id = access_key_id
+        self._access_key_secret = access_key_secret
+        self._app_id = app_id
+        self._room_owner_auth_code = room_owner_auth_code
+        self._game_heartbeat_interval = game_heartbeat_interval
+
+        # 在调用init_room后初始化的字段
+        self._room_owner_uid: Optional[int] = None
+        """主播用户ID"""
+        self._room_owner_open_id: Optional[str] = None
+        """主播Open ID"""
+        self._host_server_url_list: Optional[List[str]] = []
+        """弹幕服务器URL列表"""
+        self._auth_body: Optional[str] = None
+        """连接弹幕服务器用的认证包内容"""
+        self._game_id: Optional[str] = None
+        """项目场次ID"""
+
+        # 在运行时初始化的字段
+        self._game_heartbeat_timer_handle: Optional[asyncio.TimerHandle] = None
+        """发项目心跳包定时器的handle"""
+
+    @property
+    def room_owner_uid(self) -> Optional[int]:
+        """
+        主播用户ID，调用init_room后初始化
+        """
+        return self._room_owner_uid
+
+    @property
+    def room_owner_open_id(self) -> Optional[str]:
+        """
+        主播Open ID，调用init_room后初始化
+        """
+        return self._room_owner_open_id
+
+    @property
+    def room_owner_auth_code(self):
+        """
+        主播身份码
+        """
+        return self._room_owner_auth_code
+
+    @property
+    def app_id(self):
+        """
+        在开放平台创建的项目ID
+        """
+        return self._app_id
+
+    @property
+    def game_id(self) -> Optional[str]:
+        """
+        项目场次ID，调用init_room后初始化
+        """
+        return self._game_id
+
+    async def close(self):
+        """
+        释放本客户端的资源，调用后本客户端将不可用
+        """
+        if self.is_running:
+            logger.warning('room=%s is calling close(), but client is running', self.room_id)
+
+        if self._game_heartbeat_timer_handle is not None:
+            self._game_heartbeat_timer_handle.cancel()
+            self._game_heartbeat_timer_handle = None
+        await self._end_game()
+
+        await super().close()
+
+    def _request_open_live(self, url, body: dict):
+        body_bytes = json.dumps(body).encode('utf-8')
+        headers = {
+            'x-bili-accesskeyid': self._access_key_id,
+            'x-bili-content-md5': hashlib.md5(body_bytes).hexdigest(),
+            'x-bili-signature-method': 'HMAC-SHA256',
+            'x-bili-signature-nonce': uuid.uuid4().hex,
+            'x-bili-signature-version': '1.0',
+            'x-bili-timestamp': str(int(datetime.datetime.now().timestamp())),
+        }
+
+        str_to_sign = '\n'.join(
+            f'{key}:{value}'
+            for key, value in headers.items()
+        )
+        signature = hmac.new(
+            self._access_key_secret.encode('utf-8'), str_to_sign.encode('utf-8'), hashlib.sha256
+        ).hexdigest()
+        headers['Authorization'] = signature
+
+        headers['Content-Type'] = 'application/json'
+        headers['Accept'] = 'application/json'
+        return self._session.post(url, headers=headers, data=body_bytes)
+
+    async def init_room(self):
+        """
+        开启项目，并初始化连接房间需要的字段
+
+        :return: 是否成功
+        """
+        if not await self._start_game():
+            return False
+
+        if self._game_id != '' and self._game_heartbeat_timer_handle is None:
+            self._game_heartbeat_timer_handle = asyncio.get_running_loop().call_later(
+                self._game_heartbeat_interval, self._on_send_game_heartbeat
+            )
+        return True
+
+    async def _start_game(self):
+        try:
+            async with self._request_open_live(
+                START_URL,
+                {'code': self._room_owner_auth_code, 'app_id': self._app_id}
+            ) as res:
+                if res.status != 200:
+                    logger.warning('_start_game() failed, status=%d, reason=%s', res.status, res.reason)
+                    return False
+                data = await res.json()
+                if data['code'] != 0:
+                    logger.warning('_start_game() failed, code=%d, message=%s, request_id=%s',
+                                   data['code'], data['message'], data['request_id'])
+                    return False
+                if not self._parse_start_game(data['data']):
+                    return False
+        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+            logger.exception('_start_game() failed:')
+            return False
+        return True
+
+    def _parse_start_game(self, data):
+        self._game_id = data['game_info']['game_id']
+        websocket_info = data['websocket_info']
+        self._auth_body = websocket_info['auth_body']
+        self._host_server_url_list = websocket_info['wss_link']
+        anchor_info = data['anchor_info']
+        self._room_id = anchor_info['room_id']
+        self._room_owner_uid = anchor_info['uid']
+        self._room_owner_open_id = anchor_info['open_id']
+        return True
+
+    async def _end_game(self):
+        """
+        关闭项目。建议关闭客户端时保证调用到这个函数（close会调用），否则可能短时间内无法重复连接同一个房间
+        """
+        if self._game_id in (None, ''):
+            return True
+
+        try:
+            async with self._request_open_live(
+                END_URL,
+                {'app_id': self._app_id, 'game_id': self._game_id}
+            ) as res:
+                if res.status != 200:
+                    logger.warning('room=%d _end_game() failed, status=%d, reason=%s',
+                                   self._room_id, res.status, res.reason)
+                    return False
+                data = await res.json()
+                code = data['code']
+                if code != 0:
+                    if code in (7000, 7003):
+                        # 项目已经关闭了也算成功
+                        return True
+
+                    logger.warning('room=%d _end_game() failed, code=%d, message=%s, request_id=%s',
+                                   self._room_id, code, data['message'], data['request_id'])
+                    return False
+        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+            logger.exception('room=%d _end_game() failed:', self._room_id)
+            return False
+        return True
+
+    def _on_send_game_heartbeat(self):
+        """
+        定时发送项目心跳包的回调
+        """
+        self._game_heartbeat_timer_handle = asyncio.get_running_loop().call_later(
+            self._game_heartbeat_interval, self._on_send_game_heartbeat
+        )
+        asyncio.create_task(self._send_game_heartbeat())
+
+    async def _send_game_heartbeat(self):
+        """
+        发送项目心跳包
+        """
+        if self._game_id in (None, ''):
+            logger.warning('game=%d _send_game_heartbeat() failed, game_id not found', self._game_id)
+            return False
+
+        try:
+            # 保存一下，防止await之后game_id改变
+            game_id = self._game_id
+            async with self._request_open_live(
+                HEARTBEAT_URL,
+                {'game_id': game_id}
+            ) as res:
+                if res.status != 200:
+                    logger.warning('room=%d _send_game_heartbeat() failed, status=%d, reason=%s',
+                                   self._room_id, res.status, res.reason)
+                    return False
+                data = await res.json()
+                code = data['code']
+                if code != 0:
+                    logger.warning('room=%d _send_game_heartbeat() failed, code=%d, message=%s, request_id=%s',
+                                   self._room_id, code, data['message'], data['request_id'])
+
+                    if code == 7003 and self._game_id == game_id:
+                        # 项目异常关闭，可能是心跳超时，需要重新开启项目
+                        self._need_init_room = True
+                        if self._websocket is not None and not self._websocket.closed:
+                            await self._websocket.close()
+
+                    return False
+        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+            logger.exception('room=%d _send_game_heartbeat() failed:', self._room_id)
+            return False
+        return True
+
+    async def _on_before_ws_connect(self, retry_count):
+        """
+        在每次建立连接之前调用，可以用来初始化房间
+        """
+        # 重连次数太多则重新init_room，保险
+        reinit_period = max(3, len(self._host_server_url_list or ()))
+        if retry_count > 0 and retry_count % reinit_period == 0:
+            self._need_init_room = True
+        await super()._on_before_ws_connect(retry_count)
+
+    def _get_ws_url(self, retry_count) -> str:
+        """
+        返回WebSocket连接的URL，可以在这里做故障转移和负载均衡
+        """
+        return self._host_server_url_list[retry_count % len(self._host_server_url_list)]
+
+    async def _send_auth(self):
+        """
+        发送认证包
+        """
+        await self._websocket.send_bytes(self._make_packet(self._auth_body, ws_base.Operation.AUTH))

blivedm/clients/web.py

@@ -0,0 +1,266 @@
+# -*- coding: utf-8 -*-
+import asyncio
+import logging
+from typing import *
+
+import aiohttp
+import yarl
+
+from . import ws_base
+from .. import utils
+
+__all__ = (
+    'BLiveClient',
+)
+
+logger = logging.getLogger('blivedm')
+
+UID_INIT_URL = 'https://api.bilibili.com/x/web-interface/nav'
+BUVID_INIT_URL = 'https://data.bilibili.com/v/'
+ROOM_INIT_URL = 'https://api.live.bilibili.com/xlive/web-room/v1/index/getInfoByRoom'
+DANMAKU_SERVER_CONF_URL = 'https://api.live.bilibili.com/xlive/web-room/v1/index/getDanmuInfo'
+DEFAULT_DANMAKU_SERVER_LIST = [
+    {'host': 'broadcastlv.chat.bilibili.com', 'port': 2243, 'wss_port': 443, 'ws_port': 2244}
+]
+
+
+class BLiveClient(ws_base.WebSocketClientBase):
+    """
+    web端客户端
+
+    :param room_id: URL中的房间ID，可以用短ID
+    :param uid: B站用户ID，0表示未登录，None表示自动获取
+    :param session: cookie、连接池
+    :param heartbeat_interval: 发送心跳包的间隔时间（秒）
+    """
+
+    def __init__(
+        self,
+        room_id: int,
+        *,
+        uid: Optional[int] = None,
+        session: Optional[aiohttp.ClientSession] = None,
+        heartbeat_interval=30,
+    ):
+        super().__init__(session, heartbeat_interval)
+
+        self._tmp_room_id = room_id
+        """用来init_room的临时房间ID，可以用短ID"""
+        self._uid = uid
+
+        # 在调用init_room后初始化的字段
+        self._room_owner_uid: Optional[int] = None
+        """主播用户ID"""
+        self._host_server_list: Optional[List[dict]] = None
+        """
+        弹幕服务器列表
+
+        `[{host: "tx-bj4-live-comet-04.chat.bilibili.com", port: 2243, wss_port: 443, ws_port: 2244}, ...]`
+        """
+        self._host_server_token: Optional[str] = None
+        """连接弹幕服务器用的token"""
+
+    @property
+    def tmp_room_id(self) -> int:
+        """
+        构造时传进来的room_id参数
+        """
+        return self._tmp_room_id
+
+    @property
+    def room_owner_uid(self) -> Optional[int]:
+        """
+        主播用户ID，调用init_room后初始化
+        """
+        return self._room_owner_uid
+
+    @property
+    def uid(self) -> Optional[int]:
+        """
+        当前登录的用户ID，未登录则为0，调用init_room后初始化
+        """
+        return self._uid
+
+    async def init_room(self):
+        """
+        初始化连接房间需要的字段
+
+        :return: True代表没有降级，如果需要降级后还可用，重载这个函数返回True
+        """
+        if self._uid is None:
+            if not await self._init_uid():
+                logger.warning('room=%d _init_uid() failed', self._tmp_room_id)
+                self._uid = 0
+
+        if self._get_buvid() == '':
+            if not await self._init_buvid():
+                logger.warning('room=%d _init_buvid() failed', self._tmp_room_id)
+
+        res = True
+        if not await self._init_room_id_and_owner():
+            res = False
+            # 失败了则降级
+            self._room_id = self._tmp_room_id
+            self._room_owner_uid = 0
+
+        if not await self._init_host_server():
+            res = False
+            # 失败了则降级
+            self._host_server_list = DEFAULT_DANMAKU_SERVER_LIST
+            self._host_server_token = None
+        return res
+
+    async def _init_uid(self):
+        cookies = self._session.cookie_jar.filter_cookies(yarl.URL(UID_INIT_URL))
+        sessdata_cookie = cookies.get('SESSDATA', None)
+        if sessdata_cookie is None or sessdata_cookie.value == '':
+            # cookie都没有，不用请求了
+            self._uid = 0
+            return True
+
+        try:
+            async with self._session.get(
+                UID_INIT_URL,
+                headers={'User-Agent': utils.USER_AGENT},
+            ) as res:
+                if res.status != 200:
+                    logger.warning('room=%d _init_uid() failed, status=%d, reason=%s', self._tmp_room_id,
+                                   res.status, res.reason)
+                    return False
+                data = await res.json()
+                if data['code'] != 0:
+                    if data['code'] == -101:
+                        # 未登录
+                        self._uid = 0
+                        return True
+                    logger.warning('room=%d _init_uid() failed, message=%s', self._tmp_room_id,
+                                   data['message'])
+                    return False
+
+                data = data['data']
+                if not data['isLogin']:
+                    # 未登录
+                    self._uid = 0
+                else:
+                    self._uid = data['mid']
+                return True
+        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+            logger.exception('room=%d _init_uid() failed:', self._tmp_room_id)
+            return False
+
+    def _get_buvid(self):
+        cookies = self._session.cookie_jar.filter_cookies(yarl.URL(BUVID_INIT_URL))
+        buvid_cookie = cookies.get('buvid3', None)
+        if buvid_cookie is None:
+            return ''
+        return buvid_cookie.value
+
+    async def _init_buvid(self):
+        try:
+            async with self._session.get(
+                BUVID_INIT_URL,
+                headers={'User-Agent': utils.USER_AGENT},
+            ) as res:
+                if res.status != 200:
+                    logger.warning('room=%d _init_buvid() status error, status=%d, reason=%s',
+                                   self._tmp_room_id, res.status, res.reason)
+        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+            logger.exception('room=%d _init_buvid() exception:', self._tmp_room_id)
+        return self._get_buvid() != ''
+
+    async def _init_room_id_and_owner(self):
+        try:
+            async with self._session.get(
+                ROOM_INIT_URL,
+                headers={'User-Agent': utils.USER_AGENT},
+                params={
+                    'room_id': self._tmp_room_id
+                },
+            ) as res:
+                if res.status != 200:
+                    logger.warning('room=%d _init_room_id_and_owner() failed, status=%d, reason=%s', self._tmp_room_id,
+                                   res.status, res.reason)
+                    return False
+                data = await res.json()
+                if data['code'] != 0:
+                    logger.warning('room=%d _init_room_id_and_owner() failed, message=%s', self._tmp_room_id,
+                                   data['message'])
+                    return False
+                if not self._parse_room_init(data['data']):
+                    return False
+        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+            logger.exception('room=%d _init_room_id_and_owner() failed:', self._tmp_room_id)
+            return False
+        return True
+
+    def _parse_room_init(self, data):
+        room_info = data['room_info']
+        self._room_id = room_info['room_id']
+        self._room_owner_uid = room_info['uid']
+        return True
+
+    async def _init_host_server(self):
+        try:
+            async with self._session.get(
+                DANMAKU_SERVER_CONF_URL,
+                headers={'User-Agent': utils.USER_AGENT},
+                params={
+                    'id': self._room_id,
+                    'type': 0
+                },
+            ) as res:
+                if res.status != 200:
+                    logger.warning('room=%d _init_host_server() failed, status=%d, reason=%s', self._room_id,
+                                   res.status, res.reason)
+                    return False
+                data = await res.json()
+                if data['code'] != 0:
+                    logger.warning('room=%d _init_host_server() failed, message=%s', self._room_id, data['message'])
+                    return False
+                if not self._parse_danmaku_server_conf(data['data']):
+                    return False
+        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+            logger.exception('room=%d _init_host_server() failed:', self._room_id)
+            return False
+        return True
+
+    def _parse_danmaku_server_conf(self, data):
+        self._host_server_list = data['host_list']
+        self._host_server_token = data['token']
+        if not self._host_server_list:
+            logger.warning('room=%d _parse_danmaku_server_conf() failed: host_server_list is empty', self._room_id)
+            return False
+        return True
+
+    async def _on_before_ws_connect(self, retry_count):
+        """
+        在每次建立连接之前调用，可以用来初始化房间
+        """
+        # 重连次数太多则重新init_room，保险
+        reinit_period = max(3, len(self._host_server_list or ()))
+        if retry_count > 0 and retry_count % reinit_period == 0:
+            self._need_init_room = True
+        await super()._on_before_ws_connect(retry_count)
+
+    def _get_ws_url(self, retry_count) -> str:
+        """
+        返回WebSocket连接的URL，可以在这里做故障转移和负载均衡
+        """
+        host_server = self._host_server_list[retry_count % len(self._host_server_list)]
+        return f"wss://{host_server['host']}:{host_server['wss_port']}/sub"
+
+    async def _send_auth(self):
+        """
+        发送认证包
+        """
+        auth_params = {
+            'uid': self._uid,
+            'roomid': self._room_id,
+            'protover': 3,
+            'platform': 'web',
+            'type': 2,
+            'buvid': self._get_buvid(),
+        }
+        if self._host_server_token is not None:
+            auth_params['key'] = self._host_server_token
+        await self._websocket.send_bytes(self._make_packet(auth_params, ws_base.Operation.AUTH))

blivedm/clients/ws_base.py

@@ -0,0 +1,494 @@
+# -*- coding: utf-8 -*-
+import asyncio
+import enum
+import json
+import logging
+import struct
+import zlib
+from typing import *
+
+import aiohttp
+import brotli
+
+from .. import handlers, utils
+
+logger = logging.getLogger('blivedm')
+
+USER_AGENT = (
+    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/102.0.0.0 Safari/537.36'
+)
+
+HEADER_STRUCT = struct.Struct('>I2H2I')
+
+
+class HeaderTuple(NamedTuple):
+    pack_len: int
+    raw_header_size: int
+    ver: int
+    operation: int
+    seq_id: int
+
+
+# WS_BODY_PROTOCOL_VERSION
+class ProtoVer(enum.IntEnum):
+    NORMAL = 0
+    HEARTBEAT = 1
+    DEFLATE = 2
+    BROTLI = 3
+
+
+# go-common\app\service\main\broadcast\model\operation.go
+class Operation(enum.IntEnum):
+    HANDSHAKE = 0
+    HANDSHAKE_REPLY = 1
+    HEARTBEAT = 2
+    HEARTBEAT_REPLY = 3
+    SEND_MSG = 4
+    SEND_MSG_REPLY = 5
+    DISCONNECT_REPLY = 6
+    AUTH = 7
+    AUTH_REPLY = 8
+    RAW = 9
+    PROTO_READY = 10
+    PROTO_FINISH = 11
+    CHANGE_ROOM = 12
+    CHANGE_ROOM_REPLY = 13
+    REGISTER = 14
+    REGISTER_REPLY = 15
+    UNREGISTER = 16
+    UNREGISTER_REPLY = 17
+    # B站业务自定义OP
+    # MinBusinessOp = 1000
+    # MaxBusinessOp = 10000
+
+
+# WS_AUTH
+class AuthReplyCode(enum.IntEnum):
+    OK = 0
+    TOKEN_ERROR = -101
+
+
+class InitError(Exception):
+    """初始化失败"""
+
+
+class AuthError(Exception):
+    """认证失败"""
+
+
+DEFAULT_RECONNECT_POLICY = utils.make_constant_retry_policy(1)
+
+
+class WebSocketClientBase:
+    """
+    基于WebSocket的客户端
+
+    :param session: cookie、连接池
+    :param heartbeat_interval: 发送心跳包的间隔时间（秒）
+    """
+
+    def __init__(
+        self,
+        session: Optional[aiohttp.ClientSession] = None,
+        heartbeat_interval: float = 30,
+    ):
+        if session is None:
+            self._session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=10))
+            self._own_session = True
+        else:
+            self._session = session
+            self._own_session = False
+            assert self._session.loop is asyncio.get_event_loop()  # noqa
+
+        self._heartbeat_interval = heartbeat_interval
+
+        self._need_init_room = True
+        self._handler: Optional[handlers.HandlerInterface] = None
+        """消息处理器"""
+        self._get_reconnect_interval: Callable[[int, int], float] = DEFAULT_RECONNECT_POLICY
+        """重连间隔时间增长策略"""
+
+        # 在调用init_room后初始化的字段
+        self._room_id: Optional[int] = None
+
+        # 在运行时初始化的字段
+        self._websocket: Optional[aiohttp.ClientWebSocketResponse] = None
+        """WebSocket连接"""
+        self._network_future: Optional[asyncio.Future] = None
+        """网络协程的future"""
+        self._heartbeat_timer_handle: Optional[asyncio.TimerHandle] = None
+        """发心跳包定时器的handle"""
+
+    @property
+    def is_running(self) -> bool:
+        """
+        本客户端正在运行，注意调用stop后还没完全停止也算正在运行
+        """
+        return self._network_future is not None
+
+    @property
+    def room_id(self) -> Optional[int]:
+        """
+        房间ID，调用init_room后初始化
+        """
+        return self._room_id
+
+    def set_handler(self, handler: Optional['handlers.HandlerInterface']):
+        """
+        设置消息处理器
+
+        注意消息处理器和网络协程运行在同一个协程，如果处理消息耗时太长会阻塞接收消息。如果是CPU密集型的任务，建议将消息推到线程池处理；
+        如果是IO密集型的任务，应该使用async函数，并且在handler里使用create_task创建新的协程
+
+        :param handler: 消息处理器
+        """
+        self._handler = handler
+
+    def set_reconnect_policy(self, get_reconnect_interval: Callable[[int, int], float]):
+        """
+        设置重连间隔时间增长策略
+
+        :param get_reconnect_interval: 一个可调用对象，输入重试次数 (retry_count, total_retry_count)，返回间隔时间
+        """
+        self._get_reconnect_interval = get_reconnect_interval
+
+    def start(self):
+        """
+        启动本客户端
+        """
+        if self.is_running:
+            logger.warning('room=%s client is running, cannot start() again', self.room_id)
+            return
+
+        self._network_future = asyncio.create_task(self._network_coroutine_wrapper())
+
+    def stop(self):
+        """
+        停止本客户端
+        """
+        if not self.is_running:
+            logger.warning('room=%s client is stopped, cannot stop() again', self.room_id)
+            return
+
+        self._network_future.cancel()
+
+    async def stop_and_close(self):
+        """
+        便利函数，停止本客户端并释放本客户端的资源，调用后本客户端将不可用
+        """
+        if self.is_running:
+            self.stop()
+            await self.join()
+        await self.close()
+
+    async def join(self):
+        """
+        等待本客户端停止
+        """
+        if not self.is_running:
+            logger.warning('room=%s client is stopped, cannot join()', self.room_id)
+            return
+
+        await asyncio.shield(self._network_future)
+
+    async def close(self):
+        """
+        释放本客户端的资源，调用后本客户端将不可用
+        """
+        if self.is_running:
+            logger.warning('room=%s is calling close(), but client is running', self.room_id)
+
+        # 如果session是自己创建的则关闭session
+        if self._own_session:
+            await self._session.close()
+
+    async def init_room(self) -> bool:
+        """
+        初始化连接房间需要的字段
+
+        :return: True代表没有降级，如果需要降级后还可用，重载这个函数返回True
+        """
+        raise NotImplementedError
+
+    @staticmethod
+    def _make_packet(data: Union[dict, str, bytes], operation: int) -> bytes:
+        """
+        创建一个要发送给服务器的包
+
+        :param data: 包体JSON数据
+        :param operation: 操作码，见Operation
+        :return: 整个包的数据
+        """
+        if isinstance(data, dict):
+            body = json.dumps(data).encode('utf-8')
+        elif isinstance(data, str):
+            body = data.encode('utf-8')
+        else:
+            body = data
+        header = HEADER_STRUCT.pack(*HeaderTuple(
+            pack_len=HEADER_STRUCT.size + len(body),
+            raw_header_size=HEADER_STRUCT.size,
+            ver=1,
+            operation=operation,
+            seq_id=1
+        ))
+        return header + body
+
+    async def _network_coroutine_wrapper(self):
+        """
+        负责处理网络协程的异常，网络协程具体逻辑在_network_coroutine里
+        """
+        exc = None
+        try:
+            await self._network_coroutine()
+        except asyncio.CancelledError:
+            # 正常停止
+            pass
+        except Exception as e:
+            logger.exception('room=%s _network_coroutine() finished with exception:', self.room_id)
+            exc = e
+        finally:
+            logger.debug('room=%s _network_coroutine() finished', self.room_id)
+            self._network_future = None
+
+        if self._handler is not None:
+            self._handler.on_client_stopped(self, exc)
+
+    async def _network_coroutine(self):
+        """
+        网络协程，负责连接服务器、接收消息、解包
+        """
+        # retry_count在连接成功后会重置为0，total_retry_count不会
+        retry_count = 0
+        total_retry_count = 0
+        while True:
+            try:
+                await self._on_before_ws_connect(retry_count)
+
+                # 连接
+                async with self._session.ws_connect(
+                    self._get_ws_url(retry_count),
+                    headers={'User-Agent': utils.USER_AGENT},  # web端的token也会签名UA
+                    receive_timeout=self._heartbeat_interval + 5,
+                ) as websocket:
+                    self._websocket = websocket
+                    await self._on_ws_connect()
+
+                    # 处理消息
+                    message: aiohttp.WSMessage
+                    async for message in websocket:
+                        await self._on_ws_message(message)
+                        # 至少成功处理1条消息
+                        retry_count = 0
+
+            except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+                # 掉线重连
+                pass
+            except AuthError:
+                # 认证失败了，应该重新获取token再重连
+                logger.exception('room=%d auth failed, trying init_room() again', self.room_id)
+                self._need_init_room = True
+            finally:
+                self._websocket = None
+                await self._on_ws_close()
+
+            # 准备重连
+            retry_count += 1
+            total_retry_count += 1
+            logger.warning(
+                'room=%d is reconnecting, retry_count=%d, total_retry_count=%d',
+                self.room_id, retry_count, total_retry_count
+            )
+            await asyncio.sleep(self._get_reconnect_interval(retry_count, total_retry_count))
+
+    async def _on_before_ws_connect(self, retry_count):
+        """
+        在每次建立连接之前调用，可以用来初始化房间
+        """
+        if not self._need_init_room:
+            return
+
+        if not await self.init_room():
+            raise InitError('init_room() failed')
+        self._need_init_room = False
+
+    def _get_ws_url(self, retry_count) -> str:
+        """
+        返回WebSocket连接的URL，可以在这里做故障转移和负载均衡
+        """
+        raise NotImplementedError
+
+    async def _on_ws_connect(self):
+        """
+        WebSocket连接成功
+        """
+        await self._send_auth()
+        self._heartbeat_timer_handle = asyncio.get_running_loop().call_later(
+            self._heartbeat_interval, self._on_send_heartbeat
+        )
+
+    async def _on_ws_close(self):
+        """
+        WebSocket连接断开
+        """
+        if self._heartbeat_timer_handle is not None:
+            self._heartbeat_timer_handle.cancel()
+            self._heartbeat_timer_handle = None
+
+    async def _send_auth(self):
+        """
+        发送认证包
+        """
+        raise NotImplementedError
+
+    def _on_send_heartbeat(self):
+        """
+        定时发送心跳包的回调
+        """
+        if self._websocket is None or self._websocket.closed:
+            self._heartbeat_timer_handle = None
+            return
+
+        self._heartbeat_timer_handle = asyncio.get_running_loop().call_later(
+            self._heartbeat_interval, self._on_send_heartbeat
+        )
+        asyncio.create_task(self._send_heartbeat())
+
+    async def _send_heartbeat(self):
+        """
+        发送心跳包
+        """
+        if self._websocket is None or self._websocket.closed:
+            return
+
+        try:
+            await self._websocket.send_bytes(self._make_packet({}, Operation.HEARTBEAT))
+        except (ConnectionResetError, aiohttp.ClientConnectionError) as e:
+            logger.warning('room=%d _send_heartbeat() failed: %r', self.room_id, e)
+        except Exception:  # noqa
+            logger.exception('room=%d _send_heartbeat() failed:', self.room_id)
+
+    async def _on_ws_message(self, message: aiohttp.WSMessage):
+        """
+        收到WebSocket消息
+
+        :param message: WebSocket消息
+        """
+        if message.type != aiohttp.WSMsgType.BINARY:
+            logger.warning('room=%d unknown websocket message type=%s, data=%s', self.room_id,
+                           message.type, message.data)
+            return
+
+        try:
+            await self._parse_ws_message(message.data)
+        except AuthError:
+            # 认证失败，让外层处理
+            raise
+        except Exception:  # noqa
+            logger.exception('room=%d _parse_ws_message() error:', self.room_id)
+
+    async def _parse_ws_message(self, data: bytes):
+        """
+        解析WebSocket消息
+
+        :param data: WebSocket消息数据
+        """
+        offset = 0
+        try:
+            header = HeaderTuple(*HEADER_STRUCT.unpack_from(data, offset))
+        except struct.error:
+            logger.exception('room=%d parsing header failed, offset=%d, data=%s', self.room_id, offset, data)
+            return
+
+        if header.operation in (Operation.SEND_MSG_REPLY, Operation.AUTH_REPLY):
+            # 业务消息，可能有多个包一起发，需要分包
+            while True:
+                body = data[offset + header.raw_header_size: offset + header.pack_len]
+                await self._parse_business_message(header, body)
+
+                offset += header.pack_len
+                if offset >= len(data):
+                    break
+
+                try:
+                    header = HeaderTuple(*HEADER_STRUCT.unpack_from(data, offset))
+                except struct.error:
+                    logger.exception('room=%d parsing header failed, offset=%d, data=%s', self.room_id, offset, data)
+                    break
+
+        elif header.operation == Operation.HEARTBEAT_REPLY:
+            # 服务器心跳包，前4字节是人气值，后面是客户端发的心跳包内容
+            # pack_len不包括客户端发的心跳包内容，不知道是不是服务器BUG
+            body = data[offset + header.raw_header_size: offset + header.raw_header_size + 4]
+            popularity = int.from_bytes(body, 'big')
+            # 自己造个消息当成业务消息处理
+            body = {
+                'cmd': '_HEARTBEAT',
+                'data': {
+                    'popularity': popularity
+                }
+            }
+            self._handle_command(body)
+
+        else:
+            # 未知消息
+            body = data[offset + header.raw_header_size: offset + header.pack_len]
+            logger.warning('room=%d unknown message operation=%d, header=%s, body=%s', self.room_id,
+                           header.operation, header, body)
+
+    async def _parse_business_message(self, header: HeaderTuple, body: bytes):
+        """
+        解析业务消息
+        """
+        if header.operation == Operation.SEND_MSG_REPLY:
+            # 业务消息
+            if header.ver == ProtoVer.BROTLI:
+                # 压缩过的先解压，为了避免阻塞网络线程，放在其他线程执行
+                body = await asyncio.get_running_loop().run_in_executor(None, brotli.decompress, body)
+                await self._parse_ws_message(body)
+            elif header.ver == ProtoVer.DEFLATE:
+                # web端已经不用zlib压缩了，但是开放平台会用
+                body = await asyncio.get_running_loop().run_in_executor(None, zlib.decompress, body)
+                await self._parse_ws_message(body)
+            elif header.ver == ProtoVer.NORMAL:
+                # 没压缩过的直接反序列化，因为有万恶的GIL，这里不能并行避免阻塞
+                if len(body) != 0:
+                    try:
+                        body = json.loads(body.decode('utf-8'))
+                        self._handle_command(body)
+                    except Exception:
+                        logger.error('room=%d, body=%s', self.room_id, body)
+                        raise
+            else:
+                # 未知格式
+                logger.warning('room=%d unknown protocol version=%d, header=%s, body=%s', self.room_id,
+                               header.ver, header, body)
+
+        elif header.operation == Operation.AUTH_REPLY:
+            # 认证响应
+            body = json.loads(body.decode('utf-8'))
+            if body['code'] != AuthReplyCode.OK:
+                raise AuthError(f"auth reply error, code={body['code']}, body={body}")
+            await self._websocket.send_bytes(self._make_packet({}, Operation.HEARTBEAT))
+
+        else:
+            # 未知消息
+            logger.warning('room=%d unknown message operation=%d, header=%s, body=%s', self.room_id,
+                           header.operation, header, body)
+
+    def _handle_command(self, command: dict):
+        """
+        处理业务消息
+
+        :param command: 业务消息
+        """
+        if self._handler is None:
+            return
+        try:
+            # 为什么不做成异步的：
+            # 1. 为了保持处理消息的顺序，这里不使用call_soon、create_task等方法延迟处理
+            # 2. 如果支持handle使用async函数，用户可能会在里面处理耗时很长的异步操作，导致网络协程阻塞
+            # 这里做成同步的，强制用户使用create_task或消息队列处理异步操作，这样就不会阻塞网络协程
+            self._handler.handle(self, command)
+        except Exception as e:
+            logger.exception('room=%d _handle_command() failed, command=%s', self.room_id, command, exc_info=e)

blivedm/handlers.py

@@ -0,0 +1,199 @@
+# -*- coding: utf-8 -*-
+import logging
+from typing import *
+
+from .clients import ws_base
+from .models import web as web_models, open_live as open_models
+
+__all__ = (
+    'HandlerInterface',
+    'BaseHandler',
+)
+
+logger = logging.getLogger('blivedm')
+
+logged_unknown_cmds = {
+    'COMBO_SEND',
+    'ENTRY_EFFECT',
+    'HOT_RANK_CHANGED',
+    'HOT_RANK_CHANGED_V2',
+    'INTERACT_WORD',
+    'LIVE',
+    'LIVE_INTERACTIVE_GAME',
+    'NOTICE_MSG',
+    'ONLINE_RANK_COUNT',
+    'ONLINE_RANK_TOP3',
+    'ONLINE_RANK_V2',
+    'PK_BATTLE_END',
+    'PK_BATTLE_FINAL_PROCESS',
+    'PK_BATTLE_PROCESS',
+    'PK_BATTLE_PROCESS_NEW',
+    'PK_BATTLE_SETTLE',
+    'PK_BATTLE_SETTLE_USER',
+    'PK_BATTLE_SETTLE_V2',
+    'PREPARING',
+    'ROOM_REAL_TIME_MESSAGE_UPDATE',
+    'STOP_LIVE_ROOM_LIST',
+    'SUPER_CHAT_MESSAGE_JPN',
+    'WIDGET_BANNER',
+}
+"""已打日志的未知cmd"""
+
+
+class HandlerInterface:
+    """
+    直播消息处理器接口
+    """
+
+    def handle(self, client: ws_base.WebSocketClientBase, command: dict):
+        raise NotImplementedError
+
+    def on_client_stopped(self, client: ws_base.WebSocketClientBase, exception: Optional[Exception]):
+        """
+        当客户端停止时调用。可以在这里close或者重新start
+        """
+
+
+def _make_msg_callback(method_name, message_cls):
+    def callback(self: 'BaseHandler', client: ws_base.WebSocketClientBase, command: dict):
+        method = getattr(self, method_name)
+        return method(client, message_cls.from_command(command['data']))
+    return callback
+
+
+class BaseHandler(HandlerInterface):
+    """
+    一个简单的消息处理器实现，带消息分发和消息类型转换。继承并重写_on_xxx方法即可实现自己的处理器
+    """
+
+    def __danmu_msg_callback(self, client: ws_base.WebSocketClientBase, command: dict):
+        return self._on_danmaku(client, web_models.DanmakuMessage.from_command(command['info']))
+
+    _CMD_CALLBACK_DICT: Dict[
+        str,
+        Optional[Callable[
+            ['BaseHandler', ws_base.WebSocketClientBase, dict],
+            Any
+        ]]
+    ] = {
+        # 收到心跳包，这是blivedm自造的消息，原本的心跳包格式不一样
+        '_HEARTBEAT': _make_msg_callback('_on_heartbeat', web_models.HeartbeatMessage),
+        # 收到弹幕
+        # go-common\app\service\live\live-dm\service\v1\send.go
+        'DANMU_MSG': __danmu_msg_callback,
+        # 有人送礼
+        'SEND_GIFT': _make_msg_callback('_on_gift', web_models.GiftMessage),
+        # 有人上舰
+        'GUARD_BUY': _make_msg_callback('_on_buy_guard', web_models.GuardBuyMessage),
+        # 醒目留言
+        'SUPER_CHAT_MESSAGE': _make_msg_callback('_on_super_chat', web_models.SuperChatMessage),
+        # 删除醒目留言
+        'SUPER_CHAT_MESSAGE_DELETE': _make_msg_callback('_on_super_chat_delete', web_models.SuperChatDeleteMessage),
+
+        #
+        # 开放平台消息
+        #
+
+        # 收到弹幕
+        'LIVE_OPEN_PLATFORM_DM': _make_msg_callback('_on_open_live_danmaku', open_models.DanmakuMessage),
+        # 有人送礼
+        'LIVE_OPEN_PLATFORM_SEND_GIFT': _make_msg_callback('_on_open_live_gift', open_models.GiftMessage),
+        # 有人上舰
+        'LIVE_OPEN_PLATFORM_GUARD': _make_msg_callback('_on_open_live_buy_guard', open_models.GuardBuyMessage),
+        # 醒目留言
+        'LIVE_OPEN_PLATFORM_SUPER_CHAT': _make_msg_callback('_on_open_live_super_chat', open_models.SuperChatMessage),
+        # 删除醒目留言
+        'LIVE_OPEN_PLATFORM_SUPER_CHAT_DEL': _make_msg_callback(
+            '_on_open_live_super_chat_delete', open_models.SuperChatDeleteMessage
+        ),
+        # 点赞
+        'LIVE_OPEN_PLATFORM_LIKE': _make_msg_callback('_on_open_live_like', open_models.LikeMessage),
+    }
+    """cmd -> 处理回调"""
+
+    def handle(self, client: ws_base.WebSocketClientBase, command: dict):
+        cmd = command.get('cmd', '')
+        pos = cmd.find(':')  # 2019-5-29 B站弹幕升级新增了参数
+        if pos != -1:
+            cmd = cmd[:pos]
+
+        if cmd not in self._CMD_CALLBACK_DICT:
+            # 只有第一次遇到未知cmd时打日志
+            if cmd not in logged_unknown_cmds:
+                logger.warning('room=%d unknown cmd=%s, command=%s', client.room_id, cmd, command)
+                logged_unknown_cmds.add(cmd)
+            return
+
+        callback = self._CMD_CALLBACK_DICT[cmd]
+        if callback is not None:
+            callback(self, client, command)
+
+    def _on_heartbeat(self, client: ws_base.WebSocketClientBase, message: web_models.HeartbeatMessage):
+        """
+        收到心跳包
+        """
+
+    def _on_danmaku(self, client: ws_base.WebSocketClientBase, message: web_models.DanmakuMessage):
+        """
+        收到弹幕
+        """
+
+    def _on_gift(self, client: ws_base.WebSocketClientBase, message: web_models.GiftMessage):
+        """
+        收到礼物
+        """
+
+    def _on_buy_guard(self, client: ws_base.WebSocketClientBase, message: web_models.GuardBuyMessage):
+        """
+        有人上舰
+        """
+
+    def _on_super_chat(self, client: ws_base.WebSocketClientBase, message: web_models.SuperChatMessage):
+        """
+        醒目留言
+        """
+
+    def _on_super_chat_delete(
+        self, client: ws_base.WebSocketClientBase, message: web_models.SuperChatDeleteMessage
+    ):
+        """
+        删除醒目留言
+        """
+
+    #
+    # 开放平台消息
+    #
+
+    def _on_open_live_danmaku(self, client: ws_base.WebSocketClientBase, message: open_models.DanmakuMessage):
+        """
+        收到弹幕
+        """
+
+    def _on_open_live_gift(self, client: ws_base.WebSocketClientBase, message: open_models.GiftMessage):
+        """
+        收到礼物
+        """
+
+    def _on_open_live_buy_guard(self, client: ws_base.WebSocketClientBase, message: open_models.GuardBuyMessage):
+        """
+        有人上舰
+        """
+
+    def _on_open_live_super_chat(
+        self, client: ws_base.WebSocketClientBase, message: open_models.SuperChatMessage
+    ):
+        """
+        醒目留言
+        """
+
+    def _on_open_live_super_chat_delete(
+        self, client: ws_base.WebSocketClientBase, message: open_models.SuperChatDeleteMessage
+    ):
+        """
+        删除醒目留言
+        """
+
+    def _on_open_live_like(self, client: ws_base.WebSocketClientBase, message: open_models.LikeMessage):
+        """
+        点赞
+        """

blivedm/models/__init__.py

@@ -0,0 +1 @@
+# -*- coding: utf-8 -*-

blivedm/models/open_live.py

@@ -0,0 +1,390 @@
+# -*- coding: utf-8 -*-
+import dataclasses
+from typing import *
+
+__all__ = (
+    'DanmakuMessage',
+    'GiftMessage',
+    'GuardBuyMessage',
+    'SuperChatMessage',
+    'SuperChatDeleteMessage',
+    'LikeMessage',
+)
+
+# 注释都是复制自官方文档的，看不懂的话问B站
+# https://open-live.bilibili.com/document/f9ce25be-312e-1f4a-85fd-fef21f1637f8
+
+
+@dataclasses.dataclass
+class DanmakuMessage:
+    """
+    弹幕消息
+    """
+
+    uname: str = ''
+    """用户昵称"""
+    open_id: str = ''
+    """用户唯一标识"""
+    uface: str = ''
+    """用户头像"""
+    timestamp: int = 0
+    """弹幕发送时间秒级时间戳"""
+    room_id: int = 0
+    """弹幕接收的直播间"""
+    msg: str = ''
+    """弹幕内容"""
+    msg_id: str = ''
+    """消息唯一id"""
+    guard_level: int = 0
+    """对应房间大航海等级"""
+    fans_medal_wearing_status: bool = False
+    """该房间粉丝勋章佩戴情况"""
+    fans_medal_name: str = ''
+    """粉丝勋章名"""
+    fans_medal_level: int = 0
+    """对应房间勋章信息"""
+    emoji_img_url: str = ''
+    """表情包图片地址"""
+    dm_type: int = 0
+    """弹幕类型 0：普通弹幕 1：表情包弹幕"""
+
+    @classmethod
+    def from_command(cls, data: dict):
+        return cls(
+            uname=data['uname'],
+            open_id=data['open_id'],
+            uface=data['uface'],
+            timestamp=data['timestamp'],
+            room_id=data['room_id'],
+            msg=data['msg'],
+            msg_id=data['msg_id'],
+            guard_level=data['guard_level'],
+            fans_medal_wearing_status=data['fans_medal_wearing_status'],
+            fans_medal_name=data['fans_medal_name'],
+            fans_medal_level=data['fans_medal_level'],
+            emoji_img_url=data['emoji_img_url'],
+            dm_type=data['dm_type'],
+        )
+
+
+@dataclasses.dataclass
+class AnchorInfo:
+    """
+    主播信息
+    """
+
+    uid: int = 0
+    """收礼主播uid"""
+    open_id: str = ''
+    """收礼主播唯一标识"""
+    uname: str = ''
+    """收礼主播昵称"""
+    uface: str = ''
+    """收礼主播头像"""
+
+    @classmethod
+    def from_dict(cls, data: dict):
+        return cls(
+            uid=data['uid'],
+            open_id=data['open_id'],
+            uname=data['uname'],
+            uface=data['uface'],
+        )
+
+
+@dataclasses.dataclass
+class ComboInfo:
+    """
+    连击信息
+    """
+
+    combo_base_num: int = 0
+    """每次连击赠送的道具数量"""
+    combo_count: int = 0
+    """连击次数"""
+    combo_id: str = ''
+    """连击id"""
+    combo_timeout: int = 0
+    """连击有效期秒"""
+
+    @classmethod
+    def from_dict(cls, data: dict):
+        return cls(
+            combo_base_num=data['combo_base_num'],
+            combo_count=data['combo_count'],
+            combo_id=data['combo_id'],
+            combo_timeout=data['combo_timeout'],
+        )
+
+
+@dataclasses.dataclass
+class GiftMessage:
+    """
+    礼物消息
+    """
+
+    room_id: int = 0
+    """房间号"""
+    open_id: str = ''
+    """用户唯一标识"""
+    uname: str = ''
+    """送礼用户昵称"""
+    uface: str = ''
+    """送礼用户头像"""
+    gift_id: int = 0
+    """道具id(盲盒:爆出道具id)"""
+    gift_name: str = ''
+    """道具名(盲盒:爆出道具名)"""
+    gift_num: int = 0
+    """赠送道具数量"""
+    price: int = 0
+    """礼物爆出单价，(1000 = 1元 = 10电池),盲盒:爆出道具的价值"""
+    paid: bool = False
+    """是否是付费道具"""
+    fans_medal_level: int = 0
+    """实际送礼人的勋章信息"""
+    fans_medal_name: str = ''
+    """粉丝勋章名"""
+    fans_medal_wearing_status: bool = False
+    """该房间粉丝勋章佩戴情况"""
+    guard_level: int = 0
+    """大航海等级"""
+    timestamp: int = 0
+    """收礼时间秒级时间戳"""
+    anchor_info: AnchorInfo = dataclasses.field(default_factory=AnchorInfo)
+    """主播信息"""
+    msg_id: str = ''
+    """消息唯一id"""
+    gift_icon: str = ''
+    """道具icon"""
+    combo_gift: bool = False
+    """是否是combo道具"""
+    combo_info: ComboInfo = dataclasses.field(default_factory=ComboInfo)
+    """连击信息"""
+
+    @classmethod
+    def from_command(cls, data: dict):
+        combo_info = data.get('combo_info', None)
+        if combo_info is None:
+            combo_info = ComboInfo()
+        else:
+            combo_info = ComboInfo.from_dict(combo_info)
+
+        return cls(
+            room_id=data['room_id'],
+            open_id=data['open_id'],
+            uname=data['uname'],
+            uface=data['uface'],
+            gift_id=data['gift_id'],
+            gift_name=data['gift_name'],
+            gift_num=data['gift_num'],
+            price=data['price'],
+            paid=data['paid'],
+            fans_medal_level=data['fans_medal_level'],
+            fans_medal_name=data['fans_medal_name'],
+            fans_medal_wearing_status=data['fans_medal_wearing_status'],
+            guard_level=data['guard_level'],
+            timestamp=data['timestamp'],
+            anchor_info=AnchorInfo.from_dict(data['anchor_info']),
+            msg_id=data['msg_id'],
+            gift_icon=data['gift_icon'],
+            combo_gift=data.get('combo_gift', False),  # 官方的调试工具没发这个字段
+            combo_info=combo_info,  # 官方的调试工具没发这个字段
+        )
+
+
+@dataclasses.dataclass
+class UserInfo:
+    """
+    用户信息
+    """
+
+    open_id: str = ''
+    """用户唯一标识"""
+    uname: str = ''
+    """用户昵称"""
+    uface: str = ''
+    """用户头像"""
+
+    @classmethod
+    def from_dict(cls, data: dict):
+        return cls(
+            open_id=data['open_id'],
+            uname=data['uname'],
+            uface=data['uface'],
+        )
+
+
+@dataclasses.dataclass
+class GuardBuyMessage:
+    """
+    上舰消息
+    """
+
+    user_info: UserInfo = dataclasses.field(default_factory=UserInfo)
+    """用户信息"""
+    guard_level: int = 0
+    """大航海等级"""
+    guard_num: int = 0
+    """大航海数量"""
+    guard_unit: str = ''
+    """大航海单位"""
+    price: int = 0
+    """大航海金瓜子"""
+    fans_medal_level: int = 0
+    """粉丝勋章等级"""
+    fans_medal_name: str = ''
+    """粉丝勋章名"""
+    fans_medal_wearing_status: bool = False
+    """该房间粉丝勋章佩戴情况"""
+    room_id: int = 0
+    """房间号"""
+    msg_id: str = ''
+    """消息唯一id"""
+    timestamp: int = 0
+    """上舰时间秒级时间戳"""
+
+    @classmethod
+    def from_command(cls, data: dict):
+        return cls(
+            user_info=UserInfo.from_dict(data['user_info']),
+            guard_level=data['guard_level'],
+            guard_num=data['guard_num'],
+            guard_unit=data['guard_unit'],
+            price=data['price'],
+            fans_medal_level=data['fans_medal_level'],
+            fans_medal_name=data['fans_medal_name'],
+            fans_medal_wearing_status=data['fans_medal_wearing_status'],
+            room_id=data['room_id'],
+            msg_id=data['msg_id'],
+            timestamp=data['timestamp'],
+        )
+
+
+@dataclasses.dataclass
+class SuperChatMessage:
+    """
+    醒目留言消息
+    """
+
+    room_id: int = 0
+    """直播间id"""
+    open_id: str = ''
+    """用户唯一标识"""
+    uname: str = ''
+    """购买的用户昵称"""
+    uface: str = ''
+    """购买用户头像"""
+    message_id: int = 0
+    """留言id(风控场景下撤回留言需要)"""
+    message: str = ''
+    """留言内容"""
+    rmb: int = 0
+    """支付金额(元)"""
+    timestamp: int = 0
+    """赠送时间秒级"""
+    start_time: int = 0
+    """生效开始时间"""
+    end_time: int = 0
+    """生效结束时间"""
+    guard_level: int = 0
+    """对应房间大航海等级"""
+    fans_medal_level: int = 0
+    """对应房间勋章信息"""
+    fans_medal_name: str = ''
+    """对应房间勋章名字"""
+    fans_medal_wearing_status: bool = False
+    """该房间粉丝勋章佩戴情况"""
+    msg_id: str = ''
+    """消息唯一id"""
+
+    @classmethod
+    def from_command(cls, data: dict):
+        return cls(
+            room_id=data['room_id'],
+            open_id=data['open_id'],
+            uname=data['uname'],
+            uface=data['uface'],
+            message_id=data['message_id'],
+            message=data['message'],
+            rmb=data['rmb'],
+            timestamp=data['timestamp'],
+            start_time=data['start_time'],
+            end_time=data['end_time'],
+            guard_level=data['guard_level'],
+            fans_medal_level=data['fans_medal_level'],
+            fans_medal_name=data['fans_medal_name'],
+            fans_medal_wearing_status=data['fans_medal_wearing_status'],
+            msg_id=data['msg_id'],
+        )
+
+
+@dataclasses.dataclass
+class SuperChatDeleteMessage:
+    """
+    删除醒目留言消息
+    """
+
+    room_id: int = 0
+    """直播间id"""
+    message_ids: List[int] = dataclasses.field(default_factory=list)
+    """留言id"""
+    msg_id: str = ''
+    """消息唯一id"""
+
+    @classmethod
+    def from_command(cls, data: dict):
+        return cls(
+            room_id=data['room_id'],
+            message_ids=data['message_ids'],
+            msg_id=data['msg_id'],
+        )
+
+
+@dataclasses.dataclass
+class LikeMessage:
+    """
+    点赞消息
+
+    请注意：用户端每分钟触发若干次的情况下只会推送一次该消息
+    """
+
+    uname: str = ''
+    """用户昵称"""
+    open_id: str = ''
+    """用户唯一标识"""
+    uface: str = ''
+    """用户头像"""
+    timestamp: int = 0
+    """时间秒级时间戳"""
+    room_id: int = 0
+    """发生的直播间"""
+    like_text: str = ''
+    """点赞文案(“xxx点赞了”)"""
+    like_count: int = 0
+    """对单个用户最近2秒的点赞次数聚合"""
+    fans_medal_wearing_status: bool = False
+    """该房间粉丝勋章佩戴情况"""
+    fans_medal_name: str = ''
+    """粉丝勋章名"""
+    fans_medal_level: int = 0
+    """对应房间勋章信息"""
+    msg_id: str = ''  # 官方文档表格里没列出这个字段，但是参考JSON里面有
+    """消息唯一id"""
+    # 还有个guard_level，但官方文档没有出现这个字段，就不添加了
+
+    @classmethod
+    def from_command(cls, data: dict):
+        return cls(
+            uname=data['uname'],
+            open_id=data['open_id'],
+            uface=data['uface'],
+            timestamp=data['timestamp'],
+            room_id=data['room_id'],
+            like_text=data['like_text'],
+            like_count=data['like_count'],
+            fans_medal_wearing_status=data['fans_medal_wearing_status'],
+            fans_medal_name=data['fans_medal_name'],
+            fans_medal_level=data['fans_medal_level'],
+            msg_id=data.get('msg_id', ''),  # 官方文档表格里没列出这个字段，但是参考JSON里面有
+        )

blivedm/models/web.py

@@ -0,0 +1,390 @@
+# -*- coding: utf-8 -*-
+import dataclasses
+import json
+from typing import *
+
+__all__ = (
+    'HeartbeatMessage',
+    'DanmakuMessage',
+    'GiftMessage',
+    'GuardBuyMessage',
+    'SuperChatMessage',
+    'SuperChatDeleteMessage',
+)
+
+
+@dataclasses.dataclass
+class HeartbeatMessage:
+    """
+    心跳消息
+    """
+
+    popularity: int = 0
+    """人气值，已废弃"""
+
+    @classmethod
+    def from_command(cls, data: dict):
+        return cls(
+            popularity=data['popularity'],
+        )
+
+
+@dataclasses.dataclass
+class DanmakuMessage:
+    """
+    弹幕消息
+    """
+
+    mode: int = 0
+    """弹幕显示模式（滚动、顶部、底部）"""
+    font_size: int = 0
+    """字体尺寸"""
+    color: int = 0
+    """颜色"""
+    timestamp: int = 0
+    """时间戳（毫秒）"""
+    rnd: int = 0
+    """随机数，前端叫作弹幕ID，可能是去重用的"""
+    uid_crc32: str = ''
+    """用户ID文本的CRC32"""
+    msg_type: int = 0
+    """是否礼物弹幕（节奏风暴）"""
+    bubble: int = 0
+    """右侧评论栏气泡"""
+    dm_type: int = 0
+    """弹幕类型，0文本，1表情，2语音"""
+    emoticon_options: Union[dict, str] = ''
+    """表情参数"""
+    voice_config: Union[dict, str] = ''
+    """语音参数"""
+    mode_info: dict = dataclasses.field(default_factory=dict)
+    """一些附加参数"""
+
+    msg: str = ''
+    """弹幕内容"""
+
+    uid: int = 0
+    """用户ID"""
+    uname: str = ''
+    """用户名"""
+    admin: int = 0
+    """是否房管"""
+    vip: int = 0
+    """是否月费老爷"""
+    svip: int = 0
+    """是否年费老爷"""
+    urank: int = 0
+    """用户身份，用来判断是否正式会员，猜测非正式会员为5000，正式会员为10000"""
+    mobile_verify: int = 0
+    """是否绑定手机"""
+    uname_color: str = ''
+    """用户名颜色"""
+
+    medal_level: str = ''
+    """勋章等级"""
+    medal_name: str = ''
+    """勋章名"""
+    runame: str = ''
+    """勋章房间主播名"""
+    medal_room_id: int = 0
+    """勋章房间ID"""
+    mcolor: int = 0
+    """勋章颜色"""
+    special_medal: str = ''
+    """特殊勋章"""
+
+    user_level: int = 0
+    """用户等级"""
+    ulevel_color: int = 0
+    """用户等级颜色"""
+    ulevel_rank: str = ''
+    """用户等级排名，>50000时为'>50000'"""
+
+    old_title: str = ''
+    """旧头衔"""
+    title: str = ''
+    """头衔"""
+
+    privilege_type: int = 0
+    """舰队类型，0非舰队，1总督，2提督，3舰长"""
+
+    @classmethod
+    def from_command(cls, info: list):
+        if len(info[3]) != 0:
+            medal_level = info[3][0]
+            medal_name = info[3][1]
+            runame = info[3][2]
+            room_id = info[3][3]
+            mcolor = info[3][4]
+            special_medal = info[3][5]
+        else:
+            medal_level = 0
+            medal_name = ''
+            runame = ''
+            room_id = 0
+            mcolor = 0
+            special_medal = 0
+
+        if len(info[5]) != 0:
+            old_title = info[5][0]
+            title = info[5][1]
+        else:
+            old_title = ''
+            title = ''
+
+        return cls(
+            mode=info[0][1],
+            font_size=info[0][2],
+            color=info[0][3],
+            timestamp=info[0][4],
+            rnd=info[0][5],
+            uid_crc32=info[0][7],
+            msg_type=info[0][9],
+            bubble=info[0][10],
+            dm_type=info[0][12],
+            emoticon_options=info[0][13],
+            voice_config=info[0][14],
+            mode_info=info[0][15],
+
+            msg=info[1],
+
+            uid=info[2][0],
+            uname=info[2][1],
+            admin=info[2][2],
+            vip=info[2][3],
+            svip=info[2][4],
+            urank=info[2][5],
+            mobile_verify=info[2][6],
+            uname_color=info[2][7],
+
+            medal_level=medal_level,
+            medal_name=medal_name,
+            runame=runame,
+            medal_room_id=room_id,
+            mcolor=mcolor,
+            special_medal=special_medal,
+
+            user_level=info[4][0],
+            ulevel_color=info[4][2],
+            ulevel_rank=info[4][3],
+
+            old_title=old_title,
+            title=title,
+
+            privilege_type=info[7],
+        )
+
+    @property
+    def emoticon_options_dict(self) -> dict:
+        """
+        示例：
+        {'bulge_display': 0, 'emoticon_unique': 'official_13', 'height': 60, 'in_player_area': 1, 'is_dynamic': 1,
+         'url': 'https://i0.hdslb.com/bfs/live/a98e35996545509188fe4d24bd1a56518ea5af48.png', 'width': 183}
+        """
+        if isinstance(self.emoticon_options, dict):
+            return self.emoticon_options
+        try:
+            return json.loads(self.emoticon_options)
+        except (json.JSONDecodeError, TypeError):
+            return {}
+
+    @property
+    def voice_config_dict(self) -> dict:
+        """
+        示例：
+        {'voice_url': 'https%3A%2F%2Fboss.hdslb.com%2Flive-dm-voice%2Fb5b26e48b556915cbf3312a59d3bb2561627725945.wav
+         %3FX-Amz-Algorithm%3DAWS4-HMAC-SHA256%26X-Amz-Credential%3D2663ba902868f12f%252F20210731%252Fshjd%252Fs3%25
+         2Faws4_request%26X-Amz-Date%3D20210731T100545Z%26X-Amz-Expires%3D600000%26X-Amz-SignedHeaders%3Dhost%26
+         X-Amz-Signature%3D114e7cb5ac91c72e231c26d8ca211e53914722f36309b861a6409ffb20f07ab8',
+         'file_format': 'wav', 'text': '汤，下午好。', 'file_duration': 1}
+        """
+        if isinstance(self.voice_config, dict):
+            return self.voice_config
+        try:
+            return json.loads(self.voice_config)
+        except (json.JSONDecodeError, TypeError):
+            return {}
+
+
+@dataclasses.dataclass
+class GiftMessage:
+    """
+    礼物消息
+    """
+
+    gift_name: str = ''
+    """礼物名"""
+    num: int = 0
+    """数量"""
+    uname: str = ''
+    """用户名"""
+    face: str = ''
+    """用户头像URL"""
+    guard_level: int = 0
+    """舰队等级，0非舰队，1总督，2提督，3舰长"""
+    uid: int = 0
+    """用户ID"""
+    timestamp: int = 0
+    """时间戳"""
+    gift_id: int = 0
+    """礼物ID"""
+    gift_type: int = 0
+    """礼物类型（未知）"""
+    action: str = ''
+    """目前遇到的有'喂食'、'赠送'"""
+    price: int = 0
+    """礼物单价瓜子数"""
+    rnd: str = ''
+    """随机数，可能是去重用的。有时是时间戳+去重ID，有时是UUID"""
+    coin_type: str = ''
+    """瓜子类型，'silver'或'gold'，1000金瓜子 = 1元"""
+    total_coin: int = 0
+    """总瓜子数"""
+    tid: str = ''
+    """可能是事务ID，有时和rnd相同"""
+
+    @classmethod
+    def from_command(cls, data: dict):
+        return cls(
+            gift_name=data['giftName'],
+            num=data['num'],
+            uname=data['uname'],
+            face=data['face'],
+            guard_level=data['guard_level'],
+            uid=data['uid'],
+            timestamp=data['timestamp'],
+            gift_id=data['giftId'],
+            gift_type=data['giftType'],
+            action=data['action'],
+            price=data['price'],
+            rnd=data['rnd'],
+            coin_type=data['coin_type'],
+            total_coin=data['total_coin'],
+            tid=data['tid'],
+        )
+
+
+@dataclasses.dataclass
+class GuardBuyMessage:
+    """
+    上舰消息
+    """
+
+    uid: int = 0
+    """用户ID"""
+    username: str = ''
+    """用户名"""
+    guard_level: int = 0
+    """舰队等级，0非舰队，1总督，2提督，3舰长"""
+    num: int = 0
+    """数量"""
+    price: int = 0
+    """单价金瓜子数"""
+    gift_id: int = 0
+    """礼物ID"""
+    gift_name: str = ''
+    """礼物名"""
+    start_time: int = 0
+    """开始时间戳，和结束时间戳相同"""
+    end_time: int = 0
+    """结束时间戳，和开始时间戳相同"""
+
+    @classmethod
+    def from_command(cls, data: dict):
+        return cls(
+            uid=data['uid'],
+            username=data['username'],
+            guard_level=data['guard_level'],
+            num=data['num'],
+            price=data['price'],
+            gift_id=data['gift_id'],
+            gift_name=data['gift_name'],
+            start_time=data['start_time'],
+            end_time=data['end_time'],
+        )
+
+
+@dataclasses.dataclass
+class SuperChatMessage:
+    """
+    醒目留言消息
+    """
+
+    price: int = 0
+    """价格（人民币）"""
+    message: str = ''
+    """消息"""
+    message_trans: str = ''
+    """消息日文翻译（目前只出现在SUPER_CHAT_MESSAGE_JPN）"""
+    start_time: int = 0
+    """开始时间戳"""
+    end_time: int = 0
+    """结束时间戳"""
+    time: int = 0
+    """剩余时间（约等于 结束时间戳 - 开始时间戳）"""
+    id: int = 0
+    """醒目留言ID，删除时用"""
+    gift_id: int = 0
+    """礼物ID"""
+    gift_name: str = ''
+    """礼物名"""
+    uid: int = 0
+    """用户ID"""
+    uname: str = ''
+    """用户名"""
+    face: str = ''
+    """用户头像URL"""
+    guard_level: int = 0
+    """舰队等级，0非舰队，1总督，2提督，3舰长"""
+    user_level: int = 0
+    """用户等级"""
+    background_bottom_color: str = ''
+    """底部背景色，'#rrggbb'"""
+    background_color: str = ''
+    """背景色，'#rrggbb'"""
+    background_icon: str = ''
+    """背景图标"""
+    background_image: str = ''
+    """背景图URL"""
+    background_price_color: str = ''
+    """背景价格颜色，'#rrggbb'"""
+
+    @classmethod
+    def from_command(cls, data: dict):
+        return cls(
+            price=data['price'],
+            message=data['message'],
+            message_trans=data['message_trans'],
+            start_time=data['start_time'],
+            end_time=data['end_time'],
+            time=data['time'],
+            id=data['id'],
+            gift_id=data['gift']['gift_id'],
+            gift_name=data['gift']['gift_name'],
+            uid=data['uid'],
+            uname=data['user_info']['uname'],
+            face=data['user_info']['face'],
+            guard_level=data['user_info']['guard_level'],
+            user_level=data['user_info']['user_level'],
+            background_bottom_color=data['background_bottom_color'],
+            background_color=data['background_color'],
+            background_icon=data['background_icon'],
+            background_image=data['background_image'],
+            background_price_color=data['background_price_color'],
+        )
+
+
+@dataclasses.dataclass
+class SuperChatDeleteMessage:
+    """
+    删除醒目留言消息
+    """
+
+    ids: List[int] = dataclasses.field(default_factory=list)
+    """醒目留言ID数组"""
+
+    @classmethod
+    def from_command(cls, data: dict):
+        return cls(
+            ids=data['ids'],
+        )

blivedm/utils.py

@@ -0,0 +1,19 @@
+# -*- coding: utf-8 -*-
+USER_AGENT = (
+    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/102.0.0.0 Safari/537.36'
+)
+
+
+def make_constant_retry_policy(interval: float):
+    def get_interval(_retry_count: int, _total_retry_count: int):
+        return interval
+    return get_interval
+
+
+def make_linear_retry_policy(start_interval: float, interval_step: float, max_interval: float):
+    def get_interval(retry_count: int, _total_retry_count: int):
+        return min(
+            start_interval + (retry_count - 1) * interval_step,
+            max_interval
+        )
+    return get_interval