🎨 改进 logger

.env.example

@@ -32,6 +32,7 @@ ADMINS=[{ "username": "", "user_id": -1 }]
 # VERIFY_GROUPS=[]
 
 # logger 配置 可选配置项
+LOGGER_NAME="TGPaimon"
 # 打印时的宽度
 LOGGER_WIDTH=180
 # log 文件存放目录
@@ -45,6 +46,9 @@ LOGGER_TRACEBACK_MAX_FRAMES=20
 LOGGER_LOCALS_MAX_DEPTH=0
 LOGGER_LOCALS_MAX_LENGTH=10
 LOGGER_LOCALS_MAX_STRING=80
+# 可被 logger 打印的 record 的名称（默认包含了 LOGGER_NAME ）
+LOGGER_FILTERED_NAMES=["uvicorn"]
+
 
 # 超时配置 可选配置项
 # TIMEOUT = 10

core/config.py

@@ -48,6 +48,7 @@ class BotConfig(BaseSettings):
     verify_groups: List[Union[int, str]] = []
     join_groups: Optional[JoinGroups] = JoinGroups.NO_ALLOW
 
+    logger_name: str = "TGPaimon"
     logger_width: int = 180
     logger_log_path: str = "./logs"
     logger_time_format: str = "[%Y-%m-%d %X]"
@@ -56,6 +57,7 @@ class BotConfig(BaseSettings):
     logger_locals_max_depth: Optional[int] = 0
     logger_locals_max_length: int = 10
     logger_locals_max_string: int = 80
+    logger_filtered_names: List[str] = ["uvicorn"]
 
     timeout: int = 10
     read_timeout: float = 2
@@ -104,6 +106,7 @@ class BotConfig(BaseSettings):
     @property
     def logger(self) -> "LoggerConfig":
         return LoggerConfig(
+            name=self.logger_name,
             width=self.logger_width,
             traceback_max_frames=self.logger_traceback_max_frames,
             path=PROJECT_ROOT.joinpath(self.logger_log_path).resolve(),
@@ -112,6 +115,7 @@ class BotConfig(BaseSettings):
             locals_max_length=self.logger_locals_max_length,
             locals_max_string=self.logger_locals_max_string,
             locals_max_depth=self.logger_locals_max_depth,
+            filtered_names=self.logger_filtered_names,
         )
 
     @property
@@ -154,6 +158,7 @@ class RedisConfig(BaseModel):
 
 
 class LoggerConfig(BaseModel):
+    name: str = "TGPaimon"
     width: int = 180
     time_format: str = "[%Y-%m-%d %X]"
     traceback_max_frames: int = 20
@@ -162,6 +167,7 @@ class LoggerConfig(BaseModel):
     locals_max_length: int = 10
     locals_max_string: int = 80
     locals_max_depth: Optional[int] = None
+    filtered_names: List[str] = ["uvicorn"]
 
     @validator("locals_max_depth", pre=True, check_fields=False)
     def locals_max_depth_validator(cls, value) -> Optional[int]:  # pylint: disable=R0201

utils/log/__init__.py

@@ -1 +1,44 @@
-from utils.log._logger import *
+import re
+from functools import lru_cache
+from typing import TYPE_CHECKING
+
+from core.config import config
+from utils.log._config import LoggerConfig
+from utils.log._logger import LogFilter, Logger
+
+if TYPE_CHECKING:
+    from logging import LogRecord
+
+__all__ = ["logger"]
+
+logger = Logger(
+    LoggerConfig(
+        name=config.logger.name,
+        width=config.logger.width,
+        time_format=config.logger.time_format,
+        traceback_max_frames=config.logger.traceback_max_frames,
+        log_path=config.logger.path,
+        keywords=config.logger.render_keywords,
+        traceback_locals_max_depth=config.logger.locals_max_depth,
+        traceback_locals_max_length=config.logger.locals_max_length,
+        traceback_locals_max_string=config.logger_locals_max_string,
+    )
+)
+
+
+@lru_cache
+def _name_filter(record_name: str) -> bool:
+    for name in config.logger.filtered_names + [config.logger.name]:
+        if re.match(rf"^{name}.*?$", record_name):
+            return True
+    return False
+
+
+def name_filter(record: "LogRecord") -> bool:
+    """默认的过滤器"""
+    return _name_filter(record.name)
+
+
+log_filter = LogFilter()
+log_filter.add_filter(name_filter)
+logger.addFilter(log_filter)

