diff --git a/pyproject.toml b/pyproject.toml index aafae9f..f964e2c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -130,7 +130,7 @@ disable = ["C0103", "R0903", "W0613"] [tool.pytest.ini_options] minversion = "7.0" -addopts = "-ra -q --strict-markers --strict-config" +addopts = "-ra -q --strict-markers --strict-config -p pytest_asyncio" testpaths = ["tests"] python_files = ["test_*.py"] python_classes = ["Test*"] diff --git a/src/log/log.py b/src/log/log.py index fa63c43..106afbc 100644 --- a/src/log/log.py +++ b/src/log/log.py @@ -1,14 +1,64 @@ +import logging import os import sys import json from datetime import date, datetime -from typing import Any, Dict +from typing import Any, Dict, Set from loguru import logger as loguru_logger from settings import settings +LOGGING_RESERVED_FIELDS: Set[str] = { + "name", + "msg", + "args", + "levelname", + "levelno", + "pathname", + "filename", + "module", + "exc_info", + "exc_text", + "stack_info", + "lineno", + "funcName", + "created", + "msecs", + "relativeCreated", + "thread", + "threadName", + "processName", + "process", +} + + +class InterceptHandler(logging.Handler): + """将标准 logging 日志转发到 loguru.""" + + def emit(self, record: logging.LogRecord) -> None: # pragma: no cover - 直接调用 + try: + level = loguru_logger.level(record.levelname).name + except ValueError: + level = record.levelno + + frame, depth = logging.currentframe(), 2 + while frame and frame.f_code.co_filename == logging.__file__: + frame = frame.f_back + depth += 1 + + extra = { + key: value + for key, value in record.__dict__.items() + if key not in LOGGING_RESERVED_FIELDS + } + + loguru_logger.bind(**extra).opt( + depth=depth, exception=record.exc_info + ).log(level, record.getMessage()) + + class LoggingConfig: """统一日志配置管理""" @@ -94,6 +144,20 @@ def setup_logger(self): # 清除默认处理器 loguru_logger.remove() + # 拦截标准 logging,统一输出格式 + intercept_handler = InterceptHandler() + logging.basicConfig(handlers=[intercept_handler], level=0, force=True) + + for logger_name in ( + "uvicorn", + "uvicorn.error", + "uvicorn.access", + "fastapi", + ): + standard_logger = logging.getLogger(logger_name) + standard_logger.handlers = [intercept_handler] + standard_logger.propagate = False + # 启用统一 patcher,确保所有日志输出为 JSON 结构 loguru_logger.configure(patcher=self._patch_record)