update

# Conflicts:
#	blivedm/clients/web.py

.gitignore

@@ -100,3 +100,12 @@ ENV/
 
 
 .idea/
+.pdm-python
+pdm.lock
+pdm.toml
+
+cookie.txt
+logs
+
+config.json
+

README.md

@@ -2,7 +2,7 @@
 
 Python获取bilibili直播弹幕的库，使用WebSocket协议，支持web端和B站直播开放平台两种接口
 
-[协议解释](https://blog.csdn.net/xfgryujk/article/details/80306776)（有点过时了，总体是没错的）
+[协议解释](https://open-live.bilibili.com/document/657d8e34-f926-a133-16c0-300c1afc6e6b)
 
 基于本库开发的一个应用：[blivechat](https://github.com/xfgryujk/blivechat)
 
@@ -15,4 +15,8 @@ Python获取bilibili直播弹幕的库，使用WebSocket协议，支持web端和
     pip install -r requirements.txt
     ```
 
-3. 例程看[sample.py](./sample.py)和[open_live_sample.py](./open_live_sample.py)
+<<<<<<< HEAD
+3. 例程看[sample.py](./main.py)和[open_live_sample.py](./open_live_sample.py)
+=======
+3. web端例程在[sample.py](./sample.py)，B站直播开放平台例程在[open_live_sample.py](./open_live_sample.py)
+>>>>>>> github/HEAD

blivedm/__init__.py

@@ -1,3 +1,5 @@
 # -*- coding: utf-8 -*-
+__version__ = '1.1.3'
+
 from .handlers import *
 from .clients import *

blivedm/clients/open_live.py

@@ -60,6 +60,8 @@ class OpenLiveClient(ws_base.WebSocketClientBase):
         # 在调用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
@@ -78,6 +80,13 @@ class OpenLiveClient(ws_base.WebSocketClientBase):
         """
         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):
         """
@@ -181,6 +190,7 @@ class OpenLiveClient(ws_base.WebSocketClientBase):
         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):
@@ -281,3 +291,16 @@ class OpenLiveClient(ws_base.WebSocketClientBase):
         发送认证包
         """
         await self._websocket.send_bytes(self._make_packet(self._auth_body, ws_base.Operation.AUTH))
+
+    def _handle_command(self, command: dict):
+        cmd = command.get('cmd', '')
+        if cmd == 'LIVE_OPEN_PLATFORM_INTERACTION_END' and command['data']['game_id'] == self._game_id:
+            # 服务器主动停止推送，可能是心跳超时，需要重新开启项目
+            logger.warning('room=%d game end by server, game_id=%s', self._room_id, self._game_id)
+
+            self._need_init_room = True
+            if self._websocket is not None and not self._websocket.closed:
+                asyncio.create_task(self._websocket.close())
+            return
+
+        super()._handle_command(command)

blivedm/clients/web.py

@@ -1,6 +1,10 @@
 # -*- coding: utf-8 -*-
 import asyncio
+import datetime
+import hashlib
 import logging
+import urllib
+import weakref
 from typing import *
 
 import aiohttp
@@ -16,13 +20,135 @@ __all__ = (
 logger = logging.getLogger('blivedm')
 
 UID_INIT_URL = 'https://api.bilibili.com/x/web-interface/nav'
-BUVID_INIT_URL = 'https://data.bilibili.com/v/'
+WBI_INIT_URL = UID_INIT_URL
-ROOM_INIT_URL = 'https://api.live.bilibili.com/xlive/web-room/v1/index/getInfoByRoom'
+BUVID_INIT_URL = 'https://www.bilibili.com/'
+ROOM_INIT_URL = 'https://api.live.bilibili.com/room/v1/Room/get_info'
 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}
 ]
 
+_session_to_wbi_signer = weakref.WeakKeyDictionary()
+
+
+def _get_wbi_signer(session: aiohttp.ClientSession) -> '_WbiSigner':
+    wbi_signer = _session_to_wbi_signer.get(session, None)
+    if wbi_signer is None:
+        wbi_signer = _session_to_wbi_signer[session] = _WbiSigner(session)
+    return wbi_signer
+
+
+class _WbiSigner:
+    WBI_KEY_INDEX_TABLE = [
+        46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35,
+        27, 43, 5, 49, 33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13
+    ]
+    """wbi密码表"""
+    WBI_KEY_TTL = datetime.timedelta(hours=11, minutes=59, seconds=30)
+
+    def __init__(self, session: aiohttp.ClientSession):
+        self._session = session
+
+        self._wbi_key = ''
+        """缓存的wbi鉴权口令"""
+        self._refresh_future: Optional[Awaitable] = None
+        """用来避免同时刷新"""
+        self._last_refresh_time: Optional[datetime.datetime] = None
+
+    @property
+    def wbi_key(self):
+        """
+        缓存的wbi鉴权口令
+        """
+        return self._wbi_key
+
+    def reset(self):
+        self._wbi_key = ''
+        self._last_refresh_time = None
+
+    @property
+    def need_refresh_wbi_key(self):
+        return self._wbi_key == '' or (
+            self._last_refresh_time is not None
+            and datetime.datetime.now() - self._last_refresh_time >= self.WBI_KEY_TTL
+        )
+
+    def refresh_wbi_key(self) -> Awaitable:
+        if self._refresh_future is None:
+            self._refresh_future = asyncio.create_task(self._do_refresh_wbi_key())
+
+            def on_done(_fu):
+                self._refresh_future = None
+            self._refresh_future.add_done_callback(on_done)
+
+        return self._refresh_future
+
+    async def _do_refresh_wbi_key(self):
+        wbi_key = await self._get_wbi_key()
+        if wbi_key == '':
+            return
+
+        self._wbi_key = wbi_key
+        self._last_refresh_time = datetime.datetime.now()
+
+    async def _get_wbi_key(self):
+        try:
+            async with self._session.get(
+                WBI_INIT_URL,
+                headers={'User-Agent': utils.USER_AGENT},
+            ) as res:
+                if res.status != 200:
+                    logger.warning('WbiSigner failed to get wbi key: status=%d %s', res.status, res.reason)
+                    return ''
+                data = await res.json()
+        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+            logger.exception('WbiSigner failed to get wbi key:')
+            return ''
+
+        try:
+            wbi_img = data['data']['wbi_img']
+            img_key = wbi_img['img_url'].rpartition('/')[2].partition('.')[0]
+            sub_key = wbi_img['sub_url'].rpartition('/')[2].partition('.')[0]
+        except KeyError:
+            logger.warning('WbiSigner failed to get wbi key: data=%s', data)
+            return ''
+
+        shuffled_key = img_key + sub_key
+        wbi_key = []
+        for index in self.WBI_KEY_INDEX_TABLE:
+            if index < len(shuffled_key):
+                wbi_key.append(shuffled_key[index])
+        return ''.join(wbi_key)
+
+    def add_wbi_sign(self, params: dict):
+        if self._wbi_key == '':
+            return params
+
+        wts = str(int(datetime.datetime.now().timestamp()))
+        params_to_sign = {**params, 'wts': wts}
+
+        # 按key字典序排序
+        params_to_sign = {
+            key: params_to_sign[key]
+            for key in sorted(params_to_sign.keys())
+        }
+        # 过滤一些字符
+        for key, value in params_to_sign.items():
+            value = ''.join(
+                ch
+                for ch in str(value)
+                if ch not in "!'()*"
+            )
+            params_to_sign[key] = value
+
+        str_to_sign = urllib.parse.urlencode(params_to_sign) + self._wbi_key
+        w_rid = hashlib.md5(str_to_sign.encode('utf-8')).hexdigest()
+        return {
+            **params,
+            'wts': wts,
+            'w_rid': w_rid
+        }
+
 
 class BLiveClient(ws_base.WebSocketClientBase):
     """
@@ -43,6 +169,7 @@ class BLiveClient(ws_base.WebSocketClientBase):
         heartbeat_interval=30,
     ):
         super().__init__(session, heartbeat_interval)
+        self._wbi_signer = _get_wbi_signer(self._session)
 
         self._tmp_room_id = room_id
         """用来init_room的临时房间ID，可以用短ID"""
@@ -194,20 +321,26 @@ class BLiveClient(ws_base.WebSocketClientBase):
         return True
 
     def _parse_room_init(self, data):
-        room_info = data['room_info']
+        self._room_id = data['room_id']
-        self._room_id = room_info['room_id']
+        self._room_owner_uid = data['uid']
-        self._room_owner_uid = room_info['uid']
         return True
 
     async def _init_host_server(self):
+        if self._wbi_signer.need_refresh_wbi_key:
+            await self._wbi_signer.refresh_wbi_key()
+            # 如果没刷新成功先用旧的key
+            if self._wbi_signer.wbi_key == '':
+                logger.exception('room=%d _init_host_server() failed: no wbi key', self._room_id)
+                return False
+
         try:
             async with self._session.get(
                 DANMAKU_SERVER_CONF_URL,
                 headers={'User-Agent': utils.USER_AGENT},
-                params={
+                params=self._wbi_signer.add_wbi_sign({
                     '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,
@@ -215,6 +348,9 @@ class BLiveClient(ws_base.WebSocketClientBase):
                     return False
                 data = await res.json()
                 if data['code'] != 0:
+                    if data['code'] == -352:
+                        # wbi签名错误
+                        self._wbi_signer.reset()
                     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']):

blivedm/handlers.py

@@ -17,7 +17,6 @@ logged_unknown_cmds = {
     'ENTRY_EFFECT',
     'HOT_RANK_CHANGED',
     'HOT_RANK_CHANGED_V2',
-    'INTERACT_WORD',
     'LIVE',
     'LIVE_INTERACTIVE_GAME',
     'NOTICE_MSG',
@@ -35,6 +34,7 @@ logged_unknown_cmds = {
     'ROOM_REAL_TIME_MESSAGE_UPDATE',
     'STOP_LIVE_ROOM_LIST',
     'SUPER_CHAT_MESSAGE_JPN',
+    'USER_TOAST_MSG',
     'WIDGET_BANNER',
 }
 """已打日志的未知cmd"""
@@ -58,6 +58,7 @@ 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
 
 
@@ -75,30 +76,36 @@ class BaseHandler(HandlerInterface):
             ['BaseHandler', ws_base.WebSocketClientBase, dict],
             Any
         ]]
-    ] = {
+    ]
+    """cmd -> 处理回调"""
+    _CMD_CALLBACK_DICT = {
         # 收到心跳包，这是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),
+        # 另一个上舰消息
+        'USER_TOAST_MSG_V2': _make_msg_callback('_on_user_toast_v2', web_models.UserToastV2Message),
         # 醒目留言
         '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),
+        # 进入房间、关注主播等互动消息
+        'INTERACT_WORD': _make_msg_callback('_on_interact_word', web_models.InteractWordMessage),
 
         #
         # 开放平台消息
         #
 
-        # 收到弹幕
+        # 弹幕
         '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),
@@ -108,8 +115,13 @@ class BaseHandler(HandlerInterface):
         ),
         # 点赞
         'LIVE_OPEN_PLATFORM_LIKE': _make_msg_callback('_on_open_live_like', open_models.LikeMessage),
+        # 进入房间
+        'LIVE_OPEN_PLATFORM_LIVE_ROOM_ENTER': _make_msg_callback('_on_open_live_enter_room', open_models.RoomEnterMessage),
+        # 开始直播
+        'LIVE_OPEN_PLATFORM_LIVE_START': _make_msg_callback('_on_open_live_start_live', open_models.LiveStartMessage),
+        # 结束直播
+        'LIVE_OPEN_PLATFORM_LIVE_END': _make_msg_callback('_on_open_live_end_live', open_models.LiveEndMessage),
     }
-    """cmd -> 处理回调"""
 
     def handle(self, client: ws_base.WebSocketClientBase, command: dict):
         cmd = command.get('cmd', '')
@@ -117,6 +129,8 @@ class BaseHandler(HandlerInterface):
         if pos != -1:
             cmd = cmd[:pos]
 
+        self._global_callback(client, command)
+
         if cmd not in self._CMD_CALLBACK_DICT:
             # 只有第一次遇到未知cmd时打日志
             if cmd not in logged_unknown_cmds:
@@ -128,72 +142,62 @@ class BaseHandler(HandlerInterface):
         if callback is not None:
             callback(self, client, command)
 
+    def _global_callback(self, client: ws_base.WebSocketClientBase, message: dict):
+        pass
+
     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_user_toast_v2(self, client: ws_base.WebSocketClientBase, message: web_models.UserToastV2Message):
+        """另一个上舰消息"""
 
     def _on_super_chat(self, client: ws_base.WebSocketClientBase, message: web_models.SuperChatMessage):
-        """
+        """醒目留言"""
-        醒目留言
-        """
 
-    def _on_super_chat_delete(
+    def _on_super_chat_delete(self, client: ws_base.WebSocketClientBase, message: web_models.SuperChatDeleteMessage):
-        self, client: ws_base.WebSocketClientBase, message: web_models.SuperChatDeleteMessage
+        """删除醒目留言"""
-    ):
+
-        """
+    def _on_interact_word(self, client: ws_base.WebSocketClientBase, message: web_models.InteractWordMessage):
-        删除醒目留言
+        """进入房间、关注主播等互动消息"""
-        """
 
     #
     # 开放平台消息
     #
 
     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(
+    def _on_open_live_super_chat(self, client: ws_base.WebSocketClientBase, message: open_models.SuperChatMessage):
-        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
+            self, client: ws_base.WebSocketClientBase, message: open_models.SuperChatDeleteMessage
     ):
-        """
+        """删除醒目留言"""
-        删除醒目留言
-        """
 
     def _on_open_live_like(self, client: ws_base.WebSocketClientBase, message: open_models.LikeMessage):
-        """
+        """点赞"""
-        点赞
+
-        """
+    def _on_open_live_enter_room(self, client: ws_base.WebSocketClientBase, message: open_models.RoomEnterMessage):
+        """进入房间"""
+
+    def _on_open_live_start_live(self, client: ws_base.WebSocketClientBase, message: open_models.LiveStartMessage):
+        """开始直播"""
+
+    def _on_open_live_end_live(self, client: ws_base.WebSocketClientBase, message: open_models.LiveEndMessage):
+        """结束直播"""

blivedm/models/open_live.py

@@ -23,8 +23,8 @@ class DanmakuMessage:
 
     uname: str = ''
     """用户昵称"""
-    uid: int = 0
+    open_id: str = ''
-    """用户UID"""
+    """用户唯一标识"""
     uface: str = ''
     """用户头像"""
     timestamp: int = 0
@@ -47,12 +47,20 @@ class DanmakuMessage:
     """表情包图片地址"""
     dm_type: int = 0
     """弹幕类型 0：普通弹幕 1：表情包弹幕"""
+    glory_level: int = 0
+    """直播荣耀等级"""
+    reply_open_id: str = ''
+    """被at用户唯一标识"""
+    reply_uname: str = ''
+    """被at的用户昵称"""
+    is_admin: int = 0
+    """发送弹幕的用户是否是房管，取值范围0或1，取值为1时是房管"""
 
     @classmethod
     def from_command(cls, data: dict):
         return cls(
             uname=data['uname'],
-            uid=data['uid'],
+            open_id=data['open_id'],
             uface=data['uface'],
             timestamp=data['timestamp'],
             room_id=data['room_id'],
@@ -64,6 +72,10 @@ class DanmakuMessage:
             fans_medal_level=data['fans_medal_level'],
             emoji_img_url=data['emoji_img_url'],
             dm_type=data['dm_type'],
+            glory_level=data['glory_level'],
+            reply_open_id=data['reply_open_id'],
+            reply_uname=data['reply_uname'],
+            is_admin=data['is_admin'],
         )
 
 
@@ -75,6 +87,8 @@ class AnchorInfo:
 
     uid: int = 0
     """收礼主播uid"""
+    open_id: str = ''
+    """收礼主播唯一标识"""
     uname: str = ''
     """收礼主播昵称"""
     uface: str = ''
@@ -84,6 +98,7 @@ class AnchorInfo:
     def from_dict(cls, data: dict):
         return cls(
             uid=data['uid'],
+            open_id=data['open_id'],
             uname=data['uname'],
             uface=data['uface'],
         )
@@ -122,8 +137,8 @@ class GiftMessage:
 
     room_id: int = 0
     """房间号"""
-    uid: int = 0
+    open_id: str = ''
-    """送礼用户UID"""
+    """用户唯一标识"""
     uname: str = ''
     """送礼用户昵称"""
     uface: str = ''
@@ -135,7 +150,20 @@ class GiftMessage:
     gift_num: int = 0
     """赠送道具数量"""
     price: int = 0
-    """（礼物单价）支付金额(1000 = 1元 = 10电池),盲盒:爆出道具的价值"""  # 这个B  站文档又不写清楚是单价还是总价
+    """
+    礼物爆出单价，(1000 = 1元 = 10电池),盲盒:爆出道具的价值
+
+    注意：
+
+    - 免费礼物这个字段也可能不是0，而是银瓜子数
+    - 有些打折礼物这里不是实际支付的价值，实际价值应该用 `r_price`
+    """
+    r_price: int = 0
+    """
+    实际价值(1000 = 1元 = 10电池),盲盒:爆出道具的价值
+
+    注意：免费礼物这个字段也可能不是0
+    """
     paid: bool = False
     """是否是付费道具"""
     fans_medal_level: int = 0
@@ -169,13 +197,14 @@ class GiftMessage:
 
         return cls(
             room_id=data['room_id'],
-            uid=data['uid'],
+            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'],
+            r_price=data['r_price'],
             paid=data['paid'],
             fans_medal_level=data['fans_medal_level'],
             fans_medal_name=data['fans_medal_name'],
@@ -196,8 +225,8 @@ class UserInfo:
     用户信息
     """
 
-    uid: int = 0
+    open_id: str = ''
-    """用户uid"""
+    """用户唯一标识"""
     uname: str = ''
     """用户昵称"""
     uface: str = ''
@@ -206,7 +235,7 @@ class UserInfo:
     @classmethod
     def from_dict(cls, data: dict):
         return cls(
-            uid=data['uid'],
+            open_id=data['open_id'],
             uname=data['uname'],
             uface=data['uface'],
         )
@@ -225,7 +254,9 @@ class GuardBuyMessage:
     guard_num: int = 0
     """大航海数量"""
     guard_unit: str = ''
-    """大航海单位"""
+    """大航海单位(正常单位为“月”，如为其他内容，无视`guard_num`以本字段内容为准，例如`*3天`)"""
+    price: int = 0
+    """大航海金瓜子"""
     fans_medal_level: int = 0
     """粉丝勋章等级"""
     fans_medal_name: str = ''
@@ -246,6 +277,7 @@ class GuardBuyMessage:
             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'],
@@ -263,8 +295,8 @@ class SuperChatMessage:
 
     room_id: int = 0
     """直播间id"""
-    uid: int = 0
+    open_id: str = ''
-    """购买用户UID"""
+    """用户唯一标识"""
     uname: str = ''
     """购买的用户昵称"""
     uface: str = ''
@@ -296,7 +328,7 @@ class SuperChatMessage:
     def from_command(cls, data: dict):
         return cls(
             room_id=data['room_id'],
-            uid=data['uid'],
+            open_id=data['open_id'],
             uname=data['uname'],
             uface=data['uface'],
             message_id=data['message_id'],
@@ -340,13 +372,16 @@ class LikeMessage:
     """
     点赞消息
 
-    请注意：用户端每分钟触发若干次的情况下只会推送一次该消息
+    请注意：
+
+    - 只有房间处于开播中，才会触发点赞事件
+    - 对单一用户最近2秒聚合发送一次点赞次数
     """
 
     uname: str = ''
     """用户昵称"""
-    uid: int = 0
+    open_id: str = ''
-    """用户UID"""
+    """用户唯一标识"""
     uface: str = ''
     """用户头像"""
     timestamp: int = 0
@@ -355,7 +390,7 @@ class LikeMessage:
     """发生的直播间"""
     like_text: str = ''
     """点赞文案(“xxx点赞了”)"""
-    like_count: int = 0  # 官方文档把这个字段名打错了，这个B文档真是一点都靠不住
+    like_count: int = 0
     """对单个用户最近2秒的点赞次数聚合"""
     fans_medal_wearing_status: bool = False
     """该房间粉丝勋章佩戴情况"""
@@ -371,7 +406,7 @@ class LikeMessage:
     def from_command(cls, data: dict):
         return cls(
             uname=data['uname'],
-            uid=data['uid'],
+            open_id=data['open_id'],
             uface=data['uface'],
             timestamp=data['timestamp'],
             room_id=data['room_id'],
@@ -382,3 +417,96 @@ class LikeMessage:
             fans_medal_level=data['fans_medal_level'],
             msg_id=data.get('msg_id', ''),  # 官方文档表格里没列出这个字段，但是参考JSON里面有
         )
+
+
+@dataclasses.dataclass
+class RoomEnterMessage:
+    """
+    进入房间消息
+    """
+
+    room_id: int = 0
+    """直播间id"""
+    uface: str = ''
+    """用户头像"""
+    uname: str = ''
+    """用户昵称"""
+    open_id: str = ''
+    """用户唯一标识"""
+    timestamp: int = 0
+    """发生的时间戳"""
+    msg_id: str = ''  # 官方文档表格里没列出这个字段，但是实际上有
+    """消息唯一id"""
+
+    @classmethod
+    def from_command(cls, data: dict):
+        return cls(
+            room_id=data['room_id'],
+            uface=data['uface'],
+            uname=data['uname'],
+            open_id=data['open_id'],
+            timestamp=data['timestamp'],
+            msg_id=data.get('msg_id', ''),  # 官方文档表格里没列出这个字段，但是实际上有
+        )
+
+
+@dataclasses.dataclass
+class LiveStartMessage:
+    """
+    开始直播消息
+    """
+
+    room_id: int = 0
+    """直播间id"""
+    open_id: str = ''
+    """用户唯一标识"""
+    timestamp: int = 0
+    """发生的时间戳"""
+    area_name: str = ''
+    """开播二级分区名"""
+    title: str = ''
+    """开播时刻，直播间的标题"""
+    msg_id: str = ''  # 官方文档表格里没列出这个字段，但是实际上有
+    """消息唯一id"""
+
+    @classmethod
+    def from_command(cls, data: dict):
+        return cls(
+            room_id=data['room_id'],
+            open_id=data['open_id'],
+            timestamp=data['timestamp'],
+            area_name=data['area_name'],
+            title=data['title'],
+            msg_id=data.get('msg_id', ''),  # 官方文档表格里没列出这个字段，但是实际上有
+        )
+
+
+@dataclasses.dataclass
+class LiveEndMessage:
+    """
+    结束直播消息
+    """
+
+    room_id: int = 0
+    """直播间id"""
+    open_id: str = ''
+    """用户唯一标识"""
+    timestamp: int = 0
+    """发生的时间戳"""
+    area_name: str = ''
+    """开播二级分区名"""
+    title: str = ''
+    """开播时刻，直播间的标题"""
+    msg_id: str = ''  # 官方文档表格里没列出这个字段，但是实际上有
+    """消息唯一id"""
+
+    @classmethod
+    def from_command(cls, data: dict):
+        return cls(
+            room_id=data['room_id'],
+            open_id=data['open_id'],
+            timestamp=data['timestamp'],
+            area_name=data['area_name'],
+            title=data['title'],
+            msg_id=data.get('msg_id', ''),  # 官方文档表格里没列出这个字段，但是实际上有
+        )

blivedm/models/web.py

@@ -67,6 +67,8 @@ class DanmakuMessage:
     """用户ID"""
     uname: str = ''
     """用户名"""
+    face: str = ''
+    """用户头像URL"""
     admin: int = 0
     """是否房管"""
     vip: int = 0
@@ -80,7 +82,7 @@ class DanmakuMessage:
     uname_color: str = ''
     """用户名颜色"""
 
-    medal_level: str = ''
+    medal_level: int = 0
     """勋章等级"""
     medal_name: str = ''
     """勋章名"""
@@ -108,23 +110,39 @@ class DanmakuMessage:
     privilege_type: int = 0
     """舰队类型，0非舰队，1总督，2提督，3舰长"""
 
+    wealth_level: int = 0
+    """荣耀等级"""
+
     @classmethod
     def from_command(cls, info: list):
+        mode_info = info[0][15]
+        try:
+            face = mode_info['user']['base']['face']
+        except (TypeError, KeyError):
+            face = ''
+
         if len(info[3]) != 0:
             medal_level = info[3][0]
             medal_name = info[3][1]
             runame = info[3][2]
-            room_id = info[3][3]
+            medal_room_id = info[3][3]
             mcolor = info[3][4]
             special_medal = info[3][5]
         else:
             medal_level = 0
             medal_name = ''
             runame = ''
-            room_id = 0
+            medal_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],
@@ -137,12 +155,13 @@ class DanmakuMessage:
             dm_type=info[0][12],
             emoticon_options=info[0][13],
             voice_config=info[0][14],
-            mode_info=info[0][15],
+            mode_info=mode_info,
 
             msg=info[1],
 
             uid=info[2][0],
             uname=info[2][1],
+            face=face,
             admin=info[2][2],
             vip=info[2][3],
             svip=info[2][4],
@@ -153,7 +172,7 @@ class DanmakuMessage:
             medal_level=medal_level,
             medal_name=medal_name,
             runame=runame,
-            medal_room_id=room_id,
+            medal_room_id=medal_room_id,
             mcolor=mcolor,
             special_medal=special_medal,
 
@@ -161,18 +180,23 @@ class DanmakuMessage:
             ulevel_color=info[4][2],
             ulevel_rank=info[4][3],
 
-            old_title=info[5][0],
+            old_title=old_title,
-            title=info[5][1],
+            title=title,
 
             privilege_type=info[7],
+
+            wealth_level=info[16][0],
         )
 
     @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
@@ -185,11 +209,14 @@ class DanmakuMessage:
     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
@@ -198,6 +225,30 @@ class DanmakuMessage:
         except (json.JSONDecodeError, TypeError):
             return {}
 
+    @property
+    def extra_dict(self) -> dict:
+        """
+        示例：
+
+        ```
+        {'send_from_me': False, 'mode': 0, 'color': 14893055, 'dm_type': 0, 'font_size': 25, 'player_mode': 4,
+        'show_player_type': 0, 'content': '确实', 'user_hash': '2904574201', 'emoticon_unique': '', 'bulge_display': 0,
+        'recommend_score': 5, 'main_state_dm_color': '', 'objective_state_dm_color': '', 'direction': 0,
+        'pk_direction': 0, 'quartet_direction': 0, 'anniversary_crowd': 0, 'yeah_space_type': '', 'yeah_space_url': '',
+        'jump_to_url': '', 'space_type': '', 'space_url': '', 'animation': {}, 'emots': None, 'is_audited': False,
+        'id_str': '6fa9959ab8feabcd1b337aa5066768334027', 'icon': None, 'show_reply': True, 'reply_mid': 0,
+        'reply_uname': '', 'reply_uname_color': '', 'reply_is_mystery': False, 'reply_type_enum': 0, 'hit_combo': 0,
+        'esports_jump_url': ''}
+        ```
+        """
+        try:
+            extra = self.mode_info['extra']
+            if isinstance(extra, dict):
+                return extra
+            return json.loads(extra)
+        except (KeyError, json.JSONDecodeError, TypeError):
+            return {}
+
 
 @dataclasses.dataclass
 class GiftMessage:
@@ -223,6 +274,8 @@ class GiftMessage:
     """礼物ID"""
     gift_type: int = 0
     """礼物类型（未知）"""
+    gift_img_basic: str = ''
+    """图标URL"""
     action: str = ''
     """目前遇到的有'喂食'、'赠送'"""
     price: int = 0
@@ -235,9 +288,29 @@ class GiftMessage:
     """总瓜子数"""
     tid: str = ''
     """可能是事务ID，有时和rnd相同"""
+    medal_level: int = 0
+    """勋章等级"""
+    medal_name: str = ''
+    """勋章名"""
+    medal_room_id: int = 0
+    """勋章房间ID，未登录时是0"""
+    medal_ruid: int = 0
+    """勋章主播ID"""
 
     @classmethod
     def from_command(cls, data: dict):
+        medal_info = data.get('medal_info', None)
+        if medal_info is not None:
+            medal_level = medal_info['medal_level']
+            medal_name = medal_info['medal_name']
+            medal_room_id = medal_info['anchor_roomid']
+            medal_ruid = medal_info['target_id']
+        else:
+            medal_level = 0
+            medal_name = ''
+            medal_room_id = 0
+            medal_ruid = 0
+
         return cls(
             gift_name=data['giftName'],
             num=data['num'],
@@ -248,12 +321,17 @@ class GiftMessage:
             timestamp=data['timestamp'],
             gift_id=data['giftId'],
             gift_type=data['giftType'],
+            gift_img_basic=data['gift_info']['img_basic'],
             action=data['action'],
             price=data['price'],
             rnd=data['rnd'],
             coin_type=data['coin_type'],
             total_coin=data['total_coin'],
             tid=data['tid'],
+            medal_level=medal_level,
+            medal_name=medal_name,
+            medal_room_id=medal_room_id,
+            medal_ruid=medal_ruid,
         )
 
 
@@ -269,7 +347,7 @@ class GuardBuyMessage:
     """用户名"""
     guard_level: int = 0
     """舰队等级，0非舰队，1总督，2提督，3舰长"""
-    num: int = 0
+    num: int = 0  # 可以理解为礼物数量？
     """数量"""
     price: int = 0
     """单价金瓜子数"""
@@ -297,6 +375,57 @@ class GuardBuyMessage:
         )
 
 
+@dataclasses.dataclass
+class UserToastV2Message:
+    """
+    另一个上舰消息，包含的数据更多
+    """
+
+    uid: int = 0
+    """用户ID"""
+    username: str = ''
+    """用户名"""
+    guard_level: int = 0
+    """舰队等级，0非舰队，1总督，2提督，3舰长"""
+    num: int = 0  # 可以理解为礼物数量？
+    """数量"""
+    price: int = 0
+    """单价金瓜子数"""
+    unit: str = ''
+    """单位，根据开放平台的文档，正常单位为“月”，如为其他内容，无视`guard_num`以本字段内容为准，例如`*3天`"""
+    gift_id: int = 0
+    """礼物ID"""
+    start_time: int = 0
+    """开始时间戳，和结束时间戳相同"""
+    end_time: int = 0
+    """结束时间戳，和开始时间戳相同"""
+    source: int = 0
+    """猜测0是自己买的，2是别人送的，这个只影响是否播动画"""
+    toast_msg: str = ''
+    """提示信息（"<%XXX%> 在主播XXX的直播间续费了舰长，今天是TA陪伴主播的第XXX天"）"""
+
+    @classmethod
+    def from_command(cls, data: dict):
+        sender_info = data['sender_uinfo']
+        guard_info = data['guard_info']
+        pay_info = data['pay_info']
+        gift_info = data['gift_info']
+        option = data['option']
+        return cls(
+            uid=sender_info['uid'],
+            username=sender_info['base']['name'],
+            guard_level=guard_info['guard_level'],
+            num=pay_info['num'],
+            price=pay_info['price'],
+            unit=pay_info['unit'],
+            gift_id=gift_info['gift_id'],
+            start_time=guard_info['start_time'],
+            end_time=guard_info['end_time'],
+            source=option['source'],
+            toast_msg=data['toast_msg'],
+        )
+
+
 @dataclasses.dataclass
 class SuperChatMessage:
     """
@@ -308,7 +437,7 @@ class SuperChatMessage:
     message: str = ''
     """消息"""
     message_trans: str = ''
-    """消息日文翻译（目前只出现在SUPER_CHAT_MESSAGE_JPN）"""
+    """消息日文翻译"""
     start_time: int = 0
     """开始时间戳"""
     end_time: int = 0
@@ -341,9 +470,29 @@ class SuperChatMessage:
     """背景图URL"""
     background_price_color: str = ''
     """背景价格颜色，'#rrggbb'"""
+    medal_level: int = 0
+    """勋章等级"""
+    medal_name: str = ''
+    """勋章名"""
+    medal_room_id: int = 0
+    """勋章房间ID"""
+    medal_ruid: int = 0
+    """勋章主播ID"""
 
     @classmethod
     def from_command(cls, data: dict):
+        medal_info = data.get('medal_info', None)
+        if medal_info is not None:
+            medal_level = medal_info['medal_level']
+            medal_name = medal_info['medal_name']
+            medal_room_id = medal_info['anchor_roomid']
+            medal_ruid = medal_info['target_id']
+        else:
+            medal_level = 0
+            medal_name = ''
+            medal_room_id = 0
+            medal_ruid = 0
+
         return cls(
             price=data['price'],
             message=data['message'],
@@ -364,6 +513,10 @@ class SuperChatMessage:
             background_icon=data['background_icon'],
             background_image=data['background_image'],
             background_price_color=data['background_price_color'],
+            medal_level=medal_level,
+            medal_name=medal_name,
+            medal_room_id=medal_room_id,
+            medal_ruid=medal_ruid,
         )
 
 
@@ -381,3 +534,33 @@ class SuperChatDeleteMessage:
         return cls(
             ids=data['ids'],
         )
+
+
+@dataclasses.dataclass
+class InteractWordMessage:
+    """
+    进入房间、关注主播等互动消息
+    """
+
+    uid: int = 0
+    """用户ID"""
+    username: str = ''
+    """用户名"""
+    face: str = ''
+    """用户头像URL"""
+    timestamp: int = 0
+    """时间戳"""
+    msg_type: int = 0
+    """`{1: '进入', 2: '关注了', 3: '分享了', 4: '特别关注了', 5: '互粉了', 6: '为主播点赞了'}`"""
+
+    @classmethod
+    def from_command(cls, data: dict):
+        user_info = data['uinfo']
+        user_base_info = user_info['base']
+        return cls(
+            uid=user_info['uid'],
+            username=user_base_info['name'],
+            face=user_base_info['face'],
+            timestamp=data['timestamp'],
+            msg_type=data['msg_type'],
+        )

db.py

@@ -0,0 +1,36 @@
+
+import mysql.connector
+import schedule
+
+
+class Room:
+    def __init__(self, room_id: str, liver_uid, liver_name):
+        self.room_id = room_id
+        self.liver_uid = liver_uid
+        self.liver_name = liver_name
+
+
+rooms = {}
+
+connection = mysql.connector.connect(
+    host="mysql",
+    database="live_log",
+    user="live_log",
+    password="!36z@Jd6LM@uxktq",
+)
+
+
+def flush_room_info():
+    with connection.cursor() as cursor:
+        cursor.execute(
+            "SELECT room_id, liver_uid, liver_name FROM liver_current_name")
+        for room_id, liver_uid, liver_name in cursor.fetchall():
+            rooms[room_id] = Room(room_id, liver_uid, liver_name)
+
+
+def get_room(room_id: str) -> Room:
+    return rooms.get(room_id, Room(room_id, 0, str(room_id)))
+
+
+flush_room_info()
+schedule.every(1).days.do(flush_room_info)

live_status_ntf.py

@@ -0,0 +1,222 @@
+import os
+import time
+import requests
+import schedule
+from watchdog.observers import Observer
+from watchdog.events import FileSystemEventHandler
+import json
+import db
+import re
+
+path = "logs"
+room_id_re = re.compile(rf"{path}/(\d+)\.jsonl$")
+
+notify_qq_group = {
+    "12571885": lambda room, is_online: notify_group(room, ["831867573", "138981147"], is_online),
+    "147482": lambda room, is_online: notify_group(room, ["175545447"], is_online),
+    "20571": lambda room, is_online: notify_group(room, ["891117762"], is_online),
+}
+
+
+class Room:
+    def __init__(self, room_id: str) -> None:
+        self.room_id = room_id
+        self.db_room = db.get_room(room_id)
+        self.position = 0
+        self.state = 0
+        self.state_changed = False
+
+    def reset(self):
+        self.position = 0
+        self.state = 0
+        self.state_changed = False
+
+    def update_state(self, state):
+        self.state = state
+        self.state_changed = True
+
+
+_rooms: dict[str, Room] = {}
+
+
+def notify_group(room: Room, groups: list[str], is_online: bool):
+    status = "开锅了" if is_online else "下锅了"
+    msg = "\n".join([
+        f"{room.db_room.liver_name} {status}！",
+        f"https://live.bilibili.com/{room.room_id}"
+    ])
+    for qq_group_id in groups:
+        send_qq_group_msg(qq_group_id, msg)
+
+
+def notify_group_and_mc(room: Room, groups: list[str], is_online: bool):
+    notify_group(room, groups, is_online)
+    try:
+        msg = "\n".join([
+            f"{room.db_room.liver_name} {'开锅了' if is_online else '下锅了'}！",
+            f"https://live.bilibili.com/{room.room_id}"
+        ])
+        requests.post("http://mc1:5000/msg", json={"msg": msg})
+    except BaseException as e:
+        print(f"notify mc error: {e}")
+
+
+def get_room(src_path):
+    src_path_str = str(src_path)
+    room_id = room_id_re.findall(src_path_str)[0]
+    room = _rooms.get(room_id, None)
+    if room is None:
+        room = Room(room_id)
+        _rooms[room_id] = room
+
+    return room
+
+
+class MyHandler(FileSystemEventHandler):
+    def on_modified(self, event):
+        if event.is_directory:
+            return
+
+        src_path_str = str(event.src_path)
+        if not room_id_re.search(src_path_str):
+            return
+
+        room: Room = get_room(src_path_str)
+
+        try:
+            with open(event.src_path, "r", encoding="utf-8") as f:
+                file_size = os.path.getsize(event.src_path)
+                if room.position > file_size:
+                    room.reset()
+                f.seek(room.position)
+
+                for line in f:
+                    try:
+                        data = json.loads(line)
+                    except json.decoder.JSONDecodeError:
+                        continue
+                    self._handle_data(room, data)
+
+                room.position = os.path.getsize(event.src_path)
+        except UnicodeDecodeError:
+            pass
+
+        self._check_room_state_change(room)
+
+    def on_created(self, event):
+        if event.is_directory:
+            return
+
+        src_path_str = str(event.src_path)
+        if not room_id_re.search(src_path_str):
+            return
+
+        room: Room = get_room(src_path_str)
+
+        try:
+            with open(event.src_path, "r", encoding="utf-8") as f:
+                for line in f:
+                    try:
+                        data = json.loads(line)
+                    except json.decoder.JSONDecodeError:
+                        continue
+                    self._handle_data(room, data)
+
+            room.position = os.path.getsize(event.src_path)
+        except UnicodeDecodeError:
+            pass
+
+        self._check_room_state_change(room)
+
+    def on_deleted(self, event):
+        if event.is_directory:
+            return
+
+        src_path_str = str(event.src_path)
+        if not room_id_re.search(src_path_str):
+            return
+
+        room_id = room_id_re.findall(src_path_str)[0]
+        room = _rooms.get(room_id, Room(room_id))
+        room.reset()
+
+    def _handle_data(self, room: Room, data):
+        if data["cmd"] == "LIVE":
+            if room.state == 1:
+                return
+
+            room.update_state(1)
+
+        elif data["cmd"] == "PREPARING":
+            if room.state == 2:
+                return
+
+            room.update_state(2)
+
+    def _check_room_state_change(self, room: Room):
+        if not room.state_changed:
+            return
+
+        room.state_changed = False
+
+        if room.state == 1:
+            msg = "\n".join([
+                f"{room.db_room.liver_name} 开锅了！",
+                f"https://live.bilibili.com/{room.room_id}"
+            ])
+            try:
+                requests.post("http://turntf:18846/notify", json={"msg": msg})
+            except BaseException as e:
+                print(f"notify turntf error: {e}")
+
+            notify_func = notify_qq_group.get(room.room_id)
+            if notify_func:
+                notify_func(room, True)
+
+        elif room.state == 2:
+            msg = "\n".join([
+                f"{room.db_room.liver_name} 下锅了！",
+                f"https://live.bilibili.com/{room.room_id}"
+            ])
+            try:
+                requests.post("http://turntf:18846/notify", json={"msg": msg})
+            except BaseException as e:
+                print(f"notify turntf error: {e}")
+
+            notify_func = notify_qq_group.get(room.room_id)
+            if notify_func:
+                notify_func(room, False)
+
+
+def send_qq_group_msg(group_id: str, msg: str):
+    try:
+        requests.post("http://napcat:3000/send_group_msg", json={
+            "group_id": group_id,
+            "message": msg
+        })
+    except BaseException as e:
+        print(f"send qq group msg error: {e}")
+
+
+def main():
+    for f in os.listdir(path):
+        f = path+"/"+str(f)
+        if not room_id_re.search(f):
+            continue
+        get_room(f).position = os.path.getsize(f)
+
+    event_handler = MyHandler()
+    observer = Observer()
+    observer.schedule(event_handler, path, recursive=True)
+    observer.start()
+    try:
+        while True:
+            schedule.run_pending()
+            time.sleep(1)
+    except KeyboardInterrupt:
+        observer.stop()
+    observer.join()
+
+
+if __name__ == "__main__":
+    main()

main.py

@@ -0,0 +1,190 @@
+import asyncio
+import http.cookies
+import json
+import logging
+import os
+import signal
+import threading
+import time
+from typing import *
+
+import aiohttp
+import redis_util as redis
+
+import blivedm
+from blivedm.clients import ws_base
+
+
+def room_from_config():
+    return conf.get("room_ids", [])
+
+
+# noinspection PyUnusedLocal
+def handle_sighup(signum, frame):
+    global log_files
+    global room_status_log
+
+    for room_id in log_files:
+        try:
+            log_files[room_id].close()
+        except Exception as e:
+            logger.exception(f"close log_files[{room_id}] failed with {e}")
+    try:
+        room_status_log.close()
+    except Exception as e:
+        logger.exception(f"close room_status_log failed with {e}")
+
+    log_files = {}
+    room_status_log = open("logs/room_status.jsonl", "a", encoding="utf-8")
+
+
+async def on_room_changed(new_room_ids):
+    global TEST_ROOM_IDS
+
+    TEST_ROOM_IDS = list(set(new_room_ids + room_from_config()))
+
+    for removed in clients.keys() - TEST_ROOM_IDS:
+        print(f"room {removed} removed")
+        await clients[removed].stop_and_close()
+        del clients[removed]
+
+        if removed in log_files:
+            log_files[removed].close()
+            del log_files[removed]
+
+    for added in TEST_ROOM_IDS - clients.keys():
+        await connect_room_safety(added)
+
+
+async def connect_room_safety(room_id: str):
+    client = None
+    try:
+        client = await connect_room(room_id)
+    except Exception as e:
+        if client is not None:
+            asyncio.create_task(client.stop_and_close())
+
+
+async def main():
+    global TEST_ROOM_IDS
+
+    flush_files()
+
+    redis.init(conf.get("redis", None))
+    redis.on_room_changed = on_room_changed
+    TEST_ROOM_IDS = list(set(TEST_ROOM_IDS + redis.get_room_list()))
+
+    init_session()
+    try:
+        for room_id in TEST_ROOM_IDS:
+            client = None
+            try:
+                client = await connect_room(room_id)
+            except Exception as e:
+                if client is not None:
+                    await client.stop_and_close()
+
+                raise e
+
+        while True:
+            await asyncio.gather(*(
+                clients[room_id].join() for room_id in clients
+            ))
+            await asyncio.sleep(1)
+    finally:
+        for room_id in clients:
+            await clients[room_id].stop_and_close()
+        await session.close()
+
+
+async def connect_room(room_id):
+    logger.info("connect to room %s", room_id)
+    client = blivedm.BLiveClient(room_id, session=session)
+    client.set_handler(jsonl_logger)
+    await client.init_room()
+    client._need_init_room = False
+    client.start()
+    clients[room_id] = client
+    logger.info("connect to room %s success", room_id)
+
+    try:
+        room_status_log.write(json.dumps(
+            {
+                "room_id": room_id,
+                # "live_status": client.live_status,
+                # "live_start_time": client.live_start_time,
+            },
+            separators=(",", ":"),
+            ensure_ascii=False,
+        ))
+        room_status_log.write("\n")
+    except Exception as e:
+        logger.exception(f"write room_status_log failed with {e}")
+    return client
+
+
+def flush_files():
+    try:
+        room_status_log.flush()
+        for room_id in log_files:
+            log_files[room_id].flush()
+    except Exception as e:
+        logger.exception(f"flush_files failed with {e}")
+
+    timer = threading.Timer(1, flush_files)
+    timer.daemon = True
+    timer.start()
+
+
+def init_session():
+    global session, cookies
+    session = aiohttp.ClientSession()
+    session.cookie_jar.update_cookies(cookies)
+
+
+class JsonlLogger(blivedm.BaseHandler):
+    def _global_callback(self, client: ws_base.WebSocketClientBase, message: dict):
+        if client.room_id not in log_files:
+            log_files[client.room_id] = open(
+                f"logs/{client.room_id}.jsonl", "a", encoding="utf-8")
+
+        if "cmd" in message and message["cmd"] == "PREPARING":
+            message["timestamp"] = int(time.time())
+
+        log_files[client.room_id].write(
+            f"{json.dumps(message, separators=(',', ':'), ensure_ascii=False)}\n")
+
+
+# set log level to INFO
+logging.basicConfig(level=logging.INFO)
+
+try:
+    with open("config.json") as conf_f:
+        conf = json.load(conf_f)
+except FileNotFoundError:
+    conf = {}
+redis_conf: dict | None = conf.get("redis", None)
+
+os.makedirs("logs", exist_ok=True)
+room_status_log = open("logs/room_status.jsonl", "a", encoding="utf-8")
+
+logger = logging.getLogger()
+
+clients: dict[str, blivedm.BLiveClient] = {}
+log_files = {}
+
+with open("cookie.txt", "r", encoding="utf-8") as f:
+    cookies = http.cookies.SimpleCookie()
+    cookies.load(f.read().strip())
+    cookies["SESSDATA"]["domain"] = "bilibili.com"
+
+session: Optional[aiohttp.ClientSession] = None
+
+TEST_ROOM_IDS: list[int] = room_from_config()
+
+signal.signal(signal.SIGHUP, handle_sighup)
+
+jsonl_logger = JsonlLogger()
+
+if __name__ == "__main__":
+    asyncio.run(main())

open_live_sample.py

@@ -71,6 +71,15 @@ class MyHandler(blivedm.BaseHandler):
     def _on_open_live_like(self, client: blivedm.OpenLiveClient, message: open_models.LikeMessage):
         print(f'[{message.room_id}] {message.uname} 点赞')
 
+    def _on_open_live_enter_room(self, client: blivedm.OpenLiveClient, message: open_models.RoomEnterMessage):
+        print(f'[{message.room_id}] {message.uname} 进入房间')
+
+    def _on_open_live_start_live(self, client: blivedm.OpenLiveClient, message: open_models.LiveStartMessage):
+        print(f'[{message.room_id}] 开始直播')
+
+    def _on_open_live_end_live(self, client: blivedm.OpenLiveClient, message: open_models.LiveEndMessage):
+        print(f'[{message.room_id}] 结束直播')
+
 
 if __name__ == '__main__':
     asyncio.run(main())

pyproject.toml

@@ -0,0 +1,46 @@
+[build-system]
+requires = ["pdm-backend"]
+build-backend = "pdm.backend"
+
+[project]
+name = "blivedm"
+dynamic = ["version"]
+description = "Python获取bilibili直播弹幕的库，使用WebSocket协议"
+readme = "README.md"
+keywords = ["bilibili", "bilibili-live", "danmaku"]
+requires-python = ">=3.8"
+authors = [
+    {name = "xfgryujk", email = "xfgryujk@126.com"},
+]
+license = {file = "LICENSE"}
+classifiers = [
+    "Development Status :: 5 - Production/Stable",
+    "Framework :: AsyncIO",
+    "Intended Audience :: Developers",
+    "License :: OSI Approved :: MIT License",
+    "Operating System :: OS Independent",
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3 :: Only",
+    "Programming Language :: Python :: 3.8",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
+    "Topic :: Internet :: WWW/HTTP",
+    "Topic :: Software Development :: Libraries",
+]
+dependencies = [
+    "aiohttp~=3.9.0",
+    "Brotli~=1.1.0",
+    "yarl~=1.9.3",
+]
+
+[project.urls]
+Homepage = "https://github.com/xfgryujk/blivedm"
+Repository = "https://github.com/xfgryujk/blivedm"
+Issues = "https://github.com/xfgryujk/blivedm/issues"
+
+[tool.pdm]
+version = {source = "file", path = "blivedm/__init__.py"}
+distribution = true

redis_util.py

@@ -0,0 +1,40 @@
+import asyncio
+import threading
+from typing import Callable, Awaitable
+
+import redis
+
+room_set_key = "bilibili:live:danmu:room_set"
+r: redis.Redis | None = None
+on_room_changed: Callable[[list[int]], Awaitable[None]] | None = None
+
+
+def init(redis_conf: dict | None) -> None:
+    global r
+
+    if redis_conf is None:
+        return
+
+    r = redis.Redis(
+        host=redis_conf.get("host", "127.0.0.1"),
+        port=redis_conf.get("port", 6379),
+        db=redis_conf.get("port", 0),
+    )
+
+    asyncio.create_task(_subscribe_redis_async())
+
+
+def get_room_list() -> list[int]:
+    if r is None:
+        return []
+
+    return [int(room_id) for room_id in r.smembers(room_set_key)]
+
+
+async def _subscribe_redis_async():
+    p = r.pubsub()
+    p.subscribe(f"__keyspace@0__:{room_set_key}")
+    while p.subscribed:
+        msg = await asyncio.to_thread(p.get_message, ignore_subscribe_messages=True, timeout=1)
+        if msg and on_room_changed is not None:
+            await on_room_changed(get_room_list())

requirements.txt

@@ -1,3 +1,10 @@
-aiohttp~=3.9.0
+aiohttp
-Brotli~=1.1.0
+Brotli
-yarl~=1.9.3
+yarl
+redis
+PyYAML
+watchdog
+requests
+mysql-connector-python
+schedule
+requests

sample.py

@@ -18,7 +18,7 @@ TEST_ROOM_IDS = [
     23105590,
 ]
 
-# 这里填一个已登录账号的cookie。不填cookie也可以连接，但是收到弹幕的用户名会打码，UID会变成0
+# 这里填一个已登录账号的cookie的SESSDATA字段的值。不填也可以连接，但是收到弹幕的用户名会打码，UID会变成0
 SESSDATA = ''
 
 session: Optional[aiohttp.ClientSession] = None
@@ -87,11 +87,10 @@ class MyHandler(blivedm.BaseHandler):
     # # 演示如何添加自定义回调
     # _CMD_CALLBACK_DICT = blivedm.BaseHandler._CMD_CALLBACK_DICT.copy()
     #
-    # # 入场消息回调
+    # # 看过数消息回调
-    # def __interact_word_callback(self, client: blivedm.BLiveClient, command: dict):
+    # def __watched_change_callback(self, client: blivedm.BLiveClient, command: dict):
-    #     print(f"[{client.room_id}] INTERACT_WORD: self_type={type(self).__name__}, room_id={client.room_id},"
+    #     print(f'[{client.room_id}] WATCHED_CHANGE: {command}')
-    #           f" uname={command['data']['uname']}")
+    # _CMD_CALLBACK_DICT['WATCHED_CHANGE'] = __watched_change_callback  # noqa
-    # _CMD_CALLBACK_DICT['INTERACT_WORD'] = __interact_word_callback  # noqa
 
     def _on_heartbeat(self, client: blivedm.BLiveClient, message: web_models.HeartbeatMessage):
         print(f'[{client.room_id}] 心跳')
@@ -103,12 +102,19 @@ class MyHandler(blivedm.BaseHandler):
         print(f'[{client.room_id}] {message.uname} 赠送{message.gift_name}x{message.num}'
               f' （{message.coin_type}瓜子x{message.total_coin}）')
 
-    def _on_buy_guard(self, client: blivedm.BLiveClient, message: web_models.GuardBuyMessage):
+    # def _on_buy_guard(self, client: blivedm.BLiveClient, message: web_models.GuardBuyMessage):
-        print(f'[{client.room_id}] {message.username} 购买{message.gift_name}')
+    #     print(f'[{client.room_id}] {message.username} 上舰，guard_level={message.guard_level}')
+
+    def _on_user_toast_v2(self, client: blivedm.BLiveClient, message: web_models.UserToastV2Message):
+        print(f'[{client.room_id}] {message.username} 上舰，guard_level={message.guard_level}')
 
     def _on_super_chat(self, client: blivedm.BLiveClient, message: web_models.SuperChatMessage):
         print(f'[{client.room_id}] 醒目留言 ¥{message.price} {message.uname}：{message.message}')
 
+    # def _on_interact_word(self, client: blivedm.BLiveClient, message: web_models.InteractWordMessage):
+    #     if message.msg_type == 1:
+    #         print(f'[{client.room_id}] {message.username} 进入房间')
+
 
 if __name__ == '__main__':
     asyncio.run(main())