utils/log/_config.py

@@ -0,0 +1,44 @@
+from multiprocessing import RLock as Lock
+from pathlib import Path
+from typing import (
+    List,
+    Optional,
+    Union,
+)
+
+from pydantic import BaseSettings
+
+from utils.const import PROJECT_ROOT
+
+__all__ = ["LoggerConfig"]
+
+
+class LoggerConfig(BaseSettings):
+    _lock = Lock()
+    _instance: Optional["LoggerConfig"] = None
+
+    def __new__(cls, *args, **kwargs) -> "LoggerConfig":
+        with cls._lock:
+            if cls._instance is None:
+                cls.update_forward_refs()
+                result = super(LoggerConfig, cls).__new__(cls)
+                result.__init__(*args, **kwargs)
+                cls._instance = result
+        return cls._instance
+
+    name: str = "logger"
+    level: Optional[Union[str, int]] = None
+
+    debug: bool = False
+    width: int = 180
+    keywords: List[str] = []
+    time_format: str = "[%Y-%m-%d %X]"
+    capture_warnings: bool = True
+
+    log_path: Union[str, Path] = "./logs"
+    project_root: Union[str, Path] = PROJECT_ROOT
+
+    traceback_max_frames: int = 20
+    traceback_locals_max_depth: Optional[int] = None
+    traceback_locals_max_length: int = 10
+    traceback_locals_max_string: int = 80

utils/log/_file.py

@@ -18,7 +18,7 @@ class FileIO(IO[str]):
         today = date.today()
         if self.file.exists():
             if not self.file.is_file():
-                raise RuntimeError(f'日志文件冲突, 请删除文件夹 "{str(self.file.resolve())}"')
+                raise FileExistsError(f'Log file conflict, please delete the folder "{str(self.file.resolve())}"')
             if self.file_stream is None or self.file_stream.closed:
                 self.file_stream = self.file.open(mode="a+", encoding="utf-8")
             modify_date = date.fromtimestamp(os.stat(self.file).st_mtime)

utils/log/_handler.py

