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() )