"""structlog 结构化日志(ARCH §9.3)。dev 走彩色 console,其余 JSON。""" from __future__ import annotations import logging import structlog from structlog.typing import Processor from ww_config import get_settings def configure_logging() -> None: settings = get_settings() shared: list[Processor] = [ structlog.contextvars.merge_contextvars, structlog.processors.add_log_level, structlog.processors.TimeStamper(fmt="iso"), ] renderer: Processor = ( structlog.processors.JSONRenderer() if settings.log_json else structlog.dev.ConsoleRenderer() ) structlog.configure( processors=[ *shared, # 交给 stdlib:ProcessorFormatter 在 root handler 上做最终渲染,统一结构化输出。 structlog.stdlib.ProcessorFormatter.wrap_for_formatter, ], logger_factory=structlog.stdlib.LoggerFactory(), wrapper_class=structlog.make_filtering_bound_logger(logging.INFO), cache_logger_on_first_use=True, ) # 接管 stdlib root:SQLAlchemy/httpx/uvicorn 等绕过 structlog 的日志也走同一 formatter # → 统一 JSON/console 输出(P2:日志接管 stdlib root)。 formatter = structlog.stdlib.ProcessorFormatter( foreign_pre_chain=shared, processors=[ structlog.stdlib.ProcessorFormatter.remove_processors_meta, renderer, ], ) handler = logging.StreamHandler() handler.setFormatter(formatter) root = logging.getLogger() root.handlers = [handler] root.setLevel(logging.INFO) def get_logger(name: str = "ww") -> structlog.stdlib.BoundLogger: logger: structlog.stdlib.BoundLogger = structlog.get_logger(name) return logger