logging

Python的logging模块提供了灵活的日志记录功能，可用于调试和记录程序运行信息。

1 基本配置

将以下代码添加到Python文件中，以配置日志记录：

import logging
from pathlib import Path

# 配置日志文件路径
script_dir = Path(__file__).parent  # 获取当前脚本所在目录
log_file_path = script_dir / 'log_file.log'  # 在脚本目录下创建日志文件

# 配置日志
logging.basicConfig(
    level=logging.DEBUG,  # 设置日志级别
    format='%(asctime)s - %(levelname)s - %(message)s',  # 设置日志格式
    handlers=[
        logging.StreamHandler(),  # 输出到控制台
        logging.FileHandler(log_file_path, encoding='utf-8')  # 同时输出到文件
    ]
)
logger = logging.getLogger(__name__)  # 获取当前模块的logger

# 使用示例：不同级别的日志记录
logger.debug("这是调试信息")
logger.info("这是一般信息") 
logger.warning("这是警告信息")
logger.error("这是错误信息")
logger.critical("这是严重错误信息")

2 高级配置

以下是一个更复杂的日志配置示例，作为模块使用。该示例展示了如何创建一个通用的日志配置类，支持多模块共享日志设置、按大小轮转、压缩旧日志、不同级别日志分文件、控制台输出和自定义过滤器。

import logging
import os
import gzip
from logging.handlers import RotatingFileHandler
from typing import Optional, Dict, List
import shutil


class CompressedRotatingFileHandler(RotatingFileHandler):
    """支持压缩的日志轮转处理器"""

    def doRollover(self):
        """重写doRollover方法来添加压缩功能"""
        if self.stream:
            self.stream.close()
            self.stream = None

        # 如果备份文件已存在，需要先轮转这些文件
        if self.backupCount > 0:
            # 删除最旧的一个文件(如果存在)
            oldest_backup = f"{self.baseFilename}.{self.backupCount}.gz"
            if os.path.exists(oldest_backup):
                os.remove(oldest_backup)

            # 轮转现有的备份文件
            for i in range(self.backupCount - 1, 0, -1):
                source = f"{self.baseFilename}.{i}.gz"
                dest = f"{self.baseFilename}.{i + 1}.gz"
                if os.path.exists(source):
                    if os.path.exists(dest):
                        os.remove(dest)
                    os.rename(source, dest)

            # 压缩当前日志为第一个备份
            dest = f"{self.baseFilename}.1.gz"
            if os.path.exists(self.baseFilename):
                with open(self.baseFilename, 'rb') as f_in:
                    with gzip.open(dest, 'wb') as f_out:
                        shutil.copyfileobj(f_in, f_out)
                os.remove(self.baseFilename)

        # 创建新的空日志文件
        self.mode = 'w'
        self.stream = self._open()


class KeywordFilter(logging.Filter):
    def __init__(self, keyword: str):
        super().__init__()
        self.keyword = keyword

    def filter(self, record: logging.LogRecord) -> bool:
        # 仅当日志消息包含特定关键词时返回 True
        return self.keyword in record.msg


