Drop ssttkkl-nonebot-utils

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "nonebot-plugin-access-control"
-version = "1.1.5"
+version = "1.2.0"
 description = ""
 authors = ["ssttkkl <[email protected]>"]
 license = "MIT"
@@ -13,14 +13,13 @@ packages = [
 [tool.poetry.dependencies]
 python = "^3.9"
 nonebot2 = ">=2.2.0, <3.0.0"
-nonebot-plugin-access-control-api = "^1.1.2"
+nonebot-plugin-access-control-api = "^1.2.0"
 nonebot-plugin-apscheduler = ">=0.3.0"
-nonebot-plugin-session = "^0.2.0"
+nonebot-plugin-session = "^0.3.0"
 nonebot-plugin-orm = ">=0.7.0, <1.0.0"
 arclet-alconna = "^1.7.24"
 shortuuid = "^1.0.11"
 pytimeparser = "^0.2.0"
-ssttkkl-nonebot-utils = ">=0.1.17"
 pydantic-settings = "^2.2.1"
 
 nb-cli = { version = ">=1.2.0", optional = true }
@@ -35,9 +34,9 @@ pre-commit = "^3.1.0"
 
 setuptools = "^68.1.2"
 nb-cli = "*"
-nonebot-plugin-orm = {extras = ["default"], version = "*"}
+nonebot-plugin-orm = { extras = ["default"], version = "*" }
 
-nonebot2 = {extras = ["fastapi"], version = "*"}
+nonebot2 = { extras = ["fastapi"], version = "*" }
 nonebot-adapter-onebot = "*"
 nonebot-adapter-kaiheila = "*"
 nonebot-adapter-qq = "*"

src/nonebot_plugin_access_control/handler/__init__.py

@@ -4,14 +4,14 @@
 from nonebot import logger
 from arclet.alconna import Arparma
 from arclet.alconna.typing import DataCollection
-from ssttkkl_nonebot_utils.errors.error_handler import ErrorHandlers
 from nonebot_plugin_access_control_api.errors import (
     PermissionDeniedError,
     AccessControlBadRequestError,
 )
 
 from ..alc import alc_ac
 from ..config import conf
+from .utils.errors import ErrorHandlers
 from . import (
     help_handler,
     limit_handler,

src/nonebot_plugin_access_control/handler/utils/errors.py

@@ -0,0 +1,95 @@
+from inspect import isawaitable
+from collections.abc import Awaitable
+from contextlib import asynccontextmanager
+from typing import Any, Union, Callable, Optional
+
+from nonebot.internal.matcher import current_matcher
+from nonebot.exception import ActionFailed, MatcherException
+from nonebot_plugin_access_control_api.errors import (
+    AccessControlQueryError,
+    AccessControlBadRequestError,
+)
+
+T_EXCEPTABLE = Union[type[BaseException], tuple[type[BaseException], ...]]
+T_ERROR_HANDLER = Union[
+    Callable[[BaseException], Optional[str]],
+    Callable[[BaseException], Awaitable[Optional[str]]],
+]
+
+
+class ErrorHandlers:
+    def __init__(self):
+        self.handlers: list[tuple[T_EXCEPTABLE, T_ERROR_HANDLER]] = []
+
+    def register(
+        self, error_type: T_EXCEPTABLE, func: Optional[T_ERROR_HANDLER] = None
+    ):
+        def decorator(func: Optional[T_ERROR_HANDLER]):
+            self.handlers.append((error_type, func))
+            return func
+
+        if func is not None:
+            decorator(func)
+        else:
+            return decorator
+
+    @asynccontextmanager
+    async def run_excepting(
+        self,
+        receive_error_message: Optional[
+            Union[Callable[[str], Any], Callable[[str], Awaitable[Any]]]
+        ] = None,
+        *,
+        reraise_unhandled: bool = False,
+    ):
+        try:
+            yield
+        except MatcherException as e:
+            raise e
+        except ActionFailed as e:
+            # 避免当发送消息错误时再尝试发送
+            raise e
+        except BaseException as e:
+            for excs, handler in self.handlers:
+                if not isinstance(excs, tuple):
+                    excs = (excs,)
+
+                for exc in excs:
+                    if isinstance(e, exc):
+                        msg = handler(e)
+                        if isawaitable(msg):
+                            msg = await msg
+
+                        if msg:
+                            coro = receive_error_message(msg)
+                            if isawaitable(coro):
+                                await coro
+                        return
+
+            if isinstance(e, (AccessControlBadRequestError, AccessControlQueryError)):
+                msg = e.message
+
+                matcher = None
+                try:
+                    matcher = current_matcher.get()
+                except LookupError:
+                    pass
+
+                help_info = getattr(matcher, "__help_info__", None)
+                if help_info is not None:
+                    msg += f"\n\n指令用法：{help_info}"
+
+                if msg:
+                    coro = receive_error_message(msg)
+                    if isawaitable(coro):
+                        await coro
+                return
+
+            # fallback
+            try:
+                coro = receive_error_message(f"内部错误：{type(e)}{str(e)}")
+                if isawaitable(coro):
+                    await coro
+            finally:
+                if reraise_unhandled:
+                    raise e  # 重新抛出未处理的异常