feat(bilibili): 增强哔哩哔哩图文动态解析

src/nonebot_plugin_parser/parsers/bilibili/__init__.py

@@ -67,10 +67,11 @@ async def _parse_av(self, searched: Match[str]):
 
     @handle("/dynamic/", r"bilibili\.com/dynamic/(?P<dynamic_id>\d+)")
     @handle("t.bili", r"t\.bilibili\.com/(?P<dynamic_id>\d+)")
+    @handle("/opus/", r"bilibili\.com/opus/(?P<dynamic_id>\d+)")
     async def _parse_dynamic(self, searched: Match[str]):
         """解析动态信息"""
         dynamic_id = int(searched.group("dynamic_id"))
-        return await self.parse_dynamic(dynamic_id)
+        return await self.parse_dynamic_or_opus(dynamic_id)
 
     @handle("live.bili", r"live\.bilibili\.com/(?P<room_id>\d+)")
     async def _parse_live(self, searched: Match[str]):
@@ -87,14 +88,12 @@ async def _parse_favlist(self, searched: Match[str]):
     @handle("/read/", r"bilibili\.com/read/cv(?P<read_id>\d+)")
     async def _parse_read(self, searched: Match[str]):
         """解析专栏信息"""
-        read_id = int(searched.group("read_id"))
-        return await self.parse_read_with_opus(read_id)
+        from bilibili_api.article import Article
 
-    @handle("/opus/", r"bilibili\.com/opus/(?P<opus_id>\d+)")
-    async def _parse_opus(self, searched: Match[str]):
-        """解析图文动态信息"""
-        opus_id = int(searched.group("opus_id"))
-        return await self.parse_opus(opus_id)
+        read_id = int(searched.group("read_id"))
+        article = Article(read_id)
+        opus = await article.turn_to_opus()
+        return await self._parse_bilibli_api_opus(opus)
 
     async def parse_video(
         self,
@@ -167,8 +166,8 @@ async def download_video():
             extra={"info": ai_summary},
         )
 
-    async def parse_dynamic(self, dynamic_id: int):
-        """解析动态信息
+    async def parse_dynamic_or_opus(self, dynamic_id: int):
+        """解析动态和图文信息
 
         Args:
             url (str): 动态链接
@@ -178,8 +177,10 @@ async def parse_dynamic(self, dynamic_id: int):
         from .dynamic import DynamicData
 
         dynamic = Dynamic(dynamic_id, await self.credential)
-        dynamic_info = convert(await dynamic.get_info(), DynamicData).item
+        if await dynamic.is_article():
+            return await self._parse_bilibli_api_opus(dynamic.turn_to_opus())
 
+        dynamic_info = convert(await dynamic.get_info(), DynamicData).item
         author = self.create_author(dynamic_info.name, dynamic_info.avatar)
 
         # 下载图片
@@ -196,27 +197,16 @@ async def parse_dynamic(self, dynamic_id: int):
             contents=contents,
         )
 
-    async def parse_opus(self, opus_id: int):
+    async def parse_opus_by_id(self, opus_id: int):
         """解析图文动态信息
 
         Args:
             opus_id (int): 图文动态 id
         """
         opus = Opus(opus_id, await self.credential)
-        return await self._parse_opus_obj(opus)
-
-    async def parse_read_with_opus(self, read_id: int):
-        """解析专栏信息, 使用 Opus 接口
-
-        Args:
-            read_id (int): 专栏 id
-        """
-        from bilibili_api.article import Article
-
-        article = Article(read_id)
-        return await self._parse_opus_obj(await article.turn_to_opus())
+        return await self._parse_bilibli_api_opus(opus)
 
