Python脚本如何定时清理过期日志

wen python案例 2026-06-27 237

本文目录导读：

Python脚本如何定时清理过期日志

方法1：使用 schedule 库（推荐）
方法2：使用 APScheduler 库（功能更强大）
方法3：使用 cron 或 Windows 任务计划（系统级别）
方法4：使用守护进程方式
安装依赖
最佳实践建议

我来介绍几种Python定时清理过期日志的方法：

方法1：使用 `schedule` 库（推荐）

import os
import time
import schedule
from datetime import datetime, timedelta
def clean_logs(log_dir, days_keep=30):
    """
    清理指定目录下过期的日志文件
    Args:
        log_dir: 日志目录路径
        days_keep: 保留最近多少天的日志
    """
    if not os.path.exists(log_dir):
        print(f"目录不存在: {log_dir}")
        return
    cutoff_date = datetime.now() - timedelta(days=days_keep)
    cleaned_count = 0
    for filename in os.listdir(log_dir):
        file_path = os.path.join(log_dir, filename)
        # 跳过目录
        if os.path.isdir(file_path):
            continue
        # 获取文件的最后修改时间
        file_mtime = datetime.fromtimestamp(os.path.getmtime(file_path))
        # 如果文件修改时间早于阈值，删除
        if file_mtime < cutoff_date:
            try:
                os.remove(file_path)
                cleaned_count += 1
                print(f"已删除: {filename} (最后修改: {file_mtime})")
            except Exception as e:
                print(f"删除失败 {filename}: {e}")
    print(f"清理完成，共删除 {cleaned_count} 个文件")
# 每天凌晨3点执行清理
schedule.every().day.at("03:00").do(clean_logs, log_dir="/var/log/app", days_keep=30)
# 也可以每隔X小时执行
# schedule.every(6).hours.do(clean_logs, log_dir="/var/log/app", days_keep=30)
# 定时任务循环
if __name__ == "__main__":
    print("日志清理服务已启动...")
    while True:
        schedule.run_pending()
        time.sleep(60)  # 每分钟检查一次

方法2：使用 `APScheduler` 库（功能更强大）

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
import os
from datetime import datetime, timedelta
import logging
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def clean_expired_logs(log_dir, days_keep=30, patterns=['*.log', '*.txt']):
    """
    清理过期日志文件
    Args:
        log_dir: 日志目录
        days_keep: 保留天数
        patterns: 文件匹配模式
    """
    import glob
    logger.info(f"开始清理 {log_dir} 下的过期日志...")
    cutoff_date = datetime.now() - timedelta(days=days_keep)
    cleaned_count = 0
    total_size = 0
    for pattern in patterns:
        for file_path in glob.glob(os.path.join(log_dir, pattern)):
            if os.path.isfile(file_path):
                file_mtime = datetime.fromtimestamp(os.path.getmtime(file_path))
                if file_mtime < cutoff_date:
                    try:
                        file_size = os.path.getsize(file_path)
                        os.remove(file_path)
                        cleaned_count += 1
                        total_size += file_size
                        logger.debug(f"已删除: {file_path}")
                    except Exception as e:
                        logger.error(f"删除失败 {file_path}: {e}")
    logger.info(f"清理完成: 删除 {cleaned_count} 个文件，释放 {total_size/1024/1024:.2f} MB")
# 创建调度器
scheduler = BlockingScheduler()
# 添加定时任务：每天早上3点执行
scheduler.add_job(
    clean_expired_logs,
    CronTrigger(hour=3, minute=0),  # 每天3:00
    args=["/var/log/app", 30, ['*.log']],
    id="clean_logs_job"
)
# 也可以添加多个任务
scheduler.add_job(
    clean_expired_logs,
    CronTrigger(day_of_week="mon", hour=1, minute=0),  # 每周一1:00
    args=["/var/log/backup", 7, ['*.log', '*.txt']],
    id="clean_backup_logs"
)
if __name__ == "__main__":
    print("APScheduler 日志清理服务启动...")
    try:
        scheduler.start()
    except KeyboardInterrupt:
        print("服务停止")

方法3：使用 `cron` 或 Windows 任务计划（系统级别）

创建独立的Python脚本 log_cleaner.py：

#!/usr/bin/env python3
"""
日志清理脚本 - 配合cron或Windows任务计划使用
"""
import os
import sys
import argparse
from datetime import datetime, timedelta
import logging
def setup_logging(log_file=None):
    """配置日志"""
    handlers = [logging.StreamHandler()]
    if log_file:
        handlers.append(logging.FileHandler(log_file))
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(levelname)s - %(message)s',
        handlers=handlers
    )
