使用场景
在日常开发和系统运维中,经常需要定期清理特定目录下的临时文件、缓存文件或日志文件。典型应用场景包括:
服务器日志清理:Web服务器每天产生大量日志文件,需要定期清理避免磁盘空间不足
缓存文件管理:应用程序缓存目录会积累大量临时文件,需要定时清理释放空间
备份文件维护:自动备份系统产生的历史备份文件需要按保留策略进行清理
临时文件处理:下载目录、临时工作目录中的过期文件需要自动清除
手动执行这些清理工作既繁琐又容易遗漏,通过Python实现自动化清理可以大大提高效率并确保任务的稳定性。
实现方法
核心思路
- 使用pathlib模块进行文件和目录操作,提供更现代的路径处理方式
- 利用shutil库处理文件和目录的删除操作
- 通过schedule库设置定时任务调度
- 结合logging模块记录清理操作的详细日志
技术要点
- 支持按文件修改时间进行过期判断
- 可配置保留天数,灵活适应不同需求
- 自动处理空目录清理
- 提供详细的执行日志和错误处理
代码示例
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()
使用说明
修改配置参数:将TARGET_DIRECTORY改为需要清理的实际路径
调整保留策略:根据需求修改RETENTION_DAYS参数
设置执行时间:通过SCHEDULE_TIME调整清理任务的执行时间
运行程序:直接执行脚本即可启动定时清理服务
这类自动化运维任务能够有效管理系统存储空间,减少手动维护工作量,特别适合需要定期清理文件的各类应用场景。
注意事项:
- 首次使用前建议先注释掉定时任务,手动执行测试清理效果
- 确保程序运行账户对目标目录有读写权限
- 生产环境建议将程序设置为系统服务运行
- 定期检查日志文件 cleanup.log 确认清理任务正常执行