MyTest/Utils/Log.py

import logging
import sys
import traceback
from datetime import datetime
from pathlib import Path
from Config.Config import Config


class Logger:
    def __init__(self, log_path: str = None, log_name: str = None):
        """
        初始化日志记录器
        Args:
            log_path: 日志存储路径，默认为项目根目录下的Log文件夹
            log_name: 日志文件名，默认为当前日期时间
        """
        # 设置默认日志路径为项目根目录下的Log文件夹

        if log_path is None:
            project_root = self._find_project_root()
            log_path = project_root / "Log"
        # 创建日志目录（如果不存在）
        self.log_path = Path(log_path)
        self.log_path.mkdir(parents=True, exist_ok=True)
        # 设置日志文件名
        if log_name is None:
            log_name = f"log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
        self.log_file = self.log_path / log_name
        # 配置日志记录器
        self.logger = logging.getLogger(__name__)
        self.logger.setLevel(Config.log_level)
        # 清除现有处理器（避免重复记录）
        self.logger.handlers.clear()
        # 创建格式化器
        formatter = logging.Formatter(Config.log_format)
        # 文件处理器（记录所有级别日志）
        file_handler = logging.FileHandler(self.log_file, encoding='utf-8')
        file_handler.setLevel(logging.DEBUG)
        file_handler.setFormatter(formatter)
        self.logger.addHandler(file_handler)
        # 控制台处理器（仅记录INFO及以上级别）
        console_handler = logging.StreamHandler()
        console_handler.setLevel(logging.INFO)
        console_handler.setFormatter(formatter)
        self.logger.addHandler(console_handler)
        # 设置异常钩子，捕获未处理的异常
        sys.excepthook = self._handle_exception
        # 重定向标准输出和标准错误
        self._redirect_stdout_stderr()

    def _find_project_root(self):
        """查找项目根目录"""
        # 查找包含.git、.project、setup.py或requirements.txt等标记文件的目录
        current_dir = Path(__file__).resolve().parent
        # 向上查找直到找到项目根目录标记
        for parent in [current_dir] + list(current_dir.parents):
            if any((parent / marker).exists() for marker in
                   ['.git', '.project', 'setup.py', 'requirements.txt', 'pyproject.toml']):
                return parent
        # 如果找不到标记文件，则使用当前工作目录
        return Path.cwd()

    def _redirect_stdout_stderr(self):
        """
        重定向标准输出和标准错误到日志
        :return:
        """   
        class StreamToLogger:
            def __init__(self, logger, level):
                self.logger = logger
                self.level = level
                self.linebuf = ''

            def write(self, buf):
                for line in buf.rstrip().splitlines():
                    if line.strip():  # 忽略空行
                        self.logger.log(self.level, line.rstrip())

            def flush(self):
                pass
        # 重定向标准输出到INFO级别
        sys.stdout = StreamToLogger(self.logger, logging.INFO)
        # 重定向标准错误到ERROR级别
        sys.stderr = StreamToLogger(self.logger, logging.ERROR)

    def _handle_exception(self, exc_type, exc_value, exc_traceback):
        """处理未捕获的异常"""
        # 忽略KeyboardInterrupt，以便控制台可以正常退出
        if issubclass(exc_type, KeyboardInterrupt):
            sys.__excepthook__(exc_type, exc_value, exc_traceback)
            return
        self.logger.error(
            "未捕获的异常:",
            exc_info=(exc_type, exc_value, exc_traceback)
        )

    def debug(self, msg):
        self.logger.debug(msg)

    def info(self, msg):
        self.logger.info(msg)

    def warning(self, msg):
        self.logger.warning(msg)

    def error(self, msg):
        self.logger.error(msg)

    def critical(self, msg):
        self.logger.critical(msg)

    def exception(self, msg, exc_info=True):
        """记录异常信息，包括堆栈跟踪"""
        self.logger.exception(msg, exc_info=exc_info)

    def log_exception(self, e, context=""):
        """记录异常及其上下文信息"""
        error_msg = f"{context}: {type(e).__name__}: {str(e)}"
        self.error(error_msg)
        self.error("堆栈跟踪:\n" + "".join(traceback.format_tb(e.__traceback__)))


# 使用示例
if __name__ == "__main__":
    # 创建日志记录器实例（使用默认路径）
    logger = Logger(log_name="my_app.log")
    # 记录不同级别的日志
    logger.debug("这是一条调试信息")
    logger.info("程序启动成功")
    logger.warning("磁盘空间不足")
    # 模拟一个异常
    try:
        result = 1 / 0
    except Exception as e:
        logger.log_exception(e, "除法运算时发生错误")
    # 模拟控制台输出
    print("这是一条普通的控制台输出")
    # 模拟控制台错误输出
    sys.stderr.write("这是一条错误输出\n")
    # 模拟未捕获的异常
    def cause_exception():
        raise ValueError("这是一个未捕获的异常")
    # 调用会抛出异常的函数
    cause_exception()
    logger.info("程序执行完成")