如何写一个脚本批量压缩日志

wen 实用脚本 2026-07-29 187

本文目录导读：

如何写一个脚本批量压缩日志

自定义选项
使用配置文件
试运行（不实际压缩）
每周日凌晨3点执行

我来提供一个批量压缩日志文件的Python脚本,支持多种压缩方式和灵活配置。

基础版本：压缩指定目录下的日志文件

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
日志批量压缩脚本
"""
import os
import gzip
import bz2
import shutil
from datetime import datetime, timedelta
from pathlib import Path
import logging
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def compress_logs_gzip(log_dir, older_than_days=0, delete_original=True):
    """
    使用gzip压缩日志文件
    Args:
        log_dir: 日志目录
        older_than_days: 压缩多少天前的文件（0表示所有）
        delete_original: 是否删除原始文件
    """
    log_path = Path(log_dir)
    if not log_path.exists():
        logger.error(f"目录不存在: {log_dir}")
        return
    # 计算截止时间
    if older_than_days > 0:
        cutoff_date = datetime.now() - timedelta(days=older_than_days)
    for file_path in log_path.glob("*"):
        # 跳过目录和已压缩的文件
        if file_path.is_dir() or file_path.suffix == '.gz':
            continue
        # 检查文件时间
        if older_than_days > 0:
            file_mtime = datetime.fromtimestamp(file_path.stat().st_mtime)
            if file_mtime > cutoff_date:
                continue
        try:
            compressed_path = file_path.with_suffix(file_path.suffix + '.gz')
            # 压缩文件
            with open(file_path, 'rb') as f_in:
                with gzip.open(compressed_path, 'wb') as f_out:
                    shutil.copyfileobj(f_in, f_out)
            # 删除原始文件
            if delete_original:
                file_path.unlink()
            logger.info(f"已压缩: {file_path.name} -> {compressed_path.name}")
        except Exception as e:
            logger.error(f"压缩失败 {file_path}: {str(e)}")
def compress_logs_bz2(log_dir, older_than_days=0):
    """使用bzip2压缩（压缩率更高但速度较慢）"""
    log_path = Path(log_dir)
    if not log_path.exists():
        logger.error(f"目录不存在: {log_dir}")
        return
    if older_than_days > 0:
        cutoff_date = datetime.now() - timedelta(days=older_than_days)
    for file_path in log_path.glob("*"):
        if file_path.is_dir() or file_path.suffix in ['.bz2', '.gz']:
            continue
        if older_than_days > 0:
            file_mtime = datetime.fromtimestamp(file_path.stat().st_mtime)
            if file_mtime > cutoff_date:
                continue
        try:
            compressed_path = file_path.with_suffix(file_path.suffix + '.bz2')
            with open(file_path, 'rb') as f_in:
                with bz2.open(compressed_path, 'wb') as f_out:
                    shutil.copyfileobj(f_in, f_out)
            file_path.unlink()
            logger.info(f"已压缩: {file_path.name} -> {compressed_path.name}")
        except Exception as e:
            logger.error(f"压缩失败 {file_path}: {str(e)}")
if __name__ == "__main__":
    # 使用示例
    LOG_DIR = "/var/log/myapp"  # 修改为实际日志目录
    # 压缩7天前的日志
    compress_logs_gzip(LOG_DIR, older_than_days=7)
    # 或者使用bzip2
    # compress_logs_bz2(LOG_DIR, older_than_days=7)

高级版本：功能更完整的脚本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
高级日志批量压缩工具
支持：多种压缩格式、文件筛选、进度显示、配置管理
"""
import os
import gzip
import bz2
import zipfile
import shutil
import argparse
import json
from datetime import datetime, timedelta
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed
import logging
from typing import List, Tuple
class LogCompressor:
    """日志压缩器"""
    def __init__(self, config_file=None):
        self.logger = self._setup_logger()
        self.config = self._load_config(config_file)
        self.stats = {'compressed': 0, 'failed': 0, 'skipped': 0}
    def _setup_logger(self):
        """配置日志系统"""
        logger = logging.getLogger(__name__)
        logger.setLevel(logging.INFO)
        # 控制台输出
        ch = logging.StreamHandler()
        ch.setLevel(logging.INFO)
        formatter = logging.Formatter(
            '%(asctime)s - %(levelname)s - %(message)s'
        )
        ch.setFormatter(formatter)
        logger.addHandler(ch)
        # 文件输出
        fh = logging.FileHandler('log_compressor.log')
        fh.setLevel(logging.DEBUG)
        fh.setFormatter(formatter)
        logger.addHandler(fh)
        return logger
    def _load_config(self, config_file):
        """加载配置文件"""
        default_config = {
            'compress_method': 'gzip',
            'delete_original': True,
            'older_than_days': 7,
            'max_size_mb': 100,
            'max_workers': 4,
            'exclude_patterns': ['*.gz', '*.bz2', '*.zip'],
            'include_patterns': ['*.log', '*.txt', '*.out']
        }
        if config_file and os.path.exists(config_file):
            with open(config_file, 'r') as f:
                config = json.load(f)
                default_config.update(config)
                return default_config
        return default_config
    def should_compress(self, file_path: Path) -> bool:
        """判断文件是否应该被压缩"""
        # 检查文件模式
        if self.config.get('include_patterns'):
            import fnmatch
            if not any(fnmatch.fnmatch(file_path.name, pattern) 
                      for pattern in self.config['include_patterns']):
                return False
        if self.config.get('exclude_patterns'):
            import fnmatch
            if any(fnmatch.fnmatch(file_path.name, pattern) 
                  for pattern in self.config['exclude_patterns']):
                return False
        # 检查文件大小
        if self.config.get('max_size_mb'):
            max_bytes = self.config['max_size_mb'] * 1024 * 1024
            if file_path.stat().st_size > max_bytes:
                return False
        # 检查文件时间
        if self.config.get('older_than_days', 0) > 0:
            cutoff = datetime.now() - timedelta(days=self.config['older_than_days'])
            file_mtime = datetime.fromtimestamp(file_path.stat().st_mtime)
            if file_mtime > cutoff:
                return False
        return True
    def compress_file(self, file_path: Path) -> Tuple[bool, str]:
        """压缩单个文件"""
        method = self.config.get('compress_method', 'gzip')
        delete_original = self.config.get('delete_original', True)
        try:
            if method == 'gzip':
                compressed_path = file_path.with_suffix(file_path.suffix + '.gz')
                with open(file_path, 'rb') as f_in:
                    with gzip.open(compressed_path, 'wb', compresslevel=9) as f_out:
                        shutil.copyfileobj(f_in, f_out)
            elif method == 'bz2':
                compressed_path = file_path.with_suffix(file_path.suffix + '.bz2')
                with open(file_path, 'rb') as f_in:
                    with bz2.open(compressed_path, 'wb', compresslevel=9) as f_out:
                        shutil.copyfileobj(f_in, f_out)
            elif method == 'zip':
                compressed_path = file_path.with_suffix('.zip')
                with zipfile.ZipFile(compressed_path, 'w', zipfile.ZIP_DEFLATED) as zf:
                    zf.write(file_path, file_path.name)
            else:
                return False, f"不支持的压缩方法: {method}"
            # 计算压缩率
            original_size = file_path.stat().st_size
            compressed_size = compressed_path.stat().st_size
            ratio = (1 - compressed_size / original_size) * 100
            # 删除原始文件
            if delete_original:
                file_path.unlink()
            return True, f"压缩完成: {ratio:.1f}% 节省空间"
        except Exception as e:
            return False, str(e)
    def compress_directory(self, log_dir: str):
        """压缩目录中的所有日志文件"""
        log_path = Path(log_dir)
        if not log_path.exists():
            self.logger.error(f"目录不存在: {log_dir}")
            return
        # 收集需要压缩的文件
        files_to_compress = []
        for file_path in log_path.iterdir():
            if file_path.is_file() and self.should_compress(file_path):
                files_to_compress.append(file_path)
        if not files_to_compress:
            self.logger.info("没有需要压缩的文件")
            return
        self.logger.info(f"找到 {len(files_to_compress)} 个需要压缩的文件")
        # 并行压缩
        max_workers = self.config.get('max_workers', 4)
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {
                executor.submit(self.compress_file, f): f 
                for f in files_to_compress
            }
            for future in as_completed(futures):
                file_path = futures[future]
                try:
                    success, message = future.result()
                    if success:
                        self.stats['compressed'] += 1
                        self.logger.info(f"✓ {file_path.name}: {message}")
                    else:
                        self.stats['failed'] += 1
                        self.logger.error(f"✗ {file_path.name}: {message}")
                except Exception as e:
                    self.stats['failed'] += 1
                    self.logger.error(f"✗ {file_path.name}: 异常 - {str(e)}")
        # 输出统计信息
        self.logger.info(f"\n压缩统计:")
        self.logger.info(f"  成功: {self.stats['compressed']}")
        self.logger.info(f"  失败: {self.stats['failed']}")
        self.logger.info(f"  总共: {len(files_to_compress)}")
