import sys
from datetime import datetime
from pathlib import Path
from loguru import logger as global_logger
from config import settings
from core.utils.trace_utils import get_current_trace_id
class Local:
"""
本地日志记录器
- 支持文件和控制台共享配置
- 自动管理日志处理器生命周期
- 确保日志格式一致性
"""
_initialized_loggers = {} # 存储已初始化的日志器配置
@staticmethod
def _get_base_config(platform: str, mode: str) -> dict:
"""获取基础日志配置(文件和控制台共享)"""
return {
"format": (
"{time:YYYY-MM-DD HH:mm:ss.SSS} | "
"{level} | "
"{extra[platform]} | "
"{extra[mode]} | "
"trace_id={extra[trace_id]} | "
"{name}:{function}:{line} | "
"{message}"
),
"backtrace": True, # 发生异常时显示完整堆栈回溯
"diagnose": True, # 发生异常时显示变量值
"enqueue": True, # 使用队列异步处理日志,提高性能
"level": "INFO", # 默认级别,将被实际配置覆盖
# 日志过滤设置
"filter": lambda record: record["extra"].get("platform") == platform
and record["extra"].get("mode") == mode
}
@staticmethod
def init_logger(
platform: str,
mode: str,
log_level: str = "INFO",
log_to_console: bool = False,
retention: str = "10 days",
rotation: str = "00:00",
compression: str = "gz",
):
"""初始化并返回带上下文的日志实例"""
key = f"{platform}_{mode}"
# 如果已初始化,直接返回
if key in Local._initialized_loggers:
return global_logger.bind(
platform=platform,
mode=mode,
trace_id=get_current_trace_id()
)
# 创建日志存放路径
log_dir = Path(f"{settings.LOG_DIR}/{platform}")
log_dir.mkdir(parents=True, exist_ok=True)
# 日志文件模式
log_pattern = log_dir / f"{platform}-{mode}-{{time:YYYY-MM-DD}}.log"
# 获取基础配置
base_config = Local._get_base_config(platform, mode)
base_config["level"] = log_level.upper()
# 清理现有处理器(避免重复)
global_logger.remove()
# 添加文件日志处理器(继承基础配置)
file_config = {
**base_config,
"sink": str(log_pattern),
"rotation": rotation,
"retention": retention,
"compression": compression,
"encoding": "utf-8"
}
global_logger.add(**file_config)
# 添加控制台日志处理器(如果需要,继承基础配置并添加颜色)
if log_to_console:
console_config = {
**base_config,
"sink": sys.stdout,
"colorize": True
}
global_logger.add(**console_config)
# 标记为已初始化
Local._initialized_loggers[key] = True
# 返回绑定上下文的日志器
return global_logger.bind(
platform=platform,
mode=mode,
trace_id=get_current_trace_id()
)