update

# Conflicts:
#	blivedm/clients/web.py

blivedm/__init__.py

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

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,6 +20,7 @@ __all__ = (
 logger = logging.getLogger('blivedm')
 
 UID_INIT_URL = 'https://api.bilibili.com/x/web-interface/nav'
+WBI_INIT_URL = UID_INIT_URL
 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'
@@ -23,6 +28,127 @@ 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):
     """
@@ -35,14 +161,15 @@ class BLiveClient(ws_base.WebSocketClientBase):
     """
 
     def __init__(
-            self,
-            room_id: int,
-            *,
-            uid: Optional[int] = None,
-            session: Optional[aiohttp.ClientSession] = None,
-            heartbeat_interval=30,
+        self,
+        room_id: int,
+        *,
+        uid: Optional[int] = None,
+        session: Optional[aiohttp.ClientSession] = None,
+        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"""
@@ -50,8 +177,6 @@ class BLiveClient(ws_base.WebSocketClientBase):
 
         # 在调用init_room后初始化的字段
         self._room_owner_uid: Optional[int] = None
-        self.live_status: Optional[int] = None
-        self.live_start_time: Optional[int] = None
         """主播用户ID"""
         self._host_server_list: Optional[List[dict]] = None
         """
@@ -122,8 +247,8 @@ class BLiveClient(ws_base.WebSocketClientBase):
 
         try:
             async with self._session.get(
-                    UID_INIT_URL,
-                    headers={'User-Agent': utils.USER_AGENT},
+                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,
@@ -160,8 +285,8 @@ class BLiveClient(ws_base.WebSocketClientBase):
     async def _init_buvid(self):
         try:
             async with self._session.get(
-                    BUVID_INIT_URL,
-                    headers={'User-Agent': utils.USER_AGENT},
+                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',
@@ -173,11 +298,11 @@ class BLiveClient(ws_base.WebSocketClientBase):
     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
-                    },
+                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,
@@ -201,14 +326,21 @@ class BLiveClient(ws_base.WebSocketClientBase):
         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={
-                        'id': self._room_id,
-                        'type': 0
-                    },
+                DANMAKU_SERVER_CONF_URL,
+                headers={'User-Agent': utils.USER_AGENT},
+                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,
@@ -216,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']):

main.py

@@ -111,8 +111,8 @@ async def connect_room(room_id):
         room_status_log.write(json.dumps(
             {
                 "room_id": room_id,
-                "live_status": client.live_status,
-                "live_start_time": client.live_start_time,
+                # "live_status": client.live_status,
+                # "live_start_time": client.live_start_time,
             },
             separators=(",", ":"),
             ensure_ascii=False,

requirements.txt

@@ -1,9 +1,9 @@
-aiohttp~=3.9.0
-Brotli~=1.1.0
-yarl~=1.9.3
-redis~=5.0.2
-PyYAML~=6.0.1
-watchdog~=5.0.3
-requests~=2.32.3
-mysql-connector-python~=9.0.0
-schedule~=1.2.2
+aiohttp
+Brotli
+yarl
+redis
+PyYAML
+watchdog
+requests
+mysql
+schedule