update blive dynamic

src/plugins/blive/config.py

@@ -4,8 +4,9 @@
 
 class Config(BaseModel, extra=Extra.ignore):
     blive_live_interval: int = 10
-    blive_dynamic_interval: int = 10
+    blive_dynamic_interval: int = 30
     blrec_address: str = "http://your_address"
+    bilibili_cookie: str = ""
 
 
 blive_config = Config.parse_obj(get_driver().config.dict())

src/plugins/blive/pusher/dynamic_pusher.py

@@ -1,18 +1,14 @@
+import traceback
 from typing import Dict
 
-from bilireq.exceptions import GrpcError
-from bilireq.grpc.dynamic import grpc_get_user_dynamics
-from bilireq.grpc.protos.bilibili.app.dynamic.v2.dynamic_pb2 import DynamicType
-from grpc import StatusCode
-from grpc.aio import AioRpcError
 from nonebot.log import logger
 from nonebot_plugin_apscheduler import scheduler
 from nonebot_plugin_saa import Image, MessageFactory
 
 from ..config import blive_config
 from ..database.db import get_dynamic_targets, get_dynamic_uids, update_user
 from ..models import BiliUser
-from ..utils import get_dynamic_screenshot
+from ..utils import get_dynamic_screenshot, get_user_dynamics
 
 dynamic_offset: Dict[str, int] = {}
 updated_uids = set()
@@ -38,74 +34,55 @@ async def _():
     uid = next_uid(uids)
 
     try:
-        dynamics = (await grpc_get_user_dynamics(int(uid), timeout=30)).list
-    except AioRpcError as e:
-        if e.code() == StatusCode.DEADLINE_EXCEEDED:
-            logger.warning(f"爬取动态超时，将在下个轮询中重试：{e.code()} {e.details()}")
-        else:
-            logger.warning(f"爬取动态失败：{e.code()} {e.details()}")
+        dynamics = (await get_user_dynamics(int(uid)))["items"]
+    except:
+        logger.warning(f"爬取动态失败：{uid}\n{traceback.format_exc()}")
         return
-    except GrpcError as e:
-        logger.warning(f"爬取动态失败：{e.code} {e.msg}")
+
+    if not dynamics:
         return
 
     if uid not in dynamic_offset:
-        if not dynamics:
-            dynamic_offset[uid] = 0
-        elif len(dynamics) == 1:
-            dynamic_offset[uid] = int(dynamics[0].extend.dyn_id_str)
-        else:
-            dynamic_offset[uid] = max(
-                int(dynamics[0].extend.dyn_id_str), int(dynamics[1].extend.dyn_id_str)
-            )
+        dynamic_offset[uid] = max([int(dynamic["id_str"]) for dynamic in dynamics])
         return
 
+    dynamics = [
+        dynamic for dynamic in dynamics if int(dynamic["id_str"]) > dynamic_offset[uid]
+    ]
     if not dynamics:
         return
 
