AutoScraperX/core/utils/log_codes.py

# core/utils/log_codes.py
"""
阿里云日志码统一管理
采用分层分类的方式管理所有日志码
"""

from enum import Enum
from typing import Dict, Any


class LogCategory(Enum):
    """日志分类"""
    SYSTEM = "系统"  # 系统相关日志
    TASK = "任务"  # 任务处理日志
    DATA = "数据"  # 数据处理日志
    ERROR = "错误"  # 错误日志
    MONITOR = "监控"  # 监控日志


class LogCodeManager:
    """日志码管理器"""

    # 系统日志码 (1000-1999)
    SYSTEM_START = ("1000", "系统启动", LogCategory.SYSTEM)
    SYSTEM_SHUTDOWN = ("1001", "系统关闭", LogCategory.SYSTEM)
    SYSTEM_HEALTHY = ("1002", "系统健康检查", LogCategory.SYSTEM)

    # 任务日志码 (2000-2999)
    TASK_RECEIVED = ("2000", "任务接收成功", LogCategory.TASK)
    TASK_START = ("2001", "任务开始执行", LogCategory.TASK)
    TASK_SUCCESS = ("2010", "任务执行成功", LogCategory.TASK)
    TASK_FAILED = ("2011", "任务执行失败", LogCategory.TASK)

    # 数据处理日志码 (3000-3999)
    DATA_PROCESS_START = ("3000", "开始处理数据", LogCategory.DATA)
    DATA_PROCESS_SUCCESS = ("3001", "数据处理成功", LogCategory.DATA)
    DATA_VALIDATION_FAILED = ("3002", "数据验证失败", LogCategory.DATA)
    DATA_TRANSFORM_SUCCESS = ("3003", "数据转换成功", LogCategory.DATA)

    # 错误日志码 (9000-9999)
    ERROR_GENERAL = ("9000", "一般错误", LogCategory.ERROR)
    ERROR_NETWORK = ("9001", "网络错误", LogCategory.ERROR)
    ERROR_DATABASE = ("9002", "数据库错误", LogCategory.ERROR)
    ERROR_CONFIG = ("9003", "配置错误", LogCategory.ERROR)

    # 监控日志码 (1500-1599)
    MONITOR_PROCESS_START = ("1500", "进程启动", LogCategory.MONITOR)
    MONITOR_PROCESS_RESTART = ("1501", "进程重启", LogCategory.MONITOR)
    MONITOR_PROCESS_STOP = ("1502", "进程停止", LogCategory.MONITOR)

    @classmethod
    def get_code_info(cls, code: str) -> Dict[str, Any]:
        """根据code获取日志信息"""
        for attr_name in dir(cls):
            if not attr_name.startswith('_') and attr_name.isupper():
                code_info = getattr(cls, attr_name)
                if isinstance(code_info, tuple) and code_info[0] == code:
                    return {
                        "code": code_info[0],
                        "message": code_info[1],
                        "category": code_info[2].value
                    }
        return None

    @classmethod
    def get_codes_by_category(cls, category: LogCategory) -> list:
        """根据分类获取所有日志码"""
        codes = []
        for attr_name in dir(cls):
            if not attr_name.startswith('_') and attr_name.isupper():
                code_info = getattr(cls, attr_name)
                if isinstance(code_info, tuple) and code_info[2] == category:
                    codes.append({
                        "name": attr_name,
                        "code": code_info[0],
                        "message": code_info[1]
                    })
        return codes

    @classmethod
    def validate_code(cls, code: str) -> bool:
        """验证日志码是否存在"""
        return cls.get_code_info(code) is not None


# 全局实例
log_codes = LogCodeManager()