Python脚本如何压缩归档旧日志文件

wen python案例 2026-06-27 253

本文目录导读：

Python脚本如何压缩归档旧日志文件

基础版本
高级版本（带更多功能）
简化实用版本
使用crontab定时执行

我来提供一个完整的Python脚本方案来压缩归档旧日志文件：

基础版本

#!/usr/bin/env python3
import os
import gzip
import shutil
from datetime import datetime, timedelta
import glob
def compress_old_logs(log_dir, days_old=7, delete_original=True):
    """
    压缩指定目录下的旧日志文件
    Args:
        log_dir: 日志文件目录路径
        days_old: 多少天前的文件算旧文件
        delete_original: 压缩后是否删除原文件
    """
    # 计算截止日期
    cutoff_date = datetime.now() - timedelta(days=days_old)
    # 查找所有.log文件
    log_files = glob.glob(os.path.join(log_dir, '*.log'))
    # 也可以添加其他扩展名
    # log_files.extend(glob.glob(os.path.join(log_dir, '*.txt')))
    compressed_count = 0
    for log_file in log_files:
        # 获取文件修改时间
        file_mtime = datetime.fromtimestamp(os.path.getmtime(log_file))
        # 判断文件是否超过指定天数
        if file_mtime < cutoff_date:
            # 创建压缩文件名
            gz_filename = log_file + '.gz'
            try:
                # 压缩文件
                with open(log_file, 'rb') as f_in:
                    with gzip.open(gz_filename, 'wb') as f_out:
                        shutil.copyfileobj(f_in, f_out)
                # 删除原文件（可选）
                if delete_original:
                    os.remove(log_file)
                print(f"已压缩: {log_file} -> {gz_filename}")
                compressed_count += 1
            except Exception as e:
                print(f"压缩失败 {log_file}: {e}")
    print(f"共压缩 {compressed_count} 个文件")
# 使用示例
if __name__ == "__main__":
    compress_old_logs("/var/log/myapp", days_old=7)

高级版本（带更多功能）