def clean_directory(directory, days_keep=30, dry_run=False):
    """
    清理指定目录的过期文件
    Returns:
        tuple: (删除文件数, 释放空间byte)
    """
    logger = logging.getLogger(__name__)
    if not os.path.exists(directory):
        logger.error(f"目录不存在: {directory}")
        return 0, 0
    cutoff_date = datetime.now() - timedelta(days=days_keep)
    cleaned_count = 0
    total_size = 0
    for root, dirs, files in os.walk(directory):
        for file in files:
            file_path = os.path.join(root, file)
            # 只处理.log和.txt文件
            if not (file.endswith('.log') or file.endswith('.txt')):
                continue
            try:
                file_mtime = datetime.fromtimestamp(os.path.getmtime(file_path))
                if file_mtime < cutoff_date:
                    file_size = os.path.getsize(file_path)
                    if dry_run:
                        logger.info(f"[模拟] 将要删除: {file_path} ({file_size/1024:.1f} KB)")
                    else:
                        os.remove(file_path)
                        logger.info(f"已删除: {file_path} ({file_size/1024:.1f} KB)")
                    cleaned_count += 1
                    total_size += file_size
            except OSError as e:
                logger.error(f"处理文件失败 {file_path}: {e}")
    return cleaned_count, total_size
def main():
    parser = argparse.ArgumentParser(description='清理过期日志文件')
    parser.add_argument('-d', '--directory', required=True, 
                       help='日志目录路径')
    parser.add_argument('--days', type=int, default=30,
                       help='保留最近N天的日志 (默认: 30)')
    parser.add_argument('--dry-run', action='store_true',
                       help='模拟运行，不实际删除')
    parser.add_argument('--log-file', help='清理日志输出文件')
    args = parser.parse_args()
    # 设置日志
    setup_logging(args.log_file)
    logger = logging.getLogger(__name__)
    logger.info(f"开始清理: {args.directory}")
    logger.info(f"保留天数: {args.days} 天")
    if args.dry_run:
        logger.info("运行模式: 模拟 (不会真正删除)")
    count, size = clean_directory(args.directory, args.days, args.dry_run)
    logger.info(f"清理完成: 删除 {count} 个文件，释放 {size/1024/1024:.2f} MB")
if __name__ == "__main__":
    main()

配合系统定时任务使用

Linux (crontab):

# 编辑crontab
crontab -e
# 每天凌晨3点执行
0 3 * * * /usr/bin/python3 /path/to/log_cleaner.py -d /var/log/app --days 30 --log-file /var/log/log_cleaner.log

Windows (任务计划程序):

schtasks /create /tn "CleanLogs" /tr "python D:\scripts\log_cleaner.py -d D:\logs --days 30" /sc daily /st 03:00

方法4：使用守护进程方式

#!/usr/bin/env python3
"""
日志清理守护进程
"""
import daemon
import time
import signal
import sys
from datetime import datetime, timedelta
import os
import logging
class LogCleanerDaemon:
    def __init__(self, log_dir, days_keep=30, interval_hours=24):
        self.log_dir = log_dir
        self.days_keep = days_keep
        self.interval_hours = interval_hours
        self.running = True
        # 配置日志
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('/var/log/log_cleaner_daemon.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def clean(self):
        """执行清理"""
        cutoff_date = datetime.now() - timedelta(days=self.days_keep)
        cleaned = 0
        for filename in os.listdir(self.log_dir):
            file_path = os.path.join(self.log_dir, filename)
            if os.path.isfile(file_path):
                file_mtime = datetime.fromtimestamp(os.path.getmtime(file_path))
                if file_mtime < cutoff_date:
                    try:
                        os.remove(file_path)
                        cleaned += 1
                    except Exception as e:
                        self.logger.error(f"删除失败 {file_path}: {e}")
        self.logger.info(f"清理完成，删除 {cleaned} 个文件")
    def run(self):
        """运行守护进程"""
        self.logger.info("日志清理守护进程启动")
        def signal_handler(signum, frame):
            self.logger.info("接收到停止信号")
            self.running = False
        signal.signal(signal.SIGTERM, signal_handler)
        signal.signal(signal.SIGINT, signal_handler)
        while self.running:
            try:
                self.clean()
                time.sleep(self.interval_hours * 3600)
            except Exception as e:
                self.logger.error(f"清理过程出错: {e}")
                time.sleep(60)  # 出错后等待1分钟重试
if __name__ == "__main__":
    # 以守护进程方式运行
    with daemon.DaemonContext():
        cleaner = LogCleanerDaemon("/var/log/app", days_keep=30, interval_hours=24)
        cleaner.run()

安装依赖

# 方法1需要的库
pip install schedule
# 方法2需要的库
pip install apscheduler
# 方法4需要的库
pip install python-daemon

最佳实践建议

先测试再部署：使用 --dry-run 参数模拟运行
保留备份：删除前可先压缩归档
监控清理效果：记录清理日志
处理异常情况：添加错误处理和重试机制
考虑日志轮转：配合 logrotate 使用效果更好

选择哪种方法取决于你的具体需求：

简单场景：使用 schedule 库
复杂定时需求：使用 APScheduler
生产环境：使用系统级 cron 或 systemd timer
需要持久运行：使用守护进程方式

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

Python脚本如何定时清理过期日志

方法1：使用 schedule 库（推荐）

方法2：使用 APScheduler 库（功能更强大）

方法3：使用 cron 或 Windows 任务计划（系统级别）

配合系统定时任务使用

方法4：使用守护进程方式

安装依赖

最佳实践建议

方法1：使用 `schedule` 库（推荐）

方法2：使用 `APScheduler` 库（功能更强大）

方法3：使用 `cron` 或 Windows 任务计划（系统级别）