-    name = dynamics[0].modules[0].module_author.author.name
-
-    dynamic = None
-    for dynamic in sorted(dynamics, key=lambda x: int(x.extend.dyn_id_str)):
-        dynamic_id = int(dynamic.extend.dyn_id_str)
-        if dynamic_id > dynamic_offset[uid]:
-            logger.info(f"检测到新动态：{dynamic_id}")
-            dynamic_offset[uid] = dynamic_id
-
-            if dynamic.card_type in [
-                DynamicType.live_rcmd,
-                DynamicType.live,
-                DynamicType.ad,
-                DynamicType.banner,
-            ]:
-                logger.info(f"无需推送的动态 {dynamic.card_type}，已跳过：{dynamic_id}")
-                return
-
-            if not (image := await get_dynamic_screenshot(dynamic_id)):
-                logger.warning(f"获取动态截图失败：{dynamic_id}")
-                return
-
-            type_msg = {
-                0: "发布了新动态",
-                DynamicType.forward: "转发了一条动态",
-                DynamicType.word: "发布了新文字动态",
-                DynamicType.draw: "发布了新图文动态",
-                DynamicType.av: "发布了新投稿",
-                DynamicType.article: "发布了新专栏",
-                DynamicType.music: "发布了新音频",
-            }
-            msg = MessageFactory([])
-            msg.append(f"{name} {type_msg.get(dynamic.card_type, type_msg[0])}：\n")
-            msg.append(Image(image))
-            msg.append(f"https://t.bilibili.com/{dynamic_id}")
-
-            targets = await get_dynamic_targets(uid)
-            for target in targets:
-                try:
-                    await msg.send_to(target)
-                except Exception as e:
-                    logger.warning(f"Error in sending dynamic message to {target}: {e}")
-                    continue
-
-    if dynamic:
-        await update_user(BiliUser(uid=uid, name=name))
+    name = dynamics[0]["modules"]["module_author"]["name"]
+    for dynamic in sorted(dynamics, key=lambda x: int(x["id_str"])):
+        dynamic_id = int(dynamic["id_str"])
+        dyn_type = dynamic["type"]
+        logger.info(f"检测到新动态：{dynamic_id}")
+        dynamic_offset[uid] = dynamic_id
+
+        if not (image := await get_dynamic_screenshot(dynamic_id)):
+            logger.warning(f"获取动态截图失败：{dynamic_id}")
+            return
+
+        type_msg = {
+            "DYNAMIC_TYPE_FORWARD": "转发了一条动态",
+            "DYNAMIC_TYPE_WORD": "发布了新文字动态",
+            "DYNAMIC_TYPE_DRAW": "发布了新图文动态",
+            "DYNAMIC_TYPE_AV": "发布了新投稿",
+            "DYNAMIC_TYPE_ARTICLE": "发布了新专栏",
+            "DYNAMIC_TYPE_MUSIC": "发布了新音频",
+        }
+        text = type_msg.get(dyn_type, "发布了新动态")
+        msg = MessageFactory([])
+        msg.append(f"{name} {text}：\n")
+        msg.append(Image(image))
+        msg.append(f"https://t.bilibili.com/{dynamic_id}")
+
+        targets = await get_dynamic_targets(uid)
+        for target in targets:
+            try:
+                await msg.send_to(target)
+            except Exception as e:
+                logger.warning(f"Error in sending dynamic message to {target}: {e}")
+                continue
+
+    await update_user(BiliUser(uid=uid, name=name))

src/plugins/blive/utils.py

@@ -1,18 +1,29 @@
 import traceback
 from typing import Optional
 
+import httpx
 from bilireq.user import get_user_info
-from bilireq.utils import get
+from bilireq.utils import DEFAULT_HEADERS, get
 from nonebot.log import logger
 from nonebot_plugin_htmlrender import get_new_page
 
+from .config import blive_config
 from .models import BiliUser
 
+USER_AGENT = (
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+    "(KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36 Edg/113.0.1774.35"
+)
+
 
 async def search_user(keyword: str):
     url = "https://api.bilibili.com/x/web-interface/search/type"
     data = {"keyword": keyword, "search_type": "bili_user"}
-    return await get(url, params=data)
+    headers = {"user-agent": USER_AGENT, "Cookie": blive_config.bilibili_cookie}
+    async with httpx.AsyncClient(timeout=10) as client:
+        await client.get("https://www.bilibili.com", headers=headers)
+        resp = await client.get(url, params=data)
+        return resp.json()["data"]
 
 
 async def get_uset_info_by_uid(uid: str) -> Optional[BiliUser]:
@@ -37,13 +48,26 @@ async def get_user_info_by_name(name: str) -> Optional[BiliUser]:
                 )
 
 
+async def get_user_dynamics(uid: int):
+    url = "https://api.bilibili.com/x/polymer/web-dynamic/v1/feed/space"
+    data = {"host_mid": uid}
+    headers = {
+        **DEFAULT_HEADERS,
+        **{
+            "Origin": "https://space.bilibili.com",
+            "Referer": f"https://space.bilibili.com/{uid}/dynamic",
+            "Cookie": blive_config.bilibili_cookie,
+        },
+    }
+    return await get(url, params=data, headers=headers, timeout=20)
+
+
 async def get_dynamic_screenshot(dynamic_id: int) -> Optional[bytes]:
     url = f"https://t.bilibili.com/{dynamic_id}"
     try:
         async with get_new_page(
             viewport={"width": 2000, "height": 1000},
-            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
-            "(KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36 Edg/114.0.1823.82",
+            user_agent=USER_AGENT,
             device_scale_factor=3,
         ) as page:
             await page.goto(url, wait_until="networkidle")