class LoggerConfig:
    """
    通用日志配置类，支持多模块共享日志设置
    特性：
    1. 按大小轮转
    2. 日志压缩
    3. 不同级别日志分文件
    4. 支持控制台输出
    5. 支持自定义过滤器
    """
    _instance = None
    _initialized = False
    _loggers: Dict[str, logging.Logger] = {}

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super(LoggerConfig, cls).__new__(cls)
        return cls._instance

    def __init__(self,
                 log_dir: str = './logs',
                 log_level: int = logging.INFO,
                 max_bytes: int = 5 * 1024 * 1024,  # 默认5MB
                 backup_count: int = 20,
                 enable_console: bool = True,
                 compress_logs: bool = True,
                 custom_filters: List[Dict] = None,
                 extra_handlers: List[logging.Handler] = None):
        """
        初始化日志配置

        Args:
            log_dir: 日志目录
            log_level: 基础日志级别
            max_bytes: 单个日志文件最大字节数
            backup_count: 保留的备份文件数量
            enable_console: 是否启用控制台输出
            compress_logs: 是否压缩旧日志
            custom_filters: 自定义过滤器列表
            extra_handlers: 额外的处理器列表
        """
        # 单例模式：确保只初始化一次
        if self._initialized:
            return

        self.log_dir = log_dir
        self.log_level = log_level
        self.max_bytes = max_bytes
        self.backup_count = backup_count
        self.enable_console = enable_console
        self.compress_logs = compress_logs
        self.custom_filters = custom_filters or []
        self.extra_handlers = extra_handlers or []

        # 创建日志目录
        os.makedirs(log_dir, exist_ok=True)

        # 默认格式化器
        self.default_formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )

        self._initialized = True

    def get_logger(self, name: str, formatter: Optional[logging.Formatter] = None) -> logging.Logger:
        """
        获取或创建logger

        Args:
            name: logger名称
            formatter: 自定义格式化器

        Returns:
            logging.Logger: 配置好的logger
        """
        # 如果已存在该logger，直接返回
        if name in self._loggers:
            return self._loggers[name]

        # 创建logger
        logger = logging.getLogger(name)

        # 清理可能存在的handlers
        if logger.hasHandlers():
            logger.handlers.clear()

        # 设置日志级别
        logger.setLevel(self.log_level)
        formatter = formatter or self.default_formatter

        # 添加控制台处理器
        if self.enable_console:
            console_handler = logging.StreamHandler()
            console_handler.setFormatter(formatter)
            logger.addHandler(console_handler)

        # 为不同级别创建单独的日志文件
        log_levels = [
            (logging.DEBUG, 'debug'),
            (logging.INFO, 'info'),
            (logging.WARNING, 'warning'),
            (logging.ERROR, 'error'),
            (logging.CRITICAL, 'critical')
        ]

        for level, level_name in log_levels:
            if level >= self.log_level:
                handler = self._create_rotating_handler(
                    os.path.join(self.log_dir, f"LOG_{level_name}.log"),
                    formatter,
                    level
                )
                logger.addHandler(handler)

        # 添加额外的处理器
        if self.extra_handlers:
            for handler in self.extra_handlers:
                handler.setFormatter(formatter)
                logger.addHandler(handler)

        # 添加自定义过滤器
        if self.custom_filters:
            for filter_config in self.custom_filters:
                filter_class = filter_config.get('filter_class')
                args = filter_config.get('args', {})
                if filter_class:
                    filter_instance = filter_class(**args)
                    logger.addFilter(filter_instance)

        # 禁止向父logger传递日志
        logger.propagate = False

        # 缓存logger实例
        self._loggers[name] = logger

        return logger

    def _create_rotating_handler(self,
                                 filename: str,
                                 formatter: logging.Formatter,
                                 level: int) -> RotatingFileHandler:
        """
        创建轮转处理器

        Args:
            filename: 日志文件名
            formatter: 格式化器
            level: 日志级别

        Returns:
            RotatingFileHandler: 配置好的处理器
        """
        handler_class = CompressedRotatingFileHandler if self.compress_logs else RotatingFileHandler

        handler = handler_class(
            filename,
            maxBytes=self.max_bytes,
            backupCount=self.backup_count,
            encoding='utf-8'
        )
        handler.setLevel(level)
        handler.setFormatter(formatter)

        return handler


def setLogConfig(module_name=None,
                 log_dir='./logs',
                 log_level=logging.DEBUG,
                 max_bytes=5 * 1024 * 1024,  # 5MB
                 backup_count=20,
                 enable_console=True,
                 compress_logs=True,
                 custom_filters=None,
                 extra_handlers=None):
    """
    快速设置并获取logger的便捷函数

    Args:
        module_name: 模块名称，用于标识日志来源
        log_dir: 日志保存目录
        log_level: 日志级别
        max_bytes: 单个日志文件最大字节数
        backup_count: 保留的备份文件数量
        enable_console: 是否启用控制台输出
        compress_logs: 是否压缩旧日志
        custom_filters: 自定义过滤器列表，例如custom_filters=[{'filter_class': KeywordFilter, 'args': {'keyword': 'critical'}}]
        extra_handlers: 额外的处理器列表

    Returns:
        logging.Logger: 配置好的logger实例
    """

    # 初始化日志配置
    log_config = LoggerConfig(
        log_dir=log_dir,
        log_level=log_level,
        max_bytes=max_bytes,
        backup_count=backup_count,
        enable_console=enable_console,
        compress_logs=compress_logs,
        custom_filters=custom_filters,
        extra_handlers=extra_handlers
    )

    # 自动获取调用者的模块名称
    if module_name is None:
        import inspect
        caller_frame = inspect.stack()[1]
        module_name = caller_frame.frame.f_globals['__name__']
        # 如果没有提供module_name，则通过调用栈获取调用者的__name__

    return log_config.get_logger(name=module_name, formatter=None)


# 使用示例
if __name__ == '__main__':
    logger = setLogConfig('test_module')

    # 使用logger
    logger.debug('这是一条调试日志')
    logger.info('这是一条信息日志')
    logger.warning('这是一条警告日志')
    logger.error('这是一条错误日志')
    logger.critical('这是一条严重错误日志')