✨ 添加兼容 RSSHub 的动态 RSS 订阅地址以获取 UP 动态的功能 (#233)

README.md

@@ -175,6 +175,7 @@ bilichat_openai_proxy = "http://127.0.0.1:7890/"
 |      bilichat_text_fonts      |    str    | default  |          可供自定义的字体，仅作用于 dynamicrender 绘图           |
 |     bilichat_emoji_fonts      |    str    | default  |          可供自定义的字体，仅作用于 dynamicrender 绘图           |
 |      bilichat_webui_path      |    str    | bilichat |               WebUI 的路径，设置为空则不开启 WebUI               |
+|       bilichat_rss_base       |    str    |   None   |   兼容 RSSHub 的动态 RSS 订阅地址，例如 `https://rsshub.app/`    |
 
 注:
 
@@ -183,6 +184,7 @@ bilichat_openai_proxy = "http://127.0.0.1:7890/"
 3. `bilichat_dynamic_font` 可填写自定义的字体 url，但并不推荐修改
 4. 当使用 `bcut_asr` 接口来生成 AI 字幕时，根据视频时长和网络情况有可能会识别失败，Bot 会提示 `BCut-ASR conversion failed due to network error`。可以通过调高 `bilichat_neterror_retry` 次数或几分钟后重试来尝试重新生成字幕
 5. 当 `bilichat_cache_serive` 为 `mongodb` 时，需要安装并配置 [nonebot-plugin-mongodb](https://github.com/Well2333/nonebot-plugin-mongodb) 才可正常使用
+6. `bilichat_rss_base` 所需地址需要兼容 [RSSHub](https://docs.rsshub.app/zh/routes/social-media#up-%E4%B8%BB%E5%8A%A8%E6%80%81) 中的文件结构及路由即可使用
 
 ### 指令配置项
 

nonebot_plugin_bilichat/config.py

@@ -33,6 +33,7 @@ class Config(BaseModel):
     bilichat_text_fonts: str = "default"
     bilichat_emoji_fonts: str = "default"
     bilichat_webui_path: str | None = "bilichat"
+    bilichat_rss_base: str = ""
 
     # command and subscribe
     bilichat_command_to_me: bool = True
@@ -87,6 +88,19 @@ class Config(BaseModel):
     bilichat_openai_token_limit: int = 3500
     bilichat_openai_api_base: str = "https://api.openai.com"
 
+    @validator("bilichat_rss_base", always=True)
+    def check_rss_base(cls, v: str) -> str:
+        if not v:
+            return v
+        if not v.endswith("/"):
+            v += "/"
+        # warning rsshub.app
+        if "https://rsshub.app/" in v:
+            logger.warning(
+                "请注意 rsshub.app 作为开源项目的演示站点，请仅作为测试使用，如有需求请自行搭建 rsshub 服务，**不要滥用公共服务**"
+            )
+        return v
+
     @validator("bilichat_cache_serive", always=True, pre=True)
     def check_cache_serive(cls, v):
         if v == "json":

nonebot_plugin_bilichat/content/dynamic.py

@@ -2,7 +2,7 @@
 
 from bilireq.exceptions import GrpcError, ResponseCodeError
 from bilireq.grpc.dynamic import grpc_get_dynamic_detail
-from bilireq.grpc.protos.bilibili.app.dynamic.v2.dynamic_pb2 import DynamicType, DynModuleType
+from bilireq.grpc.protos.bilibili.app.dynamic.v2.dynamic_pb2 import DynamicItem, DynamicType, DynModuleType
 from google.protobuf.json_format import MessageToDict
 from httpx import TimeoutException
 from nonebot.log import logger
@@ -32,51 +32,57 @@ class Dynamic(BaseModel):
     def bili_id(self) -> str:
         return f"动态id: {self.id}"
 
-    async def _grpc(self):
-        logger.debug("正在使用 gRPC 获取动态信息")
-        for i in range(plugin_config.bilichat_neterror_retry):
-            try:
-                _dyn = await grpc_get_dynamic_detail(dynamic_id=self.id, auth=AuthManager.get_auth())
-                break
-            except TimeoutException:
-                logger.warning(f"请求超时，重试第 {i + 1}/{plugin_config.bilichat_neterror_retry} 次")
+    async def _grpc(self, raw_dyn: DynamicItem | None = None):
+        if not raw_dyn:
+            logger.debug("正在使用 gRPC 获取动态信息")
+            for i in range(plugin_config.bilichat_neterror_retry):
+                try:
+                    raw_dyn = (await grpc_get_dynamic_detail(dynamic_id=self.id, auth=AuthManager.get_auth())).item
+                    break
+                except TimeoutException:
+                    logger.warning(f"请求超时，重试第 {i + 1}/{plugin_config.bilichat_neterror_retry} 次")
+            else:
+                raise AbortError("请求超时")
         else:
-            raise AbortError("请求超时")
-        dyn = _dyn.item
-        self.raw = MessageToDict(dyn)
+            logger.debug("使用已有的 gRPC 动态信息")
+        self.raw = MessageToDict(raw_dyn)
         self.raw_type = "grpc"
-        self.dynamic_type = dyn.card_type
-        if dyn.card_type == DynamicType.av:
-            for module in dyn.modules:
+        self.dynamic_type = raw_dyn.card_type
+        if raw_dyn.card_type == DynamicType.av:
+            for module in raw_dyn.modules:
                 if module.module_type == DynModuleType.module_dynamic:
                     aid = module.module_dynamic.dyn_archive.avid
                     self.url = await get_b23_url(f"https://www.bilibili.com/video/av{aid}")
-        elif dyn.card_type == DynamicType.article:
-            for module in dyn.modules:
+        elif raw_dyn.card_type == DynamicType.article:
+            for module in raw_dyn.modules:
                 if module.module_type == DynModuleType.module_dynamic:
                     cvid = module.module_dynamic.dyn_article.id
                     self.url = await get_b23_url(f"https://www.bilibili.com/read/cv{cvid}")
 
-    async def _web(self):
-        logger.debug("正在使用 RestAPI 获取动态信息")
-        for i in range(plugin_config.bilichat_neterror_retry):
-            try:
-                dyn = (await get_dynamic(self.id))["item"]
-                break
-            except TimeoutException:
-                logger.warning(f"请求超时，重试第 {i + 1}/{plugin_config.bilichat_neterror_retry} 次")
+    async def _web(self, raw_dyn: dict | None = None):
+        if not raw_dyn:
+            logger.debug("正在使用 RestAPI 获取动态信息")
+            for i in range(plugin_config.bilichat_neterror_retry):
+                try:
+                    raw_dyn = (await get_dynamic(self.id))["item"]
+                    break
+                except TimeoutException:
+                    logger.warning(f"请求超时，重试第 {i + 1}/{plugin_config.bilichat_neterror_retry} 次")
+            else:
+                raise AbortError("请求超时")
         else:
-            raise AbortError("请求超时")
-        self.raw = dyn
+            logger.debug("使用已有的 RestAPI 动态信息")
+        assert raw_dyn
+        self.raw = raw_dyn
         self.raw_type = "web"
 
-        self.dynamic_type = DYNAMIC_TYPE_MAP.get(dyn["type"], DynamicType.dyn_none)
+        self.dynamic_type = DYNAMIC_TYPE_MAP.get(raw_dyn["type"], DynamicType.dyn_none)
 
         if self.dynamic_type == DynamicType.av:
-            aid = dyn["modules"]["module_dynamic"]["major"]["archive"]["aid"]
+            aid = raw_dyn["modules"]["module_dynamic"]["major"]["archive"]["aid"]
             self.url = await get_b23_url(f"https://www.bilibili.com/video/av{aid}")
         elif self.dynamic_type == DynamicType.article:
-            cvid = dyn["modules"]["module_dynamic"]["major"]["article"]["id"]
+            cvid = raw_dyn["modules"]["module_dynamic"]["major"]["article"]["id"]
             self.url = await get_b23_url(f"https://www.bilibili.com/read/cv{cvid}")
 
     @classmethod

nonebot_plugin_bilichat/lib/fetch_dynamic.py

@@ -1,12 +1,16 @@
 import asyncio
+import re
+import xml.etree.ElementTree as ET
 
+import httpx
 from bilireq.exceptions import GrpcError, ResponseCodeError
 from bilireq.grpc.dynamic import grpc_get_user_dynamics
 from google.protobuf.json_format import MessageToDict
 from grpc.aio import AioRpcError
 from httpx import TimeoutException
 from nonebot.log import logger
 
+from ..config import plugin_config
 from ..content.dynamic import Dynamic
 from ..lib.bilibili_request import get_b23_url, get_user_dynamics
 from ..lib.bilibili_request.auth import AuthManager
@@ -18,16 +22,18 @@
 
 
 async def fetch_last_dynamic(up: SearchUp) -> Dynamic | None:
-    if SubscriptionSystem.config.dynamic_grpc:
+    if SubscriptionSystem.config.dynamic_method == "grpc":
         try:
-            return await _fetch_grpc(up.mid, up.nickname)
+            return await _fetchlast_grpc(up.mid, up.nickname)
         except AbortError:
-            return await _fetch_rest(up.mid, up.nickname)
-    else:
-        return await _fetch_rest(up.mid, up.nickname)
+            return await _fetchlast_rest(up.mid, up.nickname)
+    elif SubscriptionSystem.config.dynamic_method == "rest":
+        return await _fetchlast_rest(up.mid, up.nickname)
+    elif SubscriptionSystem.config.dynamic_method == "rss":
+        return await _fetchlast_rss(up.mid)
 
 
-async def _fetch_rest(up_mid: int, up_name: str) -> Dynamic | None:
+async def _fetchlast_rest(up_mid: int, up_name: str) -> Dynamic | None:
     try:
         resp: list = (await get_user_dynamics(up_mid))["items"]
     except TimeoutException:
@@ -53,7 +59,7 @@ async def _fetch_rest(up_mid: int, up_name: str) -> Dynamic | None:
     return dynamic
 
 
-async def _fetch_grpc(up_mid: int, up_name: str) -> Dynamic | None:
+async def _fetchlast_grpc(up_mid: int, up_name: str) -> Dynamic | None:
     try:
         resp = await asyncio.wait_for(grpc_get_user_dynamics(up_mid, auth=AuthManager.get_auth()), timeout=10)
     except asyncio.TimeoutError:
@@ -81,3 +87,29 @@ async def _fetch_grpc(up_mid: int, up_name: str) -> Dynamic | None:
         id=dyn.extend.dyn_id_str, url=url, dynamic_type=dyn.card_type, raw=MessageToDict(dyn), raw_type="grpc"
     )
     return dynamic
+
+
+async def _fetchlast_rss(mid: int) -> Dynamic | None:
+    try:
+        url = f"{plugin_config.bilichat_rss_base}bilibili/user/dynamic/{mid}"
+        async with httpx.AsyncClient() as client:
+            response = await client.get(url)
+            response.raise_for_status()
+            rss_xml = response.text
+        root = ET.fromstring(rss_xml)
+        items = root.findall("channel/item")
+
+        for item in items:
+            link: str = item.find("link").text  # type: ignore
+            dynamic_id: str = re.search(r"t.bilibili.com/(\d+)", link).group(1)  # type: ignore
+            return await Dynamic.from_id(dynamic_id)
+    except httpx.HTTPStatusError as e:
+        if e.response.status_code == 503:
+            logger.error(f"RSS 订阅服务不可用 {type(e)}:{e}")
+            raise AbortError("Dynamic Abort")
+        else:
+            logger.exception(f"获取 {mid} RSS 动态失败: {e}")
+            return
+    except Exception as e:
+        logger.exception(f"[Dynamic] 获取 {mid} RSS 动态失败: {e}")
+        return

nonebot_plugin_bilichat/subscribe/__init__.py

@@ -5,14 +5,14 @@
 
 from ..model.exception import AbortError
 from ..optional import capture_exception
-from .dynamic import fetch_dynamics_grpc, fetch_dynamics_rest
+from .dynamic import fetch_dynamics_grpc, fetch_dynamics_rest, fetch_dynamics_rss
 from .live import fetch_live
 from .manager import CONFIG_LOCK, SubscriptionSystem
 
 NO_ACTIVE_UP = 0
 
 
-async def _check_activate_uploaders(func:str = "Scheduler") -> bool:
+async def _check_activate_uploaders(func: str = "Scheduler") -> bool:
     global NO_ACTIVE_UP
     if SubscriptionSystem.activate_uploaders:
         NO_ACTIVE_UP = 0
@@ -49,7 +49,7 @@ async def run_dynamic_update():
         while CONFIG_LOCK.locked():
             await asyncio.sleep(0)
         async with CONFIG_LOCK:
-            if SubscriptionSystem.config.dynamic_grpc:
+            if SubscriptionSystem.config.dynamic_method == "grpc":
                 try:
                     logger.debug(f"[Dynamic] 使用gRPC获取 {up.nickname}({up.uid}) | offset={up.dyn_offset}")
                     await fetch_dynamics_grpc(up)
@@ -59,16 +59,27 @@ async def run_dynamic_update():
                 except Exception:
                     capture_exception()
                     logger.exception(f"[Dynamic] 获取 {up} 失败.")
+            elif SubscriptionSystem.config.dynamic_method == "rest":
+                try:
+                    logger.debug(f"[Dynamic] 使用 RestAPI 获取 {up.nickname}({up.uid}) | offset={up.dyn_offset}")
+                    await fetch_dynamics_rest(up)
+                    continue
+                except AbortError:
+                    logger.error(f"[Dynamic] 获取 {up} 失败, skip...")
+                except Exception:
+                    capture_exception()
+                    logger.exception(f"[Dynamic] 获取 {up} 失败, skip...")
+            elif SubscriptionSystem.config.dynamic_method == "rss":
+                try:
+                    logger.debug(f"[Dynamic] 使用 RSS 获取 {up.nickname}({up.uid}) | offset={up.dyn_offset}")
+                    await fetch_dynamics_rss(up)
+                    continue
+                except AbortError:
+                    logger.error(f"[Dynamic] 获取 {up} 失败, skip...")
+                except Exception:
+                    capture_exception()
+                    logger.exception(f"[Dynamic] 获取 {up} 失败, skip...")
 
-            try:
-                logger.debug(f"[Dynamic] 使用 RestAPI 获取 {up.nickname}({up.uid}) | offset={up.dyn_offset}")
-                await fetch_dynamics_rest(up)
-                continue
-            except AbortError:
-                logger.error(f"[Dynamic] 获取 {up} 失败, skip...")
-            except Exception:
-                capture_exception()
-                logger.exception(f"[Dynamic] 获取 {up} 失败, skip...")
     logger.debug("[Dynamic] 获取完成")