def main():
    """主函数"""
    parser = argparse.ArgumentParser(
        description='批量压缩日志文件',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
使用示例:
  %(prog)s /var/log/myapp
  %(prog)s /var/log/myapp --method bz2 --days 30
  %(prog)s /var/log/myapp --config config.json --dry-run
        """
    )
    parser.add_argument('log_dir', help='日志目录路径')
    parser.add_argument('--method', choices=['gzip', 'bz2', 'zip'], 
                       default='gzip', help='压缩方法 (默认: gzip)')
    parser.add_argument('--days', type=int, default=7,
                       help='压缩多少天前的文件 (默认: 7)')
    parser.add_argument('--keep-original', action='store_true',
                       help='保留原始文件')
    parser.add_argument('--max-size', type=float, default=100,
                       help='最大文件大小MB (默认: 100)')
    parser.add_argument('--config', help='配置文件路径')
    parser.add_argument('--dry-run', action='store_true',
                       help='试运行，不实际压缩')
    parser.add_argument('--workers', type=int, default=4,
                       help='并行工作线程数 (默认: 4)')
    args = parser.parse_args()
    # 创建压缩器
    if args.config:
        compressor = LogCompressor(args.config)
    else:
        compressor = LogCompressor()
    # 更新配置
    compressor.config.update({
        'compress_method': args.method,
        'delete_original': not args.keep_original,
        'older_than_days': args.days,
        'max_size_mb': args.max_size,
        'max_workers': args.workers
    })
    # 试运行模式
    if args.dry_run:
        log_path = Path(args.log_dir)
        if log_path.exists():
            files_to_compress = [
                f for f in log_path.iterdir() 
                if f.is_file() and compressor.should_compress(f)
            ]
            print(f"\n试运行结果:")
            print(f"将压缩 {len(files_to_compress)} 个文件")
            print(f"压缩方法: {args.method}")
            print(f"删除原始文件: {not args.keep_original}")
            print(f"文件条件: {args.days}天前, 最大{args.max_size}MB")
            if files_to_compress:
                print("\n将要压缩的文件:")
                for f in files_to_compress[:10]:
                    mtime = datetime.fromtimestamp(f.stat().st_mtime)
                    size_mb = f.stat().st_size / (1024 * 1024)
                    print(f"  {f.name} ({mtime.strftime('%Y-%m-%d')}, {size_mb:.1f}MB)")
                if len(files_to_compress) > 10:
                    print(f"  ... 还有 {len(files_to_compress) - 10} 个文件")
        else:
            print(f"目录不存在: {args.log_dir}")
        return
    # 执行压缩
    compressor.compress_directory(args.log_dir)
if __name__ == "__main__":
    main()

Shell版本脚本

#!/bin/bash
# log_compress.sh - Shell日志压缩脚本
set -euo pipefail
# 配置
LOG_DIR="/var/log/myapp"
DAYS_OLD=7
COMPRESS_CMD="gzip"
DELETE_ORIGINAL=true
# 显示帮助
show_help() {
    cat << EOF
用法: $0 [选项]
选项:
  -d, --dir DIR        日志目录路径
  -a, --days DAYS      压缩N天前的文件 (默认: 7)
  -c, --cmd CMD        压缩命令 (gzip/bzip2/xz, 默认: gzip)
  -k, --keep           保留原始文件
  -h, --help           显示帮助信息
EOF
}
# 解析参数
while [[ $# -gt 0 ]]; do
    case $1 in
        -d|--dir)
            LOG_DIR="$2"
            shift 2
            ;;
        -a|--days)
            DAYS_OLD="$2"
            shift 2
            ;;
        -c|--cmd)
            COMPRESS_CMD="$2"
            shift 2
            ;;
        -k|--keep)
            DELETE_ORIGINAL=false
            shift
            ;;
        -h|--help)
            show_help
            exit 0
            ;;
        *)
            echo "未知选项: $1"
            show_help
            exit 1
            ;;
    esac
done
# 检查目录
if [[ ! -d "$LOG_DIR" ]]; then
    echo "错误: 目录不存在: $LOG_DIR"
    exit 1
fi
# 设置压缩选项
case $COMPRESS_CMD in
    gzip)
        EXT=".gz"
        OPTS="-9"
        ;;
    bzip2)
        EXT=".bz2"
        OPTS="-9"
        ;;
    xz)
        EXT=".xz"
        OPTS="-9"
        ;;
    *)
        echo "不支持的压缩命令: $COMPRESS_CMD"
        exit 1
        ;;
esac
echo "开始压缩日志文件..."
echo "目录: $LOG_DIR"
echo "条件: ${DAYS_OLD}天前"
echo "压缩: ${COMPRESS_CMD}"
# 统计变量
total=0
success=0
failed=0
# 查找并压缩文件
while IFS= read -r -d '' file; do
    total=$((total + 1))
    # 压缩
    if $COMPRESS_CMD $OPTS "$file"; then
        echo "✓ 已压缩: $file"
        if $DELETE_ORIGINAL; then
            # gzip默认会删除原始文件，其他需要手动删除
            if [[ "$COMPRESS_CMD" != "gzip" ]]; then
                rm "$file"
            fi
        else
            # 如果保留原始文件，gzip需要-k选项
            if [[ "$COMPRESS_CMD" == "gzip" ]]; then
                mv "${file}${EXT}" "${file}"
                $COMPRESS_CMD -k $OPTS "$file"
            fi
        fi
        success=$((success + 1))
    else
        echo "✗ 失败: $file"
        failed=$((failed + 1))
    fi
done < <(find "$LOG_DIR" -type f -name "*.log" -mtime +$DAYS_OLD -print0)
echo ""
echo "压缩统计:"
echo "  总计: $total"
echo "  成功: $success"
echo "  失败: $failed"

配置文件示例 (config.json)

{
    "compress_method": "gzip",
    "delete_original": true,
    "older_than_days": 7,
    "max_size_mb": 100,
    "max_workers": 4,
    "exclude_patterns": ["*.gz", "*.bz2", "*.zip"],
    "include_patterns": ["*.log", "*.txt", "*.out"]
}

使用说明

Python脚本使用：

# 基本用法
python log_compressor.py /var/log/myapp

自定义选项

python log_compressor.py /var/log/myapp --method bz2 --days 30 --keep-original

使用配置文件

python log_compressor.py /var/log/myapp --config config.json

试运行（不实际压缩）

python log_compressor.py /var/log/myapp --dry-run


2. **Shell脚本使用**：
```bash
# 基本用法
chmod +x log_compress.sh
./log_compress.sh -d /var/log/myapp
# 自定义
./log_compress.sh -d /var/log/myapp -a 30 -c bzip2 -k

Crontab定时任务配置：

# 每天凌晨2点执行
0 2 * * * /usr/local/bin/log_compressor.py /var/log/myapp --days 7

每周日凌晨3点执行

0 3 0 /usr/local/bin/log_compress.sh -d /var/log/myapp -a 7


这个脚本提供了完整的日志压缩解决方案，支持多种压缩格式、灵活的配置、并行处理等功能。

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