Python实现日志备份数达10时自动压缩清理的问题

你的需求是当日志备份文件达到10个时，批量压缩所有备份并删除原文件，之后继续生成新备份循环执行。现有代码的问题在于每次滚动都单独压缩当前日志，而非在备份数达标时批量处理。以下是修改后的实现：

修改后的代码 #

import os
import logging
import logging.handlers
import tarfile
import time
import glob

class CompressedBatchRotatingFileHandler(logging.handlers.RotatingFileHandler):
    def doRollover(self):
        # 关闭当前日志流，避免文件占用
        if self.stream:
            self.stream.close()
            self.stream = None

        # 收集所有当前的备份文件（匹配数字后缀，排除已压缩的包）
        backup_files = []
        for file_path in glob.glob(f"{self.baseFilename}.*"):
            suffix = file_path.split('.')[-1]
            if suffix.isdigit():
                backup_files.append(file_path)
        
        # 按后缀数字排序，确保顺序正确
        backup_files.sort(key=lambda x: int(x.split('.')[-1]))

        # 当备份文件数量达到设定值时，批量压缩并删除
        if len(backup_files) >= self.backupCount:
            # 生成带时间戳的压缩包文件名，避免重复
            timestamp = time.strftime('%Y%m%d%H%M%S')
            archive_name = f"{self.baseFilename}.{timestamp}.tar.gz"
            
            # 创建压缩包并添加所有备份文件
            with tarfile.open(archive_name, 'w:gz') as tar:
                for file in backup_files:
                    tar.add(file, arcname=os.path.basename(file))
            
            # 删除原备份文件
            for file in backup_files:
                os.remove(file)

        # 执行常规的日志滚动（生成新的备份文件）
        if self.backupCount > 0:
            # 处理剩余备份的滚动逻辑（未达10个时）
            for i in range(len(backup_files)-1, 0, -1):
                src = f"{self.baseFilename}.{i}"
                dst = f"{self.baseFilename}.{i+1}"
                if os.path.exists(src):
                    if os.path.exists(dst):
                        os.remove(dst)
                    os.rename(src, dst)
            # 重命名当前主日志为.1
            dst = f"{self.baseFilename}.1"
            if os.path.exists(dst):
                os.remove(dst)
            if os.path.exists(self.baseFilename):
                os.rename(self.baseFilename, dst)

        # 重新打开主日志文件
        self.mode = 'w'
        self.stream = self._open()

if __name__ == '__main__':
    log_file = 'test.log'
    # 初始化logger，移除默认handler避免重复日志
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)
    for handler in logger.handlers[:]:
        logger.removeHandler(handler)
    
    # 添加自定义handler
    handler = CompressedBatchRotatingFileHandler(
        log_file,
        maxBytes=2000,  # 日志文件达到2000字节时滚动
        backupCount=10  # 备份文件数量阈值
    )
    formatter = logging.Formatter("%(asctime)-15s %(levelname)-8s %(message)s")
    handler.setFormatter(formatter)
    logger.addHandler(handler)

    # 测试日志生成
    count = 0
    while True:
        logger.info(f"Test log entry #{count}")
        count += 1
        time.sleep(0.001)  # 控制日志生成速度，方便观察

关键修改说明 #

• 批量压缩触发逻辑：在每次滚动时先统计现有数字后缀的备份文件数量，当数量达到backupCount（10）时，将所有备份打包成带时间戳的tar.gz压缩包，避免覆盖旧压缩包。
• 备份文件清理：压缩完成后删除所有原备份文件，释放磁盘空间。
• 修复重复日志问题：测试代码中移除了logging.basicConfig的filename参数，并清理默认handler，避免同时存在多个FileHandler导致日志重复。
• 日志滚动逻辑适配：删除备份后，重新执行滚动时会从.1开始生成新的备份文件，实现循环逻辑。

内容的提问来源于stack exchange，提问作者ditil