#!/usr/bin/env python3
import os
import gzip
import shutil
import argparse
import logging
from datetime import datetime, timedelta
from pathlib import Path
import zipfile
class LogArchiver:
    """日志归档器"""
    def __init__(self, config=None):
        self.config = config or {}
        self.setup_logging()
    def setup_logging(self):
        """设置日志记录"""
        log_format = '%(asctime)s - %(levelname)s - %(message)s'
        logging.basicConfig(
            level=logging.INFO,
            format=log_format,
            handlers=[
                logging.FileHandler('archiver.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def compress_gzip(self, source_path, output_path=None):
        """使用gzip压缩单个文件"""
        if output_path is None:
            output_path = str(source_path) + '.gz'
        try:
            with open(source_path, 'rb') as f_in:
                with gzip.open(output_path, 'wb', compresslevel=9) as f_out:
                    shutil.copyfileobj(f_in, f_out)
            self.logger.info(f"Gzip压缩完成: {source_path} -> {output_path}")
            return True
        except Exception as e:
            self.logger.error(f"Gzip压缩失败: {e}")
            return False
    def compress_zip(self, source_dir, output_path, file_pattern='*.log', days_old=7):
        """将旧日志打包为zip文件"""
        cutoff_date = datetime.now() - timedelta(days=days_old)
        files_to_archive = []
        # 查找符合条件的文件
        for log_file in Path(source_dir).glob(file_pattern):
            file_mtime = datetime.fromtimestamp(log_file.stat().st_mtime)
            if file_mtime < cutoff_date:
                files_to_archive.append(log_file)
        if not files_to_archive:
            self.logger.info("没有找到需要归档的文件")
            return False
        try:
            # 创建zip文件
            with zipfile.ZipFile(output_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
                for file_path in files_to_archive:
                    zipf.write(file_path, file_path.name)
                    self.logger.info(f"添加到zip: {file_path}")
            # 删除原文件
            for file_path in files_to_archive:
                file_path.unlink()
                self.logger.info(f"已删除原文件: {file_path}")
            self.logger.info(f"Zip归档完成: {output_path}")
            return True
        except Exception as e:
            self.logger.error(f"Zip归档失败: {e}")
            return False
    def archive_by_pattern(self, source_dir, patterns=['*.log', '*.txt'], 
                          days_old=7, archive_format='gz', output_dir=None):
        """根据文件模式归档"""
        if output_dir is None:
            output_dir = source_dir
        cutoff_date = datetime.now() - timedelta(days=days_old)
        archived_count = 0
        for pattern in patterns:
            for log_file in Path(source_dir).glob(pattern):
                file_mtime = datetime.fromtimestamp(log_file.stat().st_mtime)
                if file_mtime < cutoff_date:
                    if archive_format == 'gz':
                        # 单个文件压缩
                        self.compress_gzip(log_file)
                    elif archive_format == 'zip':
                        # 批量zip归档（这里简化处理，实际可以根据时间分批）
                        pass
                    archived_count += 1
        self.logger.info(f"共归档 {archived_count} 个文件")
        return archived_count
    def clean_old_archives(self, archive_dir, days_to_keep=30):
        """清理旧的归档文件"""
        cutoff_date = datetime.now() - timedelta(days=days_to_keep)
        cleaned_count = 0
        for archive_file in Path(archive_dir).glob('*.gz'):
            file_mtime = datetime.fromtimestamp(archive_file.stat().st_mtime)
            if file_mtime < cutoff_date:
                archive_file.unlink()
                self.logger.info(f"已删除旧归档: {archive_file}")
                cleaned_count += 1
        self.logger.info(f"共清理 {cleaned_count} 个旧归档文件")
# 命令行接口
def main():
    parser = argparse.ArgumentParser(description='日志文件归档工具')
    parser.add_argument('source_dir', help='日志文件目录')
    parser.add_argument('--days', '-d', type=int, default=7, help='归档多少天前的文件')
    parser.add_argument('--format', '-f', choices=['gz', 'zip'], default='gz', 
                       help='归档格式')
    parser.add_argument('--output', '-o', help='输出目录')
    parser.add_argument('--pattern', '-p', nargs='+', default=['*.log'], 
                       help='文件匹配模式')
    parser.add_argument('--dry-run', action='store_true', help='模拟运行，不实际执行')
    args = parser.parse_args()
    archiver = LogArchiver()
    if args.dry_run:
        print("模拟运行模式 - 不实际执行归档")
        # 这里可以添加模拟运行的逻辑
        return
    archiver.archive_by_pattern(
        source_dir=args.source_dir,
        patterns=args.pattern,
        days_old=args.days,
        archive_format=args.format,
        output_dir=args.output
    )
if __name__ == "__main__":
    main()

简化实用版本

#!/usr/bin/env python3
import os
import gzip
import shutil
from pathlib import Path
from datetime import datetime, timedelta
def simple_log_rotation(log_dir, days=7):
    """简单的日志轮转函数"""
    cutoff = datetime.now() - timedelta(days=days)
    for log_file in Path(log_dir).glob("*.log*"):
        if log_file.suffix == '.gz':  # 跳过已压缩的文件
            continue
        mtime = datetime.fromtimestamp(log_file.stat().st_mtime)
        if mtime < cutoff:
            # 压缩文件
            with open(log_file, 'rb') as f_in:
                with gzip.open(f"{log_file}.gz", 'wb') as f_out:
                    shutil.copyfileobj(f_in, f_out)
            # 删除原文件
            log_file.unlink()
            print(f"Archived: {log_file.name}")
# 使用示例
if __name__ == "__main__":
    simple_log_rotation("/var/log/myapp")

使用crontab定时执行

创建脚本后,可以添加到crontab实现定时执行：

# 每天凌晨2点执行
0 2 * * * /usr/bin/python3 /path/to/log_archiver.py /var/log/myapp --days 7

这个脚本的主要特点：

多种压缩格式：支持gzip和zip格式
灵活的文件匹配：支持通配符模式
时间控制：可配置归档天数
日志记录：记录归档过程
错误处理：有基本的异常处理
可选功能：可以选择是否删除原文件

根据你的具体需求选择合适的版本使用,如果只是简单使用，基础版本就足够了；如果需要更多功能，可以使用高级版本。

本文地址： https://www.bmjup.com/post/10453.html