Python 实现定时自动化清理指定路径下的文件的方法

import os
import schedule
import time
import logging
from pathlib import Path
from datetime import datetime, timedelta
from shutil import rmtree

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('cleanup.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)

def cleanup_directory(target_path, retention_days=30):
    """
    清理指定目录下的过期文件和空目录
    
    Args:
        target_path: 目标清理路径
        retention_days: 文件保留天数，默认30天
    """
    target_dir = Path(target_path)
    
    if not target_dir.exists():
        logging.error(f"目标路径不存在: {target_path}")
        return
    
    if not target_dir.is_dir():
        logging.error(f"目标路径不是目录: {target_path}")
        return
    
    current_time = datetime.now()
    cutoff_time = current_time - timedelta(days=retention_days)
    
    removed_files = 0
    removed_dirs = 0
    total_size = 0
    
    logging.info(f"开始清理目录: {target_path}")
    
    for item in sorted(target_dir.rglob('*'), key=lambda x: len(x.parts), reverse=True):
        try:
            if item.is_file():
                mtime = datetime.fromtimestamp(item.stat().st_mtime)
                if mtime < cutoff_time:
                    file_size = item.stat().st_size
                    item.unlink()  # 删除文件
                    removed_files += 1
                    total_size += file_size
                    logging.info(f"删除过期文件: {item} (大小: {file_size}字节)")
            
            elif item.is_dir():
                if not any(item.iterdir()):
                    item.rmdir()  # 删除空目录
                    removed_dirs += 1
                    logging.info(f"删除空目录: {item}")
                    
        except Exception as e:
            logging.error(f"处理路径 {item} 时出错: {e}")
    
    size_unit = "字节"
    size_value = total_size
    
    if total_size >= 1024 * 1024:
        size_value = round(total_size / (1024 * 1024), 2)
        size_unit = "MB"
    elif total_size >= 1024:
        size_value = round(total_size / 1024, 2)
        size_unit = "KB"
    
    logging.info(
        f"清理完成: 删除 {removed_files} 个文件, {removed_dirs} 个空目录, "
        f"释放空间: {size_value} {size_unit}"
    )

def setup_scheduler(cleanup_path, retention_days=30, schedule_time="02:00"):
    """
    设置定时清理任务
    
    Args:
        cleanup_path: 清理路径
        retention_days: 保留天数
        schedule_time: 执行时间，格式"HH:MM"
    """
    # 每周日凌晨执行清理
    schedule.every().sunday.at(schedule_time).do(
        cleanup_directory, cleanup_path, retention_days
    )
    
    # 每天执行一次清理（可选，根据需求开启）
    # schedule.every().day.at(schedule_time).do(
    #     cleanup_directory, cleanup_path, retention_days
    # )
    
    logging.info(f"定时清理任务已设置: 每周日 {schedule_time} 清理 {cleanup_path}")

def main():
    """主程序入口"""
    # 配置参数
    TARGET_DIRECTORY = "/path/to/your/cleanup/directory"  # 修改为实际路径
    RETENTION_DAYS = 7  # 文件保留7天
    SCHEDULE_TIME = "02:00"  # 执行时间
    
    logging.info("启动文件清理调度程序")
    
    # 设置定时任务
    setup_scheduler(TARGET_DIRECTORY, RETENTION_DAYS, SCHEDULE_TIME)
    
    # 立即执行一次清理（可选，用于测试）
    # cleanup_directory(TARGET_DIRECTORY, RETENTION_DAYS)
    
    try:
        while True:
            schedule.run_pending()
            time.sleep(60)  # 每分钟检查一次任务
    except KeyboardInterrupt:
        logging.info("程序被用户中断")
    except Exception as e:
        logging.error(f"程序运行出错: {e}")

if __name__ == "__main__":
    main()