feat: improve error handling

src/plugins/aliyun/__init__.py

@@ -1,6 +1,7 @@
 from pathlib import Path
 
 import nonebot
+from nonebot import get_driver
 from nonebot.plugin import PluginMetadata
 
 from .config import Config
@@ -12,7 +13,7 @@
 )
 
 # load all aliyun plugin config from global config
-config = Config.parse_obj(nonebot.get_driver().config)
+config = Config.parse_obj(get_driver().config)
 
 # load all aliyun subplugins
 _sub_plugins = set()

src/plugins/aliyun/plugins/aliyun_cdn/__init__.py

@@ -90,12 +90,18 @@ async def resolve_src_bandwidth(domain: CDNDomain):
     file_bytes = await generate_src_image(stats)
     file_url = await FileHost(file_bytes).to_url()
 
-    await src_bandwidth.send(
-        UniMessage([
-            Text(f"{domain.group_alias}统计：\n当前CDN回源带宽数据为：{value:.2f}Mbps"),
-            Image(url=file_url, raw=file_bytes),
-        ])
-    )
+    try:
+        await src_bandwidth.send(
+            UniMessage([
+                Text(
+                    f"{domain.group_alias}统计：\n当前CDN回源带宽数据为：{value:.2f}Mbps"
+                ),
+                Image(url=file_url, raw=file_bytes),
+            ])
+        )
+
+    except Exception as e:
+        await src_bandwidth.send(f"发送结果时出现错误：{e}")
 
 
 async def get_http_code_stats(domain: CDNDomain):

src/plugins/bili_parser/common.py

@@ -19,7 +19,7 @@
 from .data_source import get_av_data
 
 bili_parse = on_alconna(
-    Alconna("bvideo", Args["type", ["av", "BV"]], Args["code", str]),
+    Alconna("bvideo", Args["type?", ["av", "BV"]], Args["code?", str]),
     use_cmd_start=True,
 )
 bili_parse.shortcut(r".*av(\d{1,12}).*", {"args": ["av", "{0}"]})
@@ -53,13 +53,22 @@ async def get_bili_data(
     type_: Match[str] = AlconnaMatch("type"),
     code: Match[str] = AlconnaMatch("code"),
 ):
-    if data := await get_av_data(
-        code.result,
-        type_.result == "BV",
-    ):
-        state[BILI_DATA] = data
+    if type_.available and code.available:
+        is_bv = type_.result == "BV"
+
+        try:
+            data = await get_av_data(code.result, is_bv)
+        except httpx.HTTPError as e:
+            await bili_parse.finish(f"请求 Bilibili API 时发生错误：{e}")
+
+        if data:
+            state[BILI_DATA] = data
+
+        else:
+            await bili_parse.finish("未找到相关的视频信息")
+
     else:
-        await bili_parse.finish()
+        await bili_parse.finish("请输入正确的视频类型及视频号，例如：/bvideo av 1024")
 
 
 @bili_parse.handle(parameterless=[Depends(get_bili_data)])
@@ -80,4 +89,8 @@ async def _(
         Text(f"网页：{await ShortURL(url=data['data']['link']).to_url()}"),
     ])
 
-    await bili_parse.send(message)
+    try:
+        await bili_parse.send(message)
+
+    except Exception as e:
+        await bili_parse.send(f"发送结果时出现错误：{e}")

src/plugins/nbnhhsh/__init__.py

@@ -1,9 +1,9 @@
-import nonebot
+from nonebot import get_driver
 from nonebot.plugin import PluginMetadata
 
 from .config import Config
 
-nbnhhsh_config = Config.parse_obj(nonebot.get_driver().config.dict(exclude_none=True))
+config = Config.parse_obj(get_driver().config)
 
 __plugin_meta__ = PluginMetadata(
     "能不能好好说话",

src/plugins/nbnhhsh/common.py

@@ -1,32 +1,43 @@
 from typing import Any
 
 import httpx
-from nonebot import on_command
-from nonebot.matcher import Matcher
-from nonebot.adapters import Message
-from nonebot.params import CommandArg, ArgPlainText
+from nonebot_plugin_alconna import Args, Match, Alconna, on_alconna
 
-from . import nbnhhsh_config
+from . import config
 
-nbnhhsh = on_command("hhsh", aliases={"nbnhhsh", "好好说话", "人话"}, priority=10)
+nbnhhsh = on_alconna(
+    Alconna("hhsh", Args["text?", str]),
+    aliases={"nbnhhsh", "好好说话", "人话"},
+    priority=10,
+    use_cmd_start=True,
+)
+
+
+def get_splitted_text(r):
+    translates = r.get("trans")
+    guesses = r.get("inputting")
+
+    return ",".join(translates or guesses or []) or "暂无翻译"
 
 
 @nbnhhsh.handle()
-async def set_text(matcher: Matcher, args: Message = CommandArg()) -> None:
-    if args.extract_plain_text():
-        matcher.set_arg("text", args)
+async def set_text(text: Match[str]) -> None:
+    if text.available:
+        try:
+            result = await guess(text.result)
+        except httpx.HTTPError as e:
+            await nbnhhsh.finish(f"查询出错，请稍后重试：\n{e}")
 
+        if not result:
+            await nbnhhsh.finish("未找到相关结果")
 
-@nbnhhsh.got("text", prompt="请重新输入缩写")
-async def got_text(text: str = ArgPlainText()):
-    result = await guess(text)
+        msg_seq = [
+            f"原文：{r.get('name')}\n翻译：{get_splitted_text(r)}" for r in result
+        ]
+        await nbnhhsh.finish("\n".join(msg_seq))
 
-    msg_seq = [
-        f"原文：{r.get('name')}\n"
-        f"翻译：{','.join(r.get('trans') or r.get('inputting') or []) or '暂无翻译'}"
-        for r in result
-    ]
-    await nbnhhsh.finish("\n".join(msg_seq))
+    else:
+        await nbnhhsh.finish("未提供要查询的拼音字母缩写，试试 /hhsh jk？")
 
 
 async def guess(text: str) -> list[dict[str, Any]]:
@@ -40,9 +51,9 @@ async def guess(text: str) -> list[dict[str, Any]]:
 
     json = {"text": text}
 
-    async with httpx.AsyncClient(timeout=10) as client:
+    async with httpx.AsyncClient() as client:
         r = await client.post(
-            f"{nbnhhsh_config.nbnhhsh_api_endpoint}/api/nbnhhsh/guess",
+            f"{config.nbnhhsh_api_endpoint}/api/nbnhhsh/guess",
             headers=headers,
             json=json,
         )