AutoScraperX/core/utils/log/logger_manager.py

from core.utils.log.aliyun_log import AliyunLogger
from core.utils.log.local_log import Local
from loguru._logger import Logger as LoguruLogger


class LoggerManager:
    """
    日志管理器
    """
    _local_loggers: dict[str, LoguruLogger] = {}
    _aliyun_loggers: dict[str, AliyunLogger] = {}

    @staticmethod
    def get_logger(
        platform: str = "system",
        mode: str = "crawler",
        log_to_console: bool = False
    ) -> LoguruLogger:
        key = f"{platform}_{mode}"
        if key not in LoggerManager._local_loggers:
            LoggerManager._local_loggers[key] = Local.init_logger(
                platform=platform,
                mode=mode,
                log_to_console=log_to_console
            )
        return LoggerManager._local_loggers[key]

    @staticmethod
    def get_aliyun_logger(
        platform: str = "system",
        mode: str = "crawler",
        env: str = "prod"
    ) -> AliyunLogger:
        key = f"{platform}_{mode}"
        if key not in LoggerManager._aliyun_loggers:
            LoggerManager._aliyun_loggers[key] = AliyunLogger(
                platform=platform,
                mode=mode,
                env=env
            )
        return LoggerManager._aliyun_loggers[key]

    def log_event(
        self,
        code: str,
        message: str,
        data: dict[str, Any],
        trace_id: str = "",
        platform: str = "system",
        mode: str = "crawler",
        level: str = "info"
    ):
        logger = self.get_logger(platform=platform, mode=mode)
        aliyun_logger = self.get_aliyun_logger(platform=platform, mode=mode)

        # 本地记录
        getattr(logger, level.lower(), logger.info)(f"{code} | {message} | {data}")

        # 阿里云记录
        aliyun_logger.logging(code=code, message=message, data=data, trace_id=trace_id)