add ncm-radio

src/plugins/nonebot_bison/platform/ncm_radio.py

@@ -0,0 +1,55 @@
+from typing import Any, Optional
+
+import httpx
+from ..post import Post
+from ..types import RawPost, Target
+from .platform import TargetMixin, NewMessage
+
+class NcmRadio(TargetMixin, NewMessage):
+
+    categories = {}
+    platform_name = 'ncm-radio'
+    enable_tag = False
+    enabled = True
+    is_common = False
+    schedule_type = 'interval'
+    schedule_kw = {'minutes': 10}
+    name = "网易云-电台"
+
+    async def get_target_name(self, target: Target) -> Optional[str]:
+        async with httpx.AsyncClient() as client:
+            res = await client.post(
+                    "http://music.163.com/api/dj/program/byradio",
+                    headers={'Referer': 'https://music.163.com/'},
+                    data={"radioId": target, "limit": 1000, "offset": 0}
+                    )
+            res_data = res.json()
+            if res_data['code'] != 200 or res_data['programs'] == 0:
+                return
+            return res_data['programs'][0]['radio']['name']
+
+    async def get_sub_list(self, target: Target) -> list[RawPost]:
+        async with httpx.AsyncClient() as client:
+            res = await client.post(
+                    "http://music.163.com/api/dj/program/byradio",
+                    headers={'Referer': 'https://music.163.com/'},
+                    data={"radioId": target, "limit": 1000, "offset": 0}
+                    )
+            res_data = res.json()
+            if res_data['code'] != 200:
+                return []
+            else:
+                return res_data['programs']
+
+    def get_id(self, post: RawPost) -> Any:
+        return post['id']
+
+    def get_date(self, post: RawPost) -> int:
+        return post['createTime'] // 1000
+
+    async def parse(self, raw_post: RawPost) -> Post:
+        text = '网易云电台更新：{}'.format(raw_post['name'])
+        target_name = raw_post['radio']['name']
+        pics = [raw_post['coverUrl']]
+        url = "https://music.163.com/#/program/{}".format(raw_post['id'])
+        return Post('ncm-radio', text=text, url=url, pics=pics, target_name=target_name)

tests/platforms/test_ncm_radio.py

@@ -0,0 +1,53 @@
+
+from .utils import get_json
+import pytest
+import respx
+import typing
+import time
+from httpx import Response
+
+if typing.TYPE_CHECKING:
+    import sys
+    sys.path.append('./src/plugins')
+    import nonebot_bison
+
+@pytest.fixture
+def ncm_radio(plugin_module: 'nonebot_bison'):
+    return plugin_module.platform.platform_manager['ncm-radio']
+
+@pytest.fixture(scope='module')
+def ncm_radio_raw():
+    return get_json('ncm_radio_ark.json')
+
+@pytest.fixture(scope='module')
+def ncm_radio_0(ncm_radio_raw):
+    return {
+            **ncm_radio_raw,
+            'programs': ncm_radio_raw['programs'][1:]
+            }
+
+@pytest.fixture(scope='module')
+def ncm_radio_1(ncm_radio_raw: dict):
+    res = ncm_radio_raw.copy()
+    res['programs'] = ncm_radio_raw['programs'][:]
+    res['programs'][0]['createTime'] = int(time.time() * 1000)
+    return res 
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_fetch_new(ncm_radio, ncm_radio_0, ncm_radio_1, dummy_user_subinfo):
+    ncm_router = respx.post("http://music.163.com/api/dj/program/byradio") 
+    ncm_router.mock(return_value=Response(200, json=ncm_radio_0))
+    target = '793745436'
+    res = await ncm_radio.fetch_new_post(target, [dummy_user_subinfo])
+    assert(ncm_router.called)
+    assert(len(res) == 0)
+    ncm_router.mock(return_value=Response(200, json=ncm_radio_1))
+    res2 = await ncm_radio.fetch_new_post(target, [dummy_user_subinfo])
+    post = res2[0][1][0]
+    print(post)
+    assert(post.target_type == 'ncm-radio')
+    assert(post.text == '网易云电台更新：「松烟行动」灰齐山麓')
+    assert(post.url == 'https://music.163.com/#/program/2494997688')
+    assert(post.pics == ['http://p1.music.126.net/H5em5xUNIYXcjJhOmeaSqQ==/109951166647436789.jpg'])
+    assert(post.target_name == '《明日方舟》游戏原声OST')