app/core/logger.py

"""全局日志模块 - 单例模式的日志管理器"""

import sys
import logging
from pathlib import Path
from logging.handlers import RotatingFileHandler

from app.core.config import setting


# 过滤模式
FILTER_PATTERNS = [
    "chunk: b'",  # SSE原始字节
    "Got event:",  # SSE事件
    "Closing",  # SSE关闭
]


class MCPLogFilter(logging.Filter):
    """MCP日志过滤器 - 过滤大量数据的DEBUG日志"""

    def filter(self, record: logging.LogRecord) -> bool:
        """过滤日志"""
        # 过滤SSE的DEBUG日志
        if record.name == "sse_starlette.sse" and record.levelno == logging.DEBUG:
            msg = record.getMessage()
            return not any(p in msg for p in FILTER_PATTERNS)

        # 过滤MCP streamable_http的DEBUG日志
        if "mcp.server.streamable_http" in record.name and record.levelno == logging.DEBUG:
            return False

        return True


class LoggerManager:
    """日志管理器（单例）"""

    _instance = None
    _initialized = False

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self):
        """初始化日志系统"""
        if LoggerManager._initialized:
            return

        # 配置
        log_dir = Path(__file__).parents[2] / "logs"
        log_dir.mkdir(exist_ok=True)
        log_level = setting.global_config.get("log_level", "INFO").upper()
        log_format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
        log_file = log_dir / "app.log"

        # 根日志器
        self.logger = logging.getLogger()
        self.logger.setLevel(log_level)

        # 避免重复添加
        if self.logger.handlers:
            return

        # 格式器和过滤器
        formatter = logging.Formatter(log_format)
        mcp_filter = MCPLogFilter()

        # 控制台处理器
        console = logging.StreamHandler(sys.stdout)
        console.setLevel(log_level)
        console.setFormatter(formatter)
        console.addFilter(mcp_filter)

        # 文件处理器（10MB，5个备份）
        file_handler = RotatingFileHandler(
            log_file, maxBytes=10*1024*1024, backupCount=5, encoding="utf-8"
        )
        file_handler.setLevel(log_level)
        file_handler.setFormatter(formatter)
        file_handler.addFilter(mcp_filter)

        # 添加处理器
        self.logger.addHandler(console)
        self.logger.addHandler(file_handler)

        # 配置第三方库
        self._configure_third_party()

        LoggerManager._initialized = True
    
    def _configure_third_party(self):
        """配置第三方库日志级别"""
        config = {
            "asyncio": logging.WARNING,
            "uvicorn": logging.INFO,
            "fastapi": logging.INFO,
            "aiomysql": logging.WARNING,
            "mcp": logging.CRITICAL,
            "fastmcp": logging.CRITICAL,
        }
        
        for name, level in config.items():
            logging.getLogger(name).setLevel(level)

    def debug(self, msg: str) -> None:
        """调试日志"""
        self.logger.debug(msg)

    def info(self, msg: str) -> None:
        """信息日志"""
        self.logger.info(msg)

    def warning(self, msg: str) -> None:
        """警告日志"""
        self.logger.warning(msg)

    def error(self, msg: str) -> None:
        """错误日志"""
        self.logger.error(msg)

    def critical(self, msg: str) -> None:
        """严重错误日志"""
        self.logger.critical(msg)


# 全局实例
logger = LoggerManager()