Python脚本如何按大小切割日志文件

wen python案例 2026-06-27 241

本文目录导读：

Python脚本如何按大小切割日志文件

方法1：使用内置的 RotatingFileHandler
方法2：自定义切割函数
方法3：基于时间的按大小切割
方法4：支持压缩的日志切割
关键参数说明

我来介绍几种按大小切割Python日志文件的方法：

方法1：使用内置的 RotatingFileHandler

import logging
from logging.handlers import RotatingFileHandler
# 创建日志记录器
logger = logging.getLogger('my_logger')
logger.setLevel(logging.INFO)
# 创建 RotatingFileHandler
# 参数：文件名，最大大小（字节），备份文件数量
handler = RotatingFileHandler(
    'app.log',           # 日志文件名
    maxBytes=1024*1024,  # 1MB
    backupCount=5        # 保留5个备份文件
)
# 设置格式
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
# 添加处理器
logger.addHandler(handler)
# 使用示例
for i in range(1000):
    logger.info(f'这是第 {i} 条日志消息')

方法2：自定义切割函数

import os
import logging
class SizeRotatingFileHandler:
    def __init__(self, filename, max_size_mb=10, backup_count=5):
        self.filename = filename
        self.max_size = max_size_mb * 1024 * 1024  # 转换为字节
        self.backup_count = backup_count
    def should_rotate(self):
        """检查是否需要切割"""
        if os.path.exists(self.filename):
            size = os.path.getsize(self.filename)
            return size >= self.max_size
        return False
    def rotate(self):
        """执行文件切割"""
        if self.backup_count > 0:
            # 删除最旧的备份文件
            oldest_file = f"{self.filename}.{self.backup_count}"
            if os.path.exists(oldest_file):
                os.remove(oldest_file)
            # 重命名现有备份文件
            for i in range(self.backup_count - 1, 0, -1):
                src = f"{self.filename}.{i}"
                dst = f"{self.filename}.{i + 1}"
                if os.path.exists(src):
                    os.rename(src, dst)
            # 重命名当前日志文件
            os.rename(self.filename, f"{self.filename}.1")
    def write_log(self, message):
        """写入日志并检查是否需要切割"""
        # 检查是否需要切割
        if self.should_rotate():
            self.rotate()
        # 写入日志
        with open(self.filename, 'a', encoding='utf-8') as f:
            f.write(f"{message}\n")
# 使用示例
logger = SizeRotatingFileHandler('custom_app.log', max_size_mb=1, backup_count=3)
for i in range(5000):
    logger.write_log(f'日志消息 #{i}')

方法3：基于时间的按大小切割

import logging
from logging.handlers import TimedRotatingFileHandler
import time
def get_log_size(filename):
    """获取日志文件大小"""
    try:
        return os.path.getsize(filename)
    except:
        return 0
class SmartRotatingHandler(logging.Handler):
    def __init__(self, filename, max_size=10*1024*1024, backup_count=5):
        super().__init__()
        self.filename = filename
        self.max_size = max_size
        self.backup_count = backup_count
    def emit(self, record):
        """重写emit方法"""
        log_entry = self.format(record)
        # 检查文件大小
        if os.path.exists(self.filename):
            current_size = os.path.getsize(self.filename)
            if current_size + len(log_entry) > self.max_size:
                self.rotate()
        # 写入日志
        with open(self.filename, 'a', encoding='utf-8') as f:
            f.write(log_entry + '\n')
    def rotate(self):
        """执行旋转"""
        for i in range(self.backup_count - 1, 0, -1):
            src = f"{self.filename}.{i}"
            dst = f"{self.filename}.{i+1}"
            if os.path.exists(src):
                os.rename(src, dst)
        if os.path.exists(self.filename):
            os.rename(self.filename, f"{self.filename}.1")
# 使用示例
logger = logging.getLogger('smart_logger')
logger.setLevel(logging.INFO)
handler = SmartRotatingHandler('smart_app.log', max_size=1024*1024, backup_count=3)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
# 写入大量日志
for i in range(10000):
    logger.info(f'智能日志切割测试 #{i}')

方法4：支持压缩的日志切割

import logging
import gzip
import shutil
from logging.handlers import RotatingFileHandler
class CompressedRotatingFileHandler(RotatingFileHandler):
    def __init__(self, filename, mode='a', maxBytes=0, backupCount=0, 
                 encoding=None, delay=False, compress=True):
        super().__init__(filename, mode, maxBytes, backupCount, encoding, delay)
        self.compress = compress
    def doRollover(self):
        """重写滚动方法，添加压缩功能"""
        if self.stream:
            self.stream.close()
            self.stream = None
        if self.backupCount > 0:
            # 压缩最旧的日志文件
            if self.compress:
                for i in range(self.backupCount - 1, 0, -1):
                    sfn = f"{self.baseFilename}.{i}"
                    if os.path.exists(sfn):
                        # 压缩文件
                        with open(sfn, 'rb') as f_in:
                            with gzip.open(f"{sfn}.gz", 'wb') as f_out:
                                shutil.copyfileobj(f_in, f_out)
                        os.remove(sfn)
                        # 重命名压缩文件
                        os.rename(f"{sfn}.gz", f"{self.baseFilename}.{i}.gz")
        # 执行父类的滚动操作
        super().doRollover()
# 使用示例
handler = CompressedRotatingFileHandler(
    'compressed_app.log',
    maxBytes=1024*1024,  # 1MB
    backupCount=5,
    compress=True
)
logger = logging.getLogger('compressed_logger')
logger.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
# 写入日志
for i in range(5000):
    logger.info(f'压缩日志测试 #{i}')