Merge pull request #97 from felinae98/bilibili-gangumi

添加 Bilibili 剧集订阅

src/plugins/nonebot_bison/platform/bilibili.py

@@ -229,3 +229,75 @@ class Bilibililive(StatusChange):
             target_name=target_name,
             compress=True,
         )
+
+
+class BilibiliBangumi(StatusChange):
+
+    categories = {}
+    platform_name = "bilibili-bangumi"
+    enable_tag = False
+    enabled = True
+    is_common = True
+    scheduler_class = "bilibili.com"
+    name = "Bilibili剧集"
+    has_target = True
+
+    _url = "https://api.bilibili.com/pgc/review/user"
+
+    async def get_target_name(self, target: Target) -> Optional[str]:
+        async with http_client() as client:
+            res = await client.get(self._url, params={"media_id": target})
+            res_data = res.json()
+            if res_data["code"]:
+                return None
+            return res_data["result"]["media"]["title"]
+
+    async def get_status(self, target: Target):
+        async with http_client() as client:
+            res = await client.get(
+                self._url,
+                params={"media_id": target},
+                timeout=4.0,
+            )
+            res_dict = res.json()
+            if res_dict["code"] == 0:
+                return {
+                    "index": res_dict["result"]["media"]["new_ep"]["index"],
+                    "index_show": res_dict["result"]["media"]["new_ep"]["index"],
+                    "season_id": res_dict["result"]["media"]["season_id"],
+                }
+            else:
+                raise self.FetchError
+
+    def compare_status(self, target: Target, old_status, new_status) -> list[RawPost]:
+        if new_status["index"] != old_status["index"]:
+            return [new_status]
+        else:
+            return []
+
+    async def parse(self, raw_post: RawPost) -> Post:
+        async with http_client() as client:
+            detail_res = await client.get(
+                f'http://api.bilibili.com/pgc/view/web/season?season_id={raw_post["season_id"]}'
+            )
+        detail_dict = detail_res.json()
+        lastest_episode = None
+        for episode in detail_dict["result"]["episodes"][::-1]:
+            if episode["badge"] in ("", "会员"):
+                lastest_episode = episode
+                break
+        if not lastest_episode:
+            lastest_episode = detail_dict["result"]["episodes"]
+
+        url = lastest_episode["link"]
+        pic = [lastest_episode["cover"]]
+        target_name = detail_dict["result"]["season_title"]
+        text = lastest_episode["share_copy"]
+        return Post(
+            self.name,
+            text=text,
+            url=url,
+            pics=pic,
+            target_name=target_name,
+            compress=True,
+        )

tests/platforms/static/bilibili-gangumi-hanhua0.json

@@ -0,0 +1 @@
+{"code":0,"message":"success","result":{"media":{"areas":[{"id":1,"name":"中国大陆"}],"cover":"http://i0.hdslb.com/bfs/bangumi/image/206428990cb5b54f4c114c0b3fcc10d8f5724f7c.jpg","media_id":28235413,"new_ep":{"id":519207,"index":"1","index_show":"更新至第2话"},"season_id":39719,"share_url":"https://www.bilibili.com/bangumi/media/md28235413","title":"汉化日记 第三季","type_name":"国创"}}}

tests/platforms/static/bilibili-gangumi-hanhua1.json

@@ -0,0 +1 @@
+{"code":0,"message":"success","result":{"media":{"areas":[{"id":1,"name":"中国大陆"}],"cover":"http://i0.hdslb.com/bfs/bangumi/image/206428990cb5b54f4c114c0b3fcc10d8f5724f7c.jpg","media_id":28235413,"new_ep":{"id":519207,"index":"2","index_show":"更新至第2话"},"season_id":39719,"share_url":"https://www.bilibili.com/bangumi/media/md28235413","title":"汉化日记 第三季","type_name":"国创"}}}

tests/platforms/test_bilibili_bangumi.py

@@ -0,0 +1,62 @@
+import typing
+
+import pytest
+import respx
+from httpx import Response
+from nonebug.app import App
+
+from .utils import get_json
+
+if typing.TYPE_CHECKING:
+    from nonebot_bison.platform.bilibili import BilibiliBangumi
+
+
+@pytest.fixture
+def bili_bangumi(app: App):
+    from nonebot_bison.platform import platform_manager
+
+    return platform_manager["bilibili-bangumi"]
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_fetch_bilibili_bangumi_status(
+    bili_bangumi: "BilibiliBangumi", dummy_user_subinfo
+):
+    from nonebot_bison.types import Target
+
+    bili_bangumi_router = respx.get(
+        "https://api.bilibili.com/pgc/review/user?media_id=28235413"
+    )
+    bili_bangumi_detail_router = respx.get(
+        "http://api.bilibili.com/pgc/view/web/season?season_id=39719"
+    )
+    bili_bangumi_router.mock(
+        return_value=Response(200, json=get_json("bilibili-gangumi-hanhua0.json"))
+    )
+    target = Target("28235413")
+    res = await bili_bangumi.fetch_new_post(target, [dummy_user_subinfo])
+    assert len(res) == 0
+
+    res = await bili_bangumi.fetch_new_post(target, [dummy_user_subinfo])
+    assert len(res) == 0
+
+    bili_bangumi_router.mock(
+        return_value=Response(200, json=get_json("bilibili-gangumi-hanhua1.json"))
+    )
+    bili_bangumi_detail_router.mock(
+        return_value=Response(
+            200, json=get_json("bilibili-gangumi-hanhua1-detail.json")
+        )
+    )
+    res2 = await bili_bangumi.fetch_new_post(target, [dummy_user_subinfo])
+
+    post = res2[0][1][0]
+    assert post.target_type == "Bilibili剧集"
+    assert post.text == "《汉化日记 第三季》第2话 什么是战区导弹防御系统工作日"
+    assert post.url == "https://www.bilibili.com/bangumi/play/ep519207"
+    assert post.target_name == "汉化日记 第三季"
+    assert post.pics == [
+        "http://i0.hdslb.com/bfs/archive/ea0a302c954f9dbc3d593e676486396c551529c9.jpg"
+    ]
+    assert post.compress == True

tests/platforms/test_bilibili_live.py

@@ -1,13 +1,9 @@
-from datetime import datetime
-
-import feedparser
 import pytest
 import respx
 from httpx import Response
 from nonebug.app import App
-from pytz import timezone
 
-from .utils import get_file, get_json
+from .utils import get_json
 
 
 @pytest.fixture