✨ logger

arclet/entari/__init__.py

@@ -58,3 +58,5 @@
 
 WS = WebsocketsInfo
 WH = WebhookInfo
+
+__version__ = "0.8.2"

arclet/entari/builtins/auto_reload.py

@@ -4,10 +4,10 @@
 
 from launart import Launart, Service, any_completed
 from launart.status import Phase
-from loguru import logger
 from watchfiles import PythonFilter, awatch
 
 from arclet.entari import Plugin, metadata
+from arclet.entari.logger import log
 from arclet.entari.plugin import dispose, find_plugin_by_file, load_plugin
 
 metadata(
@@ -17,6 +17,9 @@
 )
 
 
+logger = log.wrapper("[AutoReload]")
+
+
 class Watcher(Service):
     id = "watcher"
 
@@ -36,14 +39,14 @@ async def watch(self):
         async for event in awatch(*self.dirs, watch_filter=PythonFilter()):
             for change in event:
                 if plugin := find_plugin_by_file(change[1]):
-                    logger.info(f"[AutoReload] Detected change in {plugin.id}, reloading...")
+                    logger("INFO", f"Detected change in {plugin.id}, reloading...")
                     pid = plugin.id
                     del plugin
                     dispose(pid)
                     if plugin := load_plugin(pid):
-                        logger.info(f"[AutoReload] Reloaded {plugin.id}")
+                        logger("INFO", f"Reloaded {plugin.id}")
                     else:
-                        logger.error(f"[AutoReload] Failed to reload {pid}")
+                        logger("ERROR",f"Failed to reload {pid}")
 
     async def launch(self, manager: Launart):
         async with self.stage("blocking"):

arclet/entari/core.py

@@ -7,7 +7,6 @@
 from arclet.letoderea import BaseAuxiliary, Contexts, Param, Provider, ProviderFactory, es, global_providers
 from creart import it
 from launart import Launart, Service
-from loguru import logger
 from satori import LoginStatus
 from satori.client import App
 from satori.client.account import Account
@@ -23,6 +22,7 @@
 from .plugin import load_plugin
 from .plugin.service import plugin_service
 from .session import Session, EntariProtocol
+from .logger import log
 
 
 class ApiProtocolProvider(Provider[ApiProtocol]):
@@ -62,29 +62,38 @@ def from_config(cls, config: EntariConfig | None = None):
         if not config:
             config = EntariConfig.instance
         ignore_self_message = config.basic.get("ignore_self_message", True)
+        log_level = config.basic.get("log_level", "INFO")
         configs = []
         for conf in config.basic.get("network", []):
             if conf["type"] in ("websocket", "websockets", "ws"):
                 configs.append(WebsocketsInfo(**{k: v for k, v in conf.items() if k != "type"}))
             elif conf["type"] in ("webhook", "wh", "http"):
                 configs.append(WebhookInfo(**{k: v for k, v in conf.items() if k != "type"}))
-        for plug in config.plugin:
-            load_plugin(plug)
-        return cls(*configs, ignore_self_message=ignore_self_message)
+        return cls(*configs, log_level=log_level, ignore_self_message=ignore_self_message)
 
-    def __init__(self, *configs: Config, ignore_self_message: bool = True):
+    def __init__(
+        self,
+        *configs: Config,
+        log_level: str | int = "INFO",
+        ignore_self_message: bool = True
+    ):
+        from . import __version__
+        log.core.opt(colors=True).info(f"Entari <b><c>version {__version__}</c></b>")
         super().__init__(*configs, default_api_cls=EntariProtocol)
         if not hasattr(EntariConfig, "instance"):
             EntariConfig.load()
+        log.set_level(log_level)
+        for plug in EntariConfig.instance.plugin:
+            load_plugin(plug)
         self.ignore_self_message = ignore_self_message
         es.register(_commands.publisher)
         self.register(self.handle_event)
         self.lifecycle(self.account_hook)
         self._ref_tasks = set()
 
         @self.on_message(priority=0)