@@ -3,12 +3,18 @@ import os
 import sys
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Callable, Iterable, List, Literal, Optional, TYPE_CHECKING, Union
+from typing import (
-
+    Any,
-import ujson as json
+    Callable,
-from rich.console import (
+    Iterable,
-    Console,
+    List,
+    Literal,
+    Optional,
+    TYPE_CHECKING,
+    Union,
 )
+
+from rich.console import Console
 from rich.logging import (
     LogRender as DefaultLogRender,
     RichHandler as DefaultRichHandler,
@@ -19,10 +25,7 @@ from rich.text import (
     TextType,
 )
 from rich.theme import Theme
-from ujson import JSONDecodeError
 
-from core.config import config
-from utils.const import PROJECT_ROOT
 from utils.log._file import FileIO
 from utils.log._style import DEFAULT_STYLE
 from utils.log._traceback import Traceback
@@ -45,24 +48,18 @@ if sys.platform == "win32":
     color_system = "windows"
 else:
     color_system = "truecolor"
-# noinspection SpellCheckingInspection
-log_console = Console(color_system=color_system, theme=Theme(DEFAULT_STYLE), width=config.logger.width)
 
 
 class LogRender(DefaultLogRender):
     @property
     def last_time(self):
+        """上次打印的时间"""
         return self._last_time
 
     @last_time.setter
     def last_time(self, last_time):
         self._last_time = last_time
 
-    def __init__(self, *args, **kwargs):
-        super(LogRender, self).__init__(*args, **kwargs)
-        self.show_level = True
-        self.time_format = config.logger.time_format
-
     def __call__(
         self,
         console: "Console",
@@ -87,7 +84,7 @@ class LogRender(DefaultLogRender):
             output.add_column(style="log.line_no", width=4)
         row: List["RenderableType"] = []
         if self.show_time:
-            log_time = log_time or log_console.get_datetime()
+            log_time = log_time or console.get_datetime()
             time_format = time_format or self.time_format
             if callable(time_format):
                 log_time_display = time_format(log_time)
@@ -108,7 +105,10 @@ class LogRender(DefaultLogRender):
             row.append(path_text)
 
         line_no_text = Text()
-        line_no_text.append(str(line_no), style=f"link file://{link_path}#{line_no}" if link_path else "")
+        line_no_text.append(
+            str(line_no),
+            style=f"link file://{link_path}#{line_no}" if link_path else "",
+        )
         row.append(line_no_text)
 
         output.add_row(*row)
@@ -119,16 +119,23 @@ class Handler(DefaultRichHandler):
     def __init__(
         self,
         *args,
+        width: int = None,
         rich_tracebacks: bool = True,
-        locals_max_depth: Optional[int] = config.logger.locals_max_depth,
+        locals_max_depth: Optional[int] = None,
+        tracebacks_max_frames: int = 100,
+        keywords: Optional[List[str]] = None,
+        log_time_format: Union[str, FormatTimeCallable] = "[%x %X]",
+        project_root: Union[str, Path] = "./logs",
         **kwargs,
     ) -> None:
         super(Handler, self).__init__(*args, rich_tracebacks=rich_tracebacks, **kwargs)
-        self._log_render = LogRender()
+        self._log_render = LogRender(time_format=log_time_format, show_level=True)
-        self.console = log_console
+        self.console = Console(color_system=color_system, theme=Theme(DEFAULT_STYLE), width=width)
         self.tracebacks_show_locals = True
-        self.keywords = self.KEYWORDS + config.logger.render_keywords
+        self.tracebacks_max_frames = tracebacks_max_frames
+        self.keywords = self.KEYWORDS + (keywords or [])
         self.locals_max_depth = locals_max_depth
+        self.project_root = project_root
 
     def render(
         self,
@@ -139,7 +146,7 @@ class Handler(DefaultRichHandler):
     ) -> "ConsoleRenderable":
         if record.pathname != "<input>":
             try:
-                path = str(Path(record.pathname).relative_to(PROJECT_ROOT))
+                path = str(Path(record.pathname).relative_to(self.project_root))
                 path = path.split(".")[0].replace(os.sep, ".")
             except ValueError:
                 import site
@@ -178,7 +185,11 @@ class Handler(DefaultRichHandler):
         )
         return log_renderable
 
-    def render_message(self, record: "LogRecord", message: Any) -> "ConsoleRenderable":
+    def render_message(
+        self,
+        record: "LogRecord",
+        message: Any,
+    ) -> "ConsoleRenderable":
         use_markup = getattr(record, "markup", self.markup)
         if isinstance(message, str):
             message_text = Text.from_markup(message) if use_markup else Text(message)
@@ -216,15 +227,16 @@ class Handler(DefaultRichHandler):
                 width=self.tracebacks_width,
                 extra_lines=self.tracebacks_extra_lines,
                 word_wrap=self.tracebacks_word_wrap,
-                show_locals=getattr(record, "show_locals", None) or self.tracebacks_show_locals,
+                show_locals=(getattr(record, "show_locals", None) or self.tracebacks_show_locals),
-                locals_max_length=getattr(record, "locals_max_length", None) or self.locals_max_length,
+                locals_max_length=(getattr(record, "locals_max_length", None) or self.locals_max_length),
-                locals_max_string=getattr(record, "locals_max_string", None) or self.locals_max_string,
+                locals_max_string=(getattr(record, "locals_max_string", None) or self.locals_max_string),
                 locals_max_depth=(
                     getattr(record, "locals_max_depth")
                     if hasattr(record, "locals_max_depth")
                     else self.locals_max_depth
                 ),
                 suppress=self.tracebacks_suppress,
+                max_frames=self.tracebacks_max_frames,
             )
             message = record.getMessage()
             if self.formatter:
@@ -237,10 +249,7 @@ class Handler(DefaultRichHandler):
                 message = None
 
         if message is not None:
-            try:
+            message_renderable = self.render_message(record, message)
-                message_renderable = self.render_message(record, json.loads(message))
-            except JSONDecodeError:
-                message_renderable = self.render_message(record, message)
         else:
             message_renderable = None
         log_renderable = self.render(record=record, traceback=_traceback, message_renderable=message_renderable)
@@ -252,7 +261,13 @@ class Handler(DefaultRichHandler):
 
 
 class FileHandler(Handler):
-    def __init__(self, *args, path: Path, **kwargs):
+    def __init__(
+        self,
+        *args,
+        width: int = None,
+        path: Path,
+        **kwargs,
+    ) -> None:
         super().__init__(*args, **kwargs)
         path.parent.mkdir(exist_ok=True, parents=True)
-        self.console = Console(width=180, file=FileIO(path), theme=Theme(DEFAULT_STYLE))
+        self.console = Console(width=width, file=FileIO(path), theme=Theme(DEFAULT_STYLE))

utils/log/_logger.py

@@ -5,25 +5,114 @@ import os
 import traceback as traceback_
 from multiprocessing import RLock as Lock
 from pathlib import Path
-from typing import Any, Callable, List, Mapping, Optional, TYPE_CHECKING, Tuple
+from types import TracebackType
+from typing import (
+    Any,
+    Callable,
+    List,
+    Mapping,
+    Optional,
+    TYPE_CHECKING,
+    Tuple,
+    Type,
+    Union,
+)
 
 from typing_extensions import Self
 
-from core.config import config
+from utils.log._handler import (
-from utils.const import NOT_SET
+    FileHandler,
-from utils.log._handler import FileHandler, Handler
+    Handler,
-from utils.typedefs import ExceptionInfoType
+)
+from utils.typedefs import LogFilterType
 
 if TYPE_CHECKING:
+    from utils.log._config import LoggerConfig  # pylint: disable=unused-import
     from logging import LogRecord  # pylint: disable=unused-import
 
-__all__ = ["logger"]
+__all__ = ["Logger", "LogFilter"]
+
+SysExcInfoType = Union[
+    Tuple[Type[BaseException], BaseException, Optional[TracebackType]],
+    Tuple[None, None, None],
+]
+ExceptionInfoType = Union[bool, SysExcInfoType, BaseException]
 
 _lock = Lock()
-__initialized__ = False
+NONE = object()
 
 
 class Logger(logging.Logger):
+    _instance: Optional["Logger"] = None
+
+    def __new__(cls, *args, **kwargs) -> "Logger":
+        with _lock:
+            if cls._instance is None:
+                result = super(Logger, cls).__new__(cls)
+                cls._instance = result
+        return cls._instance
+
+    def __init__(self, config: "LoggerConfig" = None) -> None:
+        from utils.log._config import LoggerConfig
+
+        self.config = config or LoggerConfig()
+
+        level_ = 10 if self.config.debug else 20
+        super().__init__(
+            name=self.config.name,
+            level=level_ if self.config.level is None else self.config.level,
+        )
+
+        log_path = Path(self.config.project_root).joinpath(self.config.log_path)
+        handler, debug_handler, error_handler = (
+            # 控制台 log 配置
+            Handler(
+                width=self.config.width,
+                locals_max_length=self.config.traceback_locals_max_length,
+                locals_max_string=self.config.traceback_locals_max_string,
+                locals_max_depth=self.config.traceback_locals_max_depth,
+                project_root=self.config.project_root,
+                log_time_format=self.config.time_format,
+            ),
+            # debug.log 配置
+            FileHandler(
+                width=self.config.width,
+                level=10,
+                path=log_path.joinpath("debug/debug.log"),
+                locals_max_depth=1,
+                locals_max_length=self.config.traceback_locals_max_length,
+                locals_max_string=self.config.traceback_locals_max_string,
+                project_root=self.config.project_root,
+                log_time_format=self.config.time_format,
+            ),
+            # error.log 配置
+            FileHandler(
+                width=self.config.width,
+                level=40,
+                path=log_path.joinpath("error/error.log"),
+                locals_max_length=self.config.traceback_locals_max_length,
+                locals_max_string=self.config.traceback_locals_max_string,
+                locals_max_depth=self.config.traceback_locals_max_depth,
+                project_root=self.config.project_root,
+                log_time_format=self.config.time_format,
+            ),
+        )
+        logging.basicConfig(
+            level=10 if self.config.debug else 20,
+            format="%(message)s",
+            datefmt=self.config.time_format,
+            handlers=[handler, debug_handler, error_handler],
+        )
+        if config.capture_warnings:
+            logging.captureWarnings(True)
+            warnings_logger = logging.getLogger("py.warnings")
+            warnings_logger.addHandler(handler)
+            warnings_logger.addHandler(debug_handler)
+
+        self.addHandler(handler)
+        self.addHandler(debug_handler)
+        self.addHandler(error_handler)
+
     def success(
         self,
         msg: Any,
@@ -33,11 +122,19 @@ class Logger(logging.Logger):
         stacklevel: int = 1,
         extra: Optional[Mapping[str, Any]] = None,
     ) -> None:
-        return self.log(25, msg, *args, exc_info=exc_info, stack_info=stack_info, stacklevel=stacklevel, extra=extra)
+        return self.log(
+            25,
+            msg,
+            *args,
+            exc_info=exc_info,
+            stack_info=stack_info,
+            stacklevel=stacklevel,
+            extra=extra,
+        )
 
     def exception(
         self,
-        msg: Any = NOT_SET,
+        msg: Any = NONE,
         *args: Any,
         exc_info: Optional[ExceptionInfoType] = True,
         stack_info: bool = False,
@@ -46,7 +143,7 @@ class Logger(logging.Logger):
         **kwargs,
     ) -> None:  # pylint: disable=W1113
         super(Logger, self).exception(
-            "" if msg is NOT_SET else msg,
+            "" if msg is NONE else msg,
             *args,
             exc_info=exc_info,
             stack_info=stack_info,
@@ -87,6 +184,10 @@ class Logger(logging.Logger):
             break
         return rv
 
+    def addFilter(self, log_filter: LogFilterType) -> None:  # pylint: disable=arguments-differ
+        for handler in self.handlers:
+            handler.addFilter(log_filter)
+
 
 class LogFilter(logging.Filter):
     _filter_list: List[Callable[["LogRecord"], bool]] = []
@@ -101,61 +202,3 @@ class LogFilter(logging.Filter):
 
     def filter(self, record: "LogRecord") -> bool:
         return all(map(lambda func: func(record), self._filter_list))
-
-
-def default_filter(record: "LogRecord") -> bool:
-    """默认的过滤器"""
-    return record.name.split(".")[0] in ["TGPaimon", "uvicorn"]
-
-
-with _lock:
-    if not __initialized__:
-        if "PYCHARM_HOSTED" in os.environ:
-            print()  # 针对 pycharm 的控制台 bug
-        logging.captureWarnings(True)
-        handler, debug_handler, error_handler = (
-            # 控制台 log 配置
-            Handler(
-                locals_max_length=config.logger.locals_max_length,
-                locals_max_string=config.logger.locals_max_string,
-                locals_max_depth=config.logger.locals_max_depth,
-            ),
-            # debug.log 配置
-            FileHandler(
-                level=10,
-                path=config.logger.path.joinpath("debug/debug.log"),
-                locals_max_depth=1,
-                locals_max_length=config.logger.locals_max_length,
-                locals_max_string=config.logger.locals_max_string,
-            ),
-            # error.log 配置
-            FileHandler(
-                level=40,
-                path=config.logger.path.joinpath("error/error.log"),
-                locals_max_length=config.logger.locals_max_length,
-                locals_max_string=config.logger.locals_max_string,
-                locals_max_depth=config.logger.locals_max_depth,
-            ),
-        )
-
-        default_log_filter = LogFilter().add_filter(default_filter)
-        handler.addFilter(default_log_filter)
-        debug_handler.addFilter(default_log_filter)
-
-        level_ = 10 if config.debug else 20
-        logging.basicConfig(
-            level=10 if config.debug else 20,
-            format="%(message)s",
-            datefmt=config.logger.time_format,
-            handlers=[handler, debug_handler, error_handler],
-        )
-        warnings_logger = logging.getLogger("py.warnings")
-        warnings_logger.addHandler(handler)
-        warnings_logger.addHandler(debug_handler)
-
-        logger = Logger("TGPaimon", level_)
-        logger.addHandler(handler)
-        logger.addHandler(debug_handler)
-        logger.addHandler(error_handler)
-
-        __initialized__ = True

utils/log/_traceback.py

@@ -1,6 +1,9 @@
 import os
 import traceback as traceback_
-from types import ModuleType, TracebackType
+from types import (
+    ModuleType,
+    TracebackType,
+)
 from typing import (
     Any,
     Dict,
@@ -40,10 +43,7 @@ from rich.traceback import (
     Traceback as BaseTraceback,
 )
 
-from core.config import config
+from utils.log._style import MonokaiProStyle
-from utils.log._style import (
-    MonokaiProStyle,
-)
 
 if TYPE_CHECKING:
     from rich.console import ConsoleRenderable  # pylint: disable=W0611
@@ -114,7 +114,8 @@ class Traceback(BaseTraceback):
     locals_max_depth: Optional[int]
 
     def __init__(self, *args, locals_max_depth: Optional[int] = None, **kwargs):
-        kwargs.update({"show_locals": True, "max_frames": config.logger.traceback_max_frames})
+
+        kwargs.update({"show_locals": True})
         super(Traceback, self).__init__(*args, **kwargs)
         self.locals_max_depth = locals_max_depth
 
@@ -128,11 +129,11 @@ class Traceback(BaseTraceback):
         extra_lines: int = 3,
         theme: Optional[str] = None,
         word_wrap: bool = False,
-        show_locals: bool = False,
+        show_locals: bool = True,
         indent_guides: bool = True,
-        locals_max_length: int = config.logger.locals_max_length,
+        locals_max_length: int = 10,
-        locals_max_string: int = config.logger.locals_max_string,
+        locals_max_string: int = 80,
-        locals_max_depth: Optional[int] = config.logger_locals_max_depth,
+        locals_max_depth: Optional[int] = None,
         suppress: Iterable[Union[str, ModuleType]] = (),
         max_frames: int = 100,
     ) -> "Traceback":
@@ -249,8 +250,7 @@ class Traceback(BaseTraceback):
                 traceback = cause.__traceback__
                 is_cause = False
                 continue
-            # No cover, code is reached but coverage doesn't recognize it.
+            break
-            break  # pragma: no cover
 
         trace = Trace(stacks=stacks)
         return trace

utils/typedefs.py

@@ -1,10 +1,20 @@
+from logging import Filter, LogRecord
 from pathlib import Path
 from types import TracebackType
-from typing import Any, Dict, Optional, Tuple, Type, Union
+from typing import Any, Callable, Dict, Optional, Tuple, Type, Union
 
 from httpx import URL
 
-__all__ = ["StrOrPath", "StrOrURL", "StrOrInt", "SysExcInfoType", "ExceptionInfoType", "JSONDict", "JSONType"]
+__all__ = [
+    "StrOrPath",
+    "StrOrURL",
+    "StrOrInt",
+    "SysExcInfoType",
+    "ExceptionInfoType",
+    "JSONDict",
+    "JSONType",
+    "LogFilterType",
+]
 
 StrOrPath = Union[str, Path]
 StrOrURL = Union[str, URL]
@@ -14,3 +24,5 @@ SysExcInfoType = Union[Tuple[Type[BaseException], BaseException, Optional[Traceb
 ExceptionInfoType = Union[bool, SysExcInfoType, BaseException]
 JSONDict = Dict[str, Any]
 JSONType = Union[JSONDict, list]
+
+LogFilterType = Union[Filter, Callable[[LogRecord], int]]