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']):

db.py

@@ -2,8 +2,9 @@
 import mysql.connector
 import schedule
 
+
 class Room:
-    def __init__(self, room_id, liver_uid, liver_name):
+    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
@@ -27,7 +28,7 @@ def flush_room_info():
             rooms[room_id] = Room(room_id, liver_uid, liver_name)
 
 
-def get_room(room_id) -> Room:
+def get_room(room_id: str) -> Room:
     return rooms.get(room_id, Room(room_id, 0, str(room_id)))
 
 

live_status_ntf.py

@@ -9,18 +9,22 @@ import db
 import re
 
 path = "logs"
-room_id_re = re.compile(rf"{path}/(\d+).jsonl$")
+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:
-    position = 0
-    # 0 unkonwn, 1 live, 2 pending
-    state = 0
-    state_changed = False
-
-    def __init__(self, room_id) -> None:
+    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
@@ -35,8 +39,31 @@ class Room:
 _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):
-    room_id = room_id_re.findall(src_path)[0]
+    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)
@@ -50,25 +77,27 @@ class MyHandler(FileSystemEventHandler):
         if event.is_directory:
             return
 
-        if not room_id_re.match(event.src_path):
+        src_path_str = str(event.src_path)
+        if not room_id_re.search(src_path_str):
             return
 
-        room: Room = get_room(event.src_path)
+        room: Room = get_room(src_path_str)
 
         try:
             with open(event.src_path, "r", encoding="utf-8") as f:
-                if room.position > os.path.getsize(event.src_path):
+                file_size = os.path.getsize(event.src_path)
+                if room.position > file_size:
                     room.reset()
                 f.seek(room.position)
 
-                room.position = os.path.getsize(event.src_path)
-
                 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
 
@@ -78,11 +107,11 @@ class MyHandler(FileSystemEventHandler):
         if event.is_directory:
             return
 
-        if not room_id_re.match(event.src_path):
+        src_path_str = str(event.src_path)
+        if not room_id_re.search(src_path_str):
             return
 
-        room: Room = get_room(event.src_path)
-        room.position = os.path.getsize(event.src_path)
+        room: Room = get_room(src_path_str)
 
         try:
             with open(event.src_path, "r", encoding="utf-8") as f:
@@ -91,7 +120,9 @@ class MyHandler(FileSystemEventHandler):
                         data = json.loads(line)
                     except json.decoder.JSONDecodeError:
                         continue
-                    self._handle_data(data)
+                    self._handle_data(room, data)
+
+            room.position = os.path.getsize(event.src_path)
         except UnicodeDecodeError:
             pass
 
@@ -101,10 +132,12 @@ class MyHandler(FileSystemEventHandler):
         if event.is_directory:
             return
 
-        if not room_id_re.match(event.src_path):
+        src_path_str = str(event.src_path)
+        if not room_id_re.search(src_path_str):
             return
 
-        room = _rooms.get(event.src_path, Room())
+        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):
@@ -131,20 +164,44 @@ class MyHandler(FileSystemEventHandler):
                 f"{room.db_room.liver_name} 开锅了！",
                 f"https://live.bilibili.com/{room.room_id}"
             ])
-            requests.post("http://turntf:18846/notify", json={"msg": msg})
+            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}"
             ])
-            requests.post("http://turntf:18846/notify", json={"msg": msg})
+            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.match(f):
+        if not room_id_re.search(f):
             continue
         get_room(f).position = os.path.getsize(f)
 

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,10 @@
-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-connector-python
+schedule
+requests