116 lines
3.7 KiB
Python
116 lines
3.7 KiB
Python
"""
|
||
结构化日志配置
|
||
|
||
使用 structlog 实现结构化、可观测的日志，便于生产环境排查与聚合。
|
||
"""
|
||
|
||
import logging
|
||
import sys
|
||
import json
|
||
from typing import Any
|
||
|
||
# 尝试使用 structlog，若未安装则回退到标准 logging
|
||
try:
|
||
import structlog
|
||
|
||
USE_STRUCTLOG = True
|
||
except ImportError:
|
||
USE_STRUCTLOG = False
|
||
|
||
|
||
def configure_logging(debug: bool = False) -> None:
|
||
"""
|
||
配置应用日志
|
||
|
||
Args:
|
||
debug: 是否为调试模式
|
||
"""
|
||
log_level = logging.DEBUG if debug else logging.INFO
|
||
|
||
if USE_STRUCTLOG:
|
||
structlog.configure(
|
||
processors=[
|
||
structlog.stdlib.filter_by_level,
|
||
structlog.stdlib.add_logger_name,
|
||
structlog.stdlib.add_log_level,
|
||
structlog.stdlib.PositionalArgumentsFormatter(),
|
||
structlog.processors.TimeStamper(fmt="iso"),
|
||
structlog.processors.StackInfoRenderer(),
|
||
structlog.processors.format_exc_info,
|
||
structlog.processors.UnicodeDecoder(),
|
||
structlog.dev.ConsoleRenderer() if debug else structlog.processors.JSONRenderer(),
|
||
],
|
||
wrapper_class=structlog.stdlib.BoundLogger,
|
||
context_class=dict,
|
||
logger_factory=structlog.stdlib.LoggerFactory(),
|
||
cache_logger_on_first_use=True,
|
||
)
|
||
else:
|
||
logging.basicConfig(
|
||
level=log_level,
|
||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||
stream=sys.stdout,
|
||
)
|
||
|
||
|
||
class _StdLoggerCompat:
|
||
"""兼容 structlog 风格 kwargs 的标准 logging 适配器。"""
|
||
|
||
_PASSTHROUGH_KEYS = {"exc_info", "stack_info", "stacklevel", "extra"}
|
||
|
||
def __init__(self, logger: logging.Logger):
|
||
self._logger = logger
|
||
|
||
def _split_kwargs(self, kwargs: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
|
||
passthrough: dict[str, Any] = {}
|
||
fields: dict[str, Any] = {}
|
||
for key, value in kwargs.items():
|
||
if key in self._PASSTHROUGH_KEYS:
|
||
passthrough[key] = value
|
||
else:
|
||
fields[key] = value
|
||
return passthrough, fields
|
||
|
||
@staticmethod
|
||
def _format_fields(fields: dict[str, Any]) -> str:
|
||
try:
|
||
return json.dumps(fields, ensure_ascii=False, default=str)
|
||
except Exception:
|
||
return str(fields)
|
||
|
||
def _log(self, method: str, msg: Any, *args: Any, **kwargs: Any) -> None:
|
||
passthrough, fields = self._split_kwargs(kwargs)
|
||
text = str(msg)
|
||
if fields:
|
||
text = f"{text} | {self._format_fields(fields)}"
|
||
getattr(self._logger, method)(text, *args, **passthrough)
|
||
|
||
def debug(self, msg: Any, *args: Any, **kwargs: Any) -> None:
|
||
self._log("debug", msg, *args, **kwargs)
|
||
|
||
def info(self, msg: Any, *args: Any, **kwargs: Any) -> None:
|
||
self._log("info", msg, *args, **kwargs)
|
||
|
||
def warning(self, msg: Any, *args: Any, **kwargs: Any) -> None:
|
||
self._log("warning", msg, *args, **kwargs)
|
||
|
||
def error(self, msg: Any, *args: Any, **kwargs: Any) -> None:
|
||
self._log("error", msg, *args, **kwargs)
|
||
|
||
def exception(self, msg: Any, *args: Any, **kwargs: Any) -> None:
|
||
kwargs.setdefault("exc_info", True)
|
||
self._log("error", msg, *args, **kwargs)
|
||
|
||
def critical(self, msg: Any, *args: Any, **kwargs: Any) -> None:
|
||
self._log("critical", msg, *args, **kwargs)
|
||
|
||
def __getattr__(self, item: str) -> Any:
|
||
return getattr(self._logger, item)
|
||
|
||
|
||
def get_logger(name: str):
|
||
"""获取 logger 实例"""
|
||
if USE_STRUCTLOG:
|
||
return structlog.get_logger(name)
|
||
return _StdLoggerCompat(logging.getLogger(name))
|