src/core/logger.py

# src/core/logger.py
import sys
import inspect
import os
import time
from loguru import logger

class Logger:
    def __init__(self, level="INFO", colorize=True, format=None, max_logs=100):
        """
        初始化 Logger 实例。

        Args:
            level (str): 日志级别.
            colorize (bool): 是否彩色输出.
            format (str): 日志格式.
            max_logs (int): 存储的最大日志条数.
        """
        self.level = level # 存储日志级别
        logger.remove()
        if format is None:
            format = (
                "{time:YYYY-MM-DD HH:mm:ss} | "
                "{level: <8} | "
                "{extra[filename]}:{extra[function]}:{extra[lineno]} | "
                "{message}"
            )
        logger.add(
            sys.stderr,
            level=level,
            format=format,
            colorize=colorize,
            backtrace=True,
            diagnose=True
        )
        self.logger = logger
        self._log_history = [] # 存储日志历史
        self._max_logs = max_logs # 最大日志条数

    def _add_log_to_history(self, message, level, source):
        """将日志添加到历史记录"""
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        log_entry = f"{timestamp} | {level.upper(): <8} | [{source}] {message}"
        self._log_history.append(log_entry)
        # 保持日志历史在最大条数内
        if len(self._log_history) > self._max_logs:
            self._log_history = self._log_history[-self._max_logs:]

    def get_recent_logs(self):
        """获取最近的日志历史"""
        return "\n".join(self._log_history)

    def _get_caller_info(self):
        frame = inspect.currentframe()
        try:
            caller_frame = frame.f_back.f_back
            full_path = caller_frame.f_code.co_filename
            function = caller_frame.f_code.co_name
            lineno = caller_frame.f_lineno
            filename = os.path.basename(full_path)
            return {
                'filename': filename,
                'function': function,
                'lineno': lineno
            }
        finally:
            del frame

    def info(self, message, source="API"):
        caller_info = self._get_caller_info()
        log_message = f"[{source}] {message}"
        self.logger.bind(**caller_info).info(log_message)
        self._add_log_to_history(message, "INFO", source)

    def error(self, message, source="API"):
        caller_info = self._get_caller_info()
        if isinstance(message, Exception):
            log_message = f"[{source}] {str(message)}"
            self.logger.bind(**caller_info).exception(log_message)
        else:
            log_message = f"[{source}] {message}"
            self.logger.bind(**caller_info).error(log_message)
        self._add_log_to_history(message, "ERROR", source)


    def warning(self, message, source="API"):
        caller_info = self._get_caller_info()
        log_message = f"[{source}] {message}"
        self.logger.bind(**caller_info).warning(log_message)
        self._add_log_to_history(message, "WARNING", source)

    def debug(self, message, source="API"):
        caller_info = self._get_caller_info()
        log_message = f"[{source}] {message}"
        self.logger.bind(**caller_info).debug(log_message)
        self._add_log_to_history(message, "DEBUG", source)


    async def request_logger(self, request):
        caller_info = self._get_caller_info()
        log_message = f"请求: {request.method} {request.path}"
        self.logger.bind(**caller_info).info(log_message, "Request")
        self._add_log_to_history(log_message, "INFO", "Request")


# 初始化全局 logger 实例
logger = Logger(level="INFO")