feat(bilibili): dynamic and live debug command

Author: ProgramRipper

src/plugins/bilibili/plugins/_utils.py

@@ -1,9 +1,11 @@
 import re
 import string
 from random import choices
-from typing import Any
+from typing import Any, NoReturn
 
 from httpx import AsyncClient, Response
+from nonebot import logger
+from nonebot_plugin_alconna import UniMessage
 
 
 def bv2av(bvid: str) -> int:
@@ -83,3 +85,9 @@ async def get_share_placard(
             },
         )
     )
+
+
+async def handle_error(message: str) -> NoReturn:
+    logger.error(message)
+    await UniMessage(message).send()
+    raise

src/plugins/bilibili/plugins/dynamic/__init__.py

@@ -7,7 +7,7 @@
 import backoff
 from arclet.alconna import Arg
 from httpx import AsyncClient
-from nonebot import get_driver, get_plugin_config, logger
+from nonebot import get_driver, get_plugin_config
 from nonebot.plugin import PluginMetadata
 from nonebot_plugin_alconna import Alconna, Image, UniMessage, on_alconna
 from nonebot_plugin_apscheduler import scheduler
@@ -20,7 +20,7 @@
 
 from .....utils import ADMIN, run_task, send_message
 from ... import plugin_config as bilibili_config
-from .._utils import get_share_click, raise_for_status
+from .._utils import get_share_click, handle_error, raise_for_status
 from .config import Config
 from .models import Dynamic, Dynamics, Subscription
 
@@ -168,9 +168,9 @@ async def broadcast(dynamic: Dynamic):
 
 
 @backoff.on_exception(backoff.constant, TimeoutError, max_tries=3)
-async def render_screenshot(dynamic: Dynamic) -> bytes:
+async def render_screenshot(id_str: str) -> bytes:
     async with get_new_page() as page:
-        await page.goto(f"https://t.bilibili.com/{dynamic['id_str']}")
+        await page.goto(f"https://t.bilibili.com/{id_str}")
         await page.wait_for_load_state("networkidle")
 
         if "opus" in page.url:
@@ -222,9 +222,7 @@ async def _(db: async_scoped_session, sess: EventSession, uid: str):
             ] not in {1, 2, 6}:
                 await modify_relation(uid, 1)
         except Exception:
-            logger.error("订阅B站动态失败")
-            await UniMessage("订阅B站动态失败").send()
-            raise
+            await handle_error("订阅B站动态失败")
 
     sub = Subscription(uid=int(uid), session_id=await get_session_persist_id(sess))
     if sub in dynamic_subs[uid]:
@@ -258,7 +256,7 @@ async def _(db: async_scoped_session, sess: EventSession, uid: str):
             ] in {1, 2, 6}:
                 await modify_relation(uid, 2)
         except Exception:
-            logger.error("取消关注失败")
+            await handle_error("取消关注失败")
 
     await UniMessage(f"成功取订 UID:{uid} 的动态").send()
 
@@ -287,3 +285,25 @@ async def _(db: async_scoped_session, sess: EventSession):
     await UniMessage(
         "已订阅动态:\n" + "\n".join(f"UID:{sub.uid}" for sub in subs)
     ).send()
+
+
+@on_alconna(Alconna("展示B站动态", Arg("id_str", r"re:\d+")), permission=ADMIN).handle()
+async def _(id_str: str):
+    try:
+        dynamic = raise_for_status(
+            await client.get("/polymer/web-dynamic/v1/detail", params={"id": id_str})
+        )["item"]
+    except Exception:
+        await handle_error("获取动态信息失败")
+
+    screenshot, url = await gather(
+        render_screenshot(id_str),
+        get_share_click(id_str, "dynamic", "dt.dt-detail.0.0.pv"),
+    )
+    await plugin_config.template.format(
+        name=dynamic["modules"]["module_author"]["name"],
+        action=dynamic["modules"]["module_author"]["pub_action"]
+        or plugin_config.types[dynamic["type"]],
+        screenshot=Image(raw=screenshot),
+        url=url,
+    ).send()

src/plugins/bilibili/plugins/live/__init__.py

@@ -5,17 +5,17 @@
 
 from arclet.alconna import Arg
 from httpx import AsyncClient
-from nonebot import get_driver, get_plugin_config, logger
+from nonebot import get_driver, get_plugin_config
 from nonebot.plugin import PluginMetadata
-from nonebot_plugin_alconna import Alconna, Image, UniMessage, on_alconna
+from nonebot_plugin_alconna import Alconna, AtAll, Image, UniMessage, on_alconna
 from nonebot_plugin_apscheduler import scheduler
 from nonebot_plugin_orm import async_scoped_session, get_session
 from nonebot_plugin_session import EventSession
 from nonebot_plugin_session_orm import get_session_persist_id
 from sqlalchemy import select
 
 from .....utils import ADMIN, run_task, send_message
-from .._utils import get_share_click, raise_for_status
+from .._utils import get_share_click, handle_error, raise_for_status
 from .config import Config
 from .models import RoomInfo, Subscription
 
@@ -106,9 +106,7 @@ async def broadcast(uids: list[str]) -> None:
                     send_message(
                         sub.session.session,
                         plugin_config.live_template.format(
-                            url=url,
-                            cover=cover,
-                            **info,
+                            url=url, cover=cover, **info
                         ),
                     )
                 )
@@ -139,9 +137,7 @@ async def _(db: async_scoped_session, sess: EventSession, uid: str):
             await client.post(GET_ROOM_STATUS_INFO_URL, json={"uids": [uid]})
         )
     except Exception:
-        logger.error("获取直播间信息失败")
-        await UniMessage("获取直播间信息失败").send()
-        raise
+        await handle_error("获取直播间信息失败")
 
     try:
         info = infos[uid]
@@ -194,3 +190,25 @@ async def _(db: async_scoped_session, sess: EventSession):
             "{uname} (UID:{uid})".format_map(room_infos[str(sub.uid)]) for sub in subs
         )
     ).send()
+
+
+@on_alconna(
+    Alconna("展示B站直播", Arg("uid", r"re:(?:UID:)?\d+")), permission=ADMIN
+).handle()
+async def _(uid: str):
+    try:
+        info = raise_for_status(
+            await client.post(GET_ROOM_STATUS_INFO_URL, json={"uids": [uid]})
+        )[uid]
+    except Exception:
+        await handle_error("获取直播间信息失败")
+
+    url = await get_share_click(
+        info["room_id"], "vertical-three-point", "live.live-room-detail.0.0.pv"
+    )
+    cover = Image(
+        raw=(await client.get(info["cover_from_user"] or info["face"])).content
+    )
+    await plugin_config.live_template.format(url=url, cover=cover, **info).exclude(
+        AtAll
+    ).send()