📜 smart log

.vscode/settings.json

@@ -21,6 +21,7 @@
     "disqus",
     "docformat",
     "DOTALL",
+    "excepthook",
     "exectarget",
     "favicon",
     "fontawesome",

docs/appendix/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## 未发布的更新
 
+### 改进
+
+`Ariadne` 会智能注入日志和异常处理元件, 就像 `0.6.x` 一样.
+
 ### 修复
 
 修复了 `MessageChain` 的 MRO 问题.

src/graia/ariadne/app.py

@@ -7,6 +7,7 @@
 import io
 import os
 import sys
+import traceback
 from datetime import datetime
 from typing import (
     TYPE_CHECKING,
@@ -62,7 +63,13 @@
     T,
     classmethod,
 )
-from .util import AttrConvertMixin, ariadne_api
+from .util import (
+    AttrConvertMixin,
+    RichLogInstallOptions,
+    ariadne_api,
+    loguru_exc_callback,
+    loguru_exc_callback_async,
+)
 
 if TYPE_CHECKING:
     from .message.element import Image, Voice
@@ -100,6 +107,10 @@ def _ensure_config(cls):
             cls.launch_manager = Launart()
         cls.held_objects.setdefault(Broadcast, cls.service.broadcast)
         cls.held_objects.setdefault(asyncio.AbstractEventLoop, cls.service.loop)
+        if "install_log" not in cls.options:
+            sys.excepthook = loguru_exc_callback
+        traceback.print_exception = loguru_exc_callback
+        cls.service.loop.set_exception_handler(loguru_exc_callback_async)
 
     @classmethod
     def config(
@@ -109,7 +120,7 @@ def config(
         broadcast: Optional[Broadcast] = None,
         launch_manager: Optional[Launart] = None,
         default_account: Optional[int] = None,
-        install_log: bool = False,
+        install_log: Union[bool, RichLogInstallOptions] = False,
         inject_bypass_listener: bool = False,
     ) -> None:
         """配置 Ariadne 全局参数, 未提供的值会自动生成合理的默认值
@@ -120,7 +131,7 @@ def config(
             broadcast (Optional[Broadcast], optional): 事件系统, 与 `loop` 参数互斥
             launch_manager (Optional[LaunchManager], optional): 启动管理器
             default_account (Optional[int], optional): 默认账号
-            install_log (bool, optional): 是否安装 rich 日志, 默认为 False
+            install_log (Union[bool, RichLogInstallOptions], optional): 是否安装 rich 日志, 默认为 False
             inject_bypass_listener (bool, optional): 是否注入透传 Broadcast, 默认为 False
         """
         if loop:
@@ -148,7 +159,11 @@ def config(
         if install_log and "installed_log" not in cls.options:
             import richuru
 
-            richuru.install()
+            option = (
+                install_log if isinstance(install_log, RichLogInstallOptions) else RichLogInstallOptions()
+            )
+
+            richuru.install(**option._asdict())
             cls.options["installed_log"] = True
 
         if inject_bypass_listener and "inject_bypass_listener" not in cls.options:

src/graia/ariadne/typing.py

@@ -5,17 +5,19 @@
 import enum
 import sys
 import typing
-from types import MethodType
+from types import MethodType, TracebackType
 from typing import (
     TYPE_CHECKING,
     AbstractSet,
     Any,
+    Callable,
     Dict,
     Generic,
     Mapping,
     Optional,
     Sequence,
     Tuple,
+    Type,
     TypedDict,
     TypeVar,
     Union,
@@ -184,6 +186,8 @@ class _SentinelClass(enum.Enum):
 
 AnnotatedType = type(Annotated[int, lambda x: x > 0])
 
+ExceptionHook = Callable[[Type[BaseException], BaseException, Optional[TracebackType]], Any]
+
 if sys.version_info >= (3, 9):
     classmethod = builtins.classmethod
 else:

src/graia/ariadne/util/__init__.py

@@ -1,13 +1,17 @@
 """本模块提供 Ariadne 内部使用的小工具, 以及方便的 `async_exec` 模块.
 """
 
+import contextlib
+
 # Utility Layout
 import functools
 import inspect
 import sys
+import types
 import typing
 import warnings
 from contextvars import ContextVar
+from types import TracebackType
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -18,10 +22,12 @@
     Iterable,
     List,
     MutableSet,
+    NamedTuple,
     Optional,
     Tuple,
     Type,
     TypeVar,
+    Union,
 )
 
 from graia.broadcast import Broadcast
@@ -30,12 +36,77 @@
 from graia.broadcast.entities.event import Dispatchable
 from graia.broadcast.entities.listener import Listener
 from graia.broadcast.entities.namespace import Namespace
+from graia.broadcast.exceptions import ExecutionStop, RequirementCrashed
 from graia.broadcast.interfaces.dispatcher import DispatcherInterface
 from graia.broadcast.typing import T_Dispatcher
 from graia.broadcast.utilles import dispatcher_mixin_handler
 from loguru import logger
 
-from ..typing import DictStrAny, P, R, T
+from ..typing import DictStrAny, ExceptionHook, P, R, T
+
+if TYPE_CHECKING:
+    from datetime import datetime
+
+    from rich.console import Console
+    from rich.text import Text
+
+
+def loguru_exc_callback(cls: Type[BaseException], val: BaseException, tb: Optional[TracebackType], *_, **__):
+    """loguru 异常回调
+
+    Args:
+        cls (Type[Exception]): 异常类
+        val (Exception): 异常的实际值
+        tb (TracebackType): 回溯消息
+    """
+    if tb:
+        exec_module_name = tb.tb_frame.f_globals.get("__name__", "")
+        if issubclass(cls, ExecutionStop) and exec_module_name.startswith("graia"):
+            return
+        elif isinstance(val, RequirementCrashed) and exec_module_name.startswith("graia.broadcast"):
+            with contextlib.suppress(Exception):
+                local_dict = tb.tb_frame.f_locals
+                _, param_name, param_anno, param_default = val.args
+                if isinstance(param_anno, type):
+                    param_anno = param_anno.__qualname__
+                param_repr = "".join(
+                    [
+                        param_name,
+                        f": {param_anno}" if param_anno else "",
+                        f" = {param_default}" if param_default else "",
+                    ]
+                )
+                val = RequirementCrashed(
+                    f"Unable to lookup parameter ({param_repr})",
+                    local_dict["dispatchers"],
+                )
+
+    logger.opt(exception=(cls, val, tb)).error("Exception:")
+
+
+def loguru_exc_callback_async(_, ctx: dict):
+    """loguru 异步异常回调
+
+    Args:
+        _ (AbstractEventLoop): 异常发生的事件循环
+        ctx (dict): 异常上下文
+    """
+    if "exception" in ctx:
+        logger.opt(exception=ctx["exception"]).error("Exception:")
+    else:
+        logger.error(f"Exception: {ctx}")
+
+
+class RichLogInstallOptions(NamedTuple):
+    rich_console: Optional["Console"] = None
+    exc_hook: Union[ExceptionHook, None] = loguru_exc_callback
+    rich_traceback: bool = False
+    tb_ctx_lines: int = 3
+    tb_theme: Optional[str] = None
+    tb_suppress: Iterable[Union[str, types.ModuleType]] = ()
+    time_format: Union[str, Callable[["datetime"], "Text"]] = "[%x %X]"
+    keywords: Optional[List[str]] = None
+    level: Union[int, str] = 20
 
 
 def inject_bypass_listener(broadcast: Broadcast):