-    async def _parse_opus_obj(self, bili_opus: Opus):
+    async def _parse_bilibli_api_opus(self, bili_opus: Opus):
         """解析图文动态信息
 
         Args:

src/nonebot_plugin_parser/parsers/bilibili/dynamic.py

@@ -63,9 +63,10 @@ class OpusContent(Struct):
 class DynamicMajor(Struct):
     """动态主要内容"""
 
-    type: str
+    type: str | None = None
     archive: VideoArchive | None = None
     opus: OpusContent | None = None
+    desc: OpusSummary | None = None
 
     @property
     def title(self) -> str | None:
@@ -81,6 +82,8 @@ def text(self) -> str | None:
             return self.archive.desc
         elif self.type == "MAJOR_TYPE_OPUS" and self.opus:
             return self.opus.summary.text
+        elif self.desc:
+            return self.desc.text
         return None
 
     @property
@@ -126,7 +129,10 @@ def pub_ts(self) -> int:
     def major_info(self) -> dict[str, Any] | None:
         """获取主要内容信息"""
         if self.module_dynamic:
-            return self.module_dynamic.get("major")
+            if major := self.module_dynamic.get("major"):
+                return major
+            # 转发类型动态没有 major
+            return self.module_dynamic
         return None
 
 

tests/parsers/test_bilibili.py

@@ -42,8 +42,7 @@ async def test_read():
     url = "https://www.bilibili.com/read/cv523868"
     parser = BilibiliParser()
     _, searched = parser.search_url(url)
-    read_id = int(searched.group("read_id"))
-    result = await parser.parse_read_with_opus(read_id)
+    result = await parser._parse_read(searched)
     logger.debug(f"result: {result}")
     assert result.title, "标题为空"
     assert result.author, "作者为空"
@@ -60,58 +59,21 @@ async def test_read():
 
 
 @pytest.mark.asyncio
-async def test_opus():
+async def test_dynamic():
     from nonebot_plugin_parser.parsers import BilibiliParser
 
-    opus_urls = [
+    dynamic_urls = [
+        "https://t.bilibili.com/1120105154190770281",
         "https://www.bilibili.com/opus/998440765151510535",
         "https://www.bilibili.com/opus/1040093151889457152",
     ]
 
     parser = BilibiliParser()
 
-    async def test_parse_opus(opus_url: str) -> None:
-        _, searched = parser.search_url(opus_url)
-        opus_id = int(searched.group("opus_id"))
-        try:
-            result = await parser.parse_opus(opus_id)
-        except Exception as e:
-            pytest.skip(f"{opus_url} | opus 解析失败: {e} (风控)")
-
-        assert result.contents, "内容为空"
-        for content in result.contents:
-            path = await content.get_path()
-            assert path.exists(), "内容不存在"
-
-        assert result.author, "作者为空"
-        avatar_path = await result.author.get_avatar_path()
-        assert avatar_path, "头像不存在"
-        assert avatar_path.exists(), "头像不存在"
-
-        graphics_contents = result.graphics_contents
-        assert graphics_contents, "图文内容为空"
-
-        for graphics_content in graphics_contents:
-            path = await graphics_content.get_path()
-            assert path.exists(), "图文内容不存在"
-
-    await asyncio.gather(*[test_parse_opus(opus_url) for opus_url in opus_urls])
-    logger.success("B站动态解析成功")
-
-
-@pytest.mark.asyncio
-async def test_dynamic():
-    from nonebot_plugin_parser.parsers import BilibiliParser
-
-    dynamic_urls = ["https://t.bilibili.com/1120105154190770281"]
-
-    parser = BilibiliParser()
-
     async def test_parse_dynamic(dynamic_url: str) -> None:
         _, searched = parser.search_url(dynamic_url)
         dynamic_id = int(searched.group("dynamic_id"))
-        result = await parser.parse_dynamic(dynamic_id)
-        assert result.title, "标题为空"
+        result = await parser.parse_dynamic_or_opus(dynamic_id)
         assert result.author, "作者为空"
         avatar_path = await result.author.get_avatar_path()
         assert avatar_path, "头像不存在"