-        def log(event: MessageCreatedEvent):
-            logger.info(
+        def log_msg(event: MessageCreatedEvent):
+            log.message.info(
                 f"[{event.channel.name or event.channel.id}] "
                 f"{event.member.nick if event.member else (event.user.name or event.user.id)}"
                 f"({event.user.id}) -> {event.message.content!r}"
@@ -93,11 +102,11 @@ def log(event: MessageCreatedEvent):
         @es.use(SendResponse.__disp_name__)
         async def log_send(event: SendResponse):
             if event.session:
-                logger.info(
+                log.message.info(
                     f"[{event.session.channel.name or event.session.channel.id}] <- {event.message!r}"
                 )
             else:
-                logger.info(
+                log.message.info(
                     f"[{event.channel}] <- {event.message!r}"
                 )
 
@@ -130,7 +139,7 @@ async def handle_event(self, account: Account, event: Event):
             es.publish(ev)
             return
 
-        logger.warning(f"received unsupported event {event.type}: {event}")
+        log.core.warning(f"received unsupported event {event.type}: {event}")
 
     async def account_hook(self, account: Account, state: LoginStatus):
         _connected = []

arclet/entari/logger.py

@@ -0,0 +1,120 @@
+import inspect
+import logging
+import sys
+from typing import Union, TYPE_CHECKING, Optional
+
+from loguru import logger
+
+if TYPE_CHECKING:
+    from loguru import Logger
+
+
+class LoggerManager:
+    def __init__(self):
+        self.loggers: dict[str, "Logger"] = {}
+        self.fork("[core]")
+        self.fork("[plugin]")
+        self.fork("[message]")
+
+    def fork(self, child_name: str):
+        patched = logger.patch(lambda r: r.update(name=child_name))
+        patched = patched.bind(name=child_name)
+        self.loggers[child_name] = patched
+        return patched
+
+    @property
+    def core(self):
+        return self.loggers["[core]"]
+
+    @property
+    def plugin(self):
+        return self.loggers["[plugin]"]
+
+    @property
+    def message(self):
+        return self.loggers["[message]"]
+
+    def wrapper(self, name: str, color: str = "blue"):
+        patched = logger.patch(lambda r: r.update(name="entari"))
+        patched = patched.bind(name=f"plugins.{name}")
+        self.loggers[f"plugin.{name}"] = patched
+
+        def _log(level: str, message: str, exception: Optional[Exception] = None):
+            patched.opt(colors=True, exception=exception).log(
+                level, f"| <{color}>{name}</{color}> {message}"
+            )
+
+        return _log
+
+    @staticmethod
+    def set_level(level: Union[str, int]):
+        if isinstance(level, str):
+            level = level.upper()
+        logging.basicConfig(
+            handlers=[LoguruHandler()],
+            level=level,
+            format="%(asctime)s | %(name)s[%(levelname)s]: %(message)s",
+        )
+        logger.configure(extra={"entari_log_level": level}, patcher=_hidden_upsteam)
+
+
+class LoguruHandler(logging.Handler):  # pragma: no cover
+    """logging 与 loguru 之间的桥梁，将 logging 的日志转发到 loguru。"""
+
+    def emit(self, record: logging.LogRecord):
+        try:
+            level = logger.level(record.levelname).name
+        except ValueError:
+            level = record.levelno
+
+        frame, depth = inspect.currentframe(), 0
+        while frame and (
+            depth == 0 or frame.f_code.co_filename == logging.__file__
+        ):
+            frame = frame.f_back
+            depth += 1
+
+        logger.opt(depth=depth, exception=record.exc_info).log(
+            level, record.getMessage()
+        )
+
+
+logging.basicConfig(
+    handlers=[LoguruHandler()],
+    level="INFO",
+    format="%(asctime)s | %(name)s[%(levelname)s]: %(message)s",
+)
+
+
+def default_filter(record):
+    log_level = record["extra"].get("entari_log_level", "INFO")
+    levelno = (
+        logger.level(log_level).no if isinstance(log_level, str) else log_level
+    )
+    return record["level"].no >= levelno
+
+
+logger.remove()
+logger.add(
+    sys.stdout,
+    level=0,
+    diagnose=True,
+    backtrace=True,
+    colorize=True,
+    filter=default_filter,
+    format="<lk>{time:YYYY-MM-DD HH:mm:ss}</lk> <lvl>{level:8}</lvl> | <m><u>{name}</u></m> <lvl>{message}</lvl>",
+)
+
+
+def _hidden_upsteam(record):
+    if record["name"].startswith("satori"):
+        record["name"] = "satori"
+    if record["name"].startswith("launart"):
+        record["name"] = "launart"
+
+
+logger.configure(patcher=_hidden_upsteam)
+log = LoggerManager()
+
+
+__all__ = ["log"]

arclet/entari/plugin/__init__.py

@@ -4,10 +4,10 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Callable
 
-from loguru import logger
 from tarina import init_spec
 
 from ..config import Config
+from ..logger import log
 from .model import Plugin
 from .model import PluginMetadata as PluginMetadata
 from .model import RegisterNotInPluginError, _current_plugin
@@ -47,9 +47,9 @@ def load_plugin(path: str, config: dict | None = None, recursive_guard: set[str]
     try:
         mod = import_plugin(path, config=config)
         if not mod:
-            logger.error(f"cannot found plugin {path!r}")
+            log.plugin.opt(colors=True).error(f"cannot found plugin <blue>{path!r}</blue>")
             return
-        logger.success(f"loaded plugin {path!r}")
+        log.plugin.opt(colors=True).success(f"loaded plugin <blue>{path!r}</blue>")
         if mod.__name__ in plugin_service._unloaded:
             if mod.__name__ in plugin_service._referents and plugin_service._referents[mod.__name__]:
                 referents = plugin_service._referents[mod.__name__].copy()
@@ -58,7 +58,7 @@ def load_plugin(path: str, config: dict | None = None, recursive_guard: set[str]
                     if referent in recursive_guard:
                         continue
                     if referent in plugin_service.plugins:
-                        logger.debug(f"reloading {mod.__name__}'s referent {referent!r}")
+                        log.plugin.opt(colors=True).debug(f"reloading <y>{mod.__name__}</y>'s referent <y>{referent!r}</y>")
                         dispose(referent)
                         if not load_plugin(referent):
                             plugin_service._referents[mod.__name__].add(referent)
@@ -67,9 +67,9 @@ def load_plugin(path: str, config: dict | None = None, recursive_guard: set[str]
             plugin_service._unloaded.discard(mod.__name__)
         return mod.__plugin__
     except RegisterNotInPluginError as e:
-        logger.exception(f"{e.args[0]}", exc_info=e)
+        log.plugin.opt(colors=True).error(f"{e.args[0]}")
     except Exception as e:
-        logger.exception(f"failed to load plugin {path!r} caused by {e!r}", exc_info=e)
+        log.plugin.opt(colors=True).exception(f"failed to load plugin <blue>{path!r}</blue> caused by {e!r}", exc_info=e)
 
 
 def load_plugins(dir_: str | PathLike | Path):

arclet/entari/plugin/model.py

@@ -13,9 +13,9 @@
 from arclet.letoderea.typing import TTarget
 from creart import it
 from launart import Launart, Service
-from loguru import logger
 from satori.client import Account
 
+from ..logger import log
 from .service import PluginLifecycleService, plugin_service
 
 if TYPE_CHECKING:
@@ -216,14 +216,14 @@ def dispose(self):
         if self.subplugins:
             subplugs = [i.removeprefix(self.id)[1:] for i in self.subplugins]
             subplugs = (subplugs[:3] + ["..."]) if len(subplugs) > 3 else subplugs
-            logger.debug(f"disposing sub-plugin {', '.join(subplugs)} of {self.id}")
+            log.plugin.opt(colors=True).debug(f"disposing sub-plugin {', '.join(subplugs)} of <y>{self.id}</y>")
             for subplug in self.subplugins:
                 if subplug not in plugin_service.plugins:
                     continue
                 try:
                     plugin_service.plugins[subplug].dispose()
                 except Exception as e:
-                    logger.error(f"failed to dispose sub-plugin {subplug} caused by {e!r}")
+                    log.plugin.opt(colors=True).error(f"failed to dispose sub-plugin <y>{subplug}</y> caused by {e!r}")
                     plugin_service.plugins.pop(subplug, None)
             self.subplugins.clear()
         for disp in self.dispatchers.values():

arclet/entari/plugin/service.py

@@ -3,7 +3,8 @@
 
 from launart import Launart, Service, any_completed
 from launart.status import Phase
-from loguru import logger
+
+from ..logger import log
 
 if TYPE_CHECKING:
     from .model import KeepingVariable, Plugin
@@ -118,11 +119,11 @@ async def launch(self, manager: Launart):
             ids = [k for k in self.plugins.keys() if k not in self._subplugined]
             for plug_id in ids:
                 plug = self.plugins[plug_id]
-                logger.debug(f"disposing plugin {plug.id}")
+                log.plugin.opt(colors=True).debug(f"disposing plugin <y>{plug.id}</y>")
                 try:
                     plug.dispose()
                 except Exception as e:
-                    logger.error(f"failed to dispose plugin {plug.id} caused by {e!r}")
+                    log.plugin.opt(colors=True).error(f"failed to dispose plugin <y>{plug.id}</y> caused by {e!r}")
                     self.plugins.pop(plug_id, None)
             for values in self._keep_values.values():
                 for value in values.values():

example.yml

@@ -5,6 +5,7 @@ basic:
       port: 5140
       path: "satori"
   ignore_self_message: true
+  log_level: "info"
 plugins:
   ::auto_reload:
     watch_dirs: ["."]