AutoScraperX/core/utils/log/local_log.py

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": (
                "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | "
                "<level>{level}</level> | "
                "<cyan>{extra[platform]}</cyan> | "
                "<cyan>{extra[mode]}</cyan> | "
                "<magenta>trace_id={extra[trace_id]}</magenta> | "
                "<cyan>{name}:{function}:{line}</cyan> | "
                "{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()
        )