本文目录导读：
我来提供一个完整的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格式
- 灵活的文件匹配：支持通配符模式
- 时间控制：可配置归档天数
- 日志记录：记录归档过程
- 错误处理：有基本的异常处理
- 可选功能：可以选择是否删除原文件
根据你的具体需求选择合适的版本使用,如果只是简单使用，基础版本就足够了；如果需要更多功能，可以使用高级版本。
