如何用Python Pandas实现工作日自动保存文件及定期清理？

以下是针对你的需求的完整实现方案，代码可直接复用，分模块拆解便于理解和调整：

1. 依赖安装 #

先安装所需第三方库：

pip install pandas openpyxl schedule

datetime、os是Python标准库，无需额外安装。

2. 核心功能实现 #

2.1 工作日自动备份文件 #

这个函数负责读取Excel/CSV源文件，仅在周一至周五生成带日期标记的备份文件：

import pandas as pd
import os
from datetime import datetime, timedelta

# 替换成你的实际路径
SOURCE_FILE_PATH = "你的源文件路径.xlsx"  # 支持.xlsx或.csv格式
BACKUP_DIR = "备份文件存放目录"

def save_daily_backup():
    # 判断是否为工作日（0=周一，4=周五）
    today = datetime.today()
    if today.weekday() >= 5:
        return
    
    # 自动创建备份目录
    os.makedirs(BACKUP_DIR, exist_ok=True)
    
    # 读取源文件
    try:
        if SOURCE_FILE_PATH.endswith(".xlsx"):
            df = pd.read_excel(SOURCE_FILE_PATH, engine="openpyxl")
        elif SOURCE_FILE_PATH.endswith(".csv"):
            df = pd.read_csv(SOURCE_FILE_PATH)
        else:
            print("不支持的文件格式")
            return
    except Exception as e:
        print(f"读取文件失败: {str(e)}")
        return
    
    # 生成备份文件名（例：2024-05-20_backup.xlsx）
    backup_filename = f"{today.strftime('%Y-%m-%d')}_backup{os.path.splitext(SOURCE_FILE_PATH)[1]}"
    backup_path = os.path.join(BACKUP_DIR, backup_filename)
    
    # 保存备份
    try:
        if SOURCE_FILE_PATH.endswith(".xlsx"):
            df.to_excel(backup_path, index=False, engine="openpyxl")
        elif SOURCE_FILE_PATH.endswith(".csv"):
            df.to_csv(backup_path, index=False)
        print(f"今日备份已保存: {backup_path}")
    except Exception as e:
        print(f"保存备份失败: {str(e)}")

2.2 文件清理逻辑 #

每周清理（周一删除上周周一至周四的备份）

def weekly_cleanup():
    # 仅在周一执行
    today = datetime.today()
    if today.weekday() != 0:
        return
    
    # 计算上周周一至周四的日期范围
    last_week_monday = today - timedelta(days=7 + today.weekday())
    last_week_thursday = last_week_monday + timedelta(days=3)
    
    # 遍历并删除符合条件的文件
    for filename in os.listdir(BACKUP_DIR):
        if "_backup" not in filename:
            continue
        date_str = filename.split("_")[0]
        try:
            file_date = datetime.strptime(date_str, "%Y-%m-%d").date()
            if last_week_monday.date() <= file_date <= last_week_thursday.date():
                os.remove(os.path.join(BACKUP_DIR, filename))
                print(f"已删除上周工作日备份: {filename}")
        except ValueError:
            continue  # 跳过非规则命名的文件

月末清理（保留最新4份备份）

def monthly_end_cleanup():
    # 判断是否为月末（明天是下月1号）
    today = datetime.today()
    tomorrow = today + timedelta(days=1)
    if tomorrow.day != 1:
        return
    
    # 按日期排序备份文件
    backup_files = []
    for filename in os.listdir(BACKUP_DIR):
        if "_backup" not in filename:
            continue
        date_str = filename.split("_")[0]
        try:
            backup_files.append((datetime.strptime(date_str, "%Y-%m-%d"), filename))
        except ValueError:
            continue
    
    backup_files.sort(key=lambda x: x[0])
    # 超过4份则删除旧文件
    if len(backup_files) > 4:
        for _, filename in backup_files[:-4]:
            os.remove(os.path.join(BACKUP_DIR, filename))
            print(f"月末清理已删除: {filename}")

月初清理（仅保留最新1份备份）

def monthly_start_cleanup():
    # 仅在每月1号执行
    today = datetime.today()
    if today.day != 1:
        return
    
    backup_files = []
    for filename in os.listdir(BACKUP_DIR):
        if "_backup" not in filename:
            continue
        date_str = filename.split("_")[0]
        try:
            backup_files.append((datetime.strptime(date_str, "%Y-%m-%d"), filename))
        except ValueError:
            continue
    
    if not backup_files:
        return
    
    # 按日期倒序，保留最新的一份
    backup_files.sort(key=lambda x: x[0], reverse=True)
    latest_file = backup_files[0][1]
    for _, filename in backup_files[1:]:
        os.remove(os.path.join(BACKUP_DIR, filename))
        print(f"月初清理已删除: {filename}")

年末清理（保留最近12份每月最新备份）

def yearly_end_cleanup():
    # 判断是否为年末（明天是下年1月1号）
    today = datetime.today()
    tomorrow = today + timedelta(days=1)
    if not (tomorrow.month == 1 and tomorrow.day == 1):
        return
    
    # 按月份分组，每个月保留最新备份
    monthly_backups = {}
    for filename in os.listdir(BACKUP_DIR):
        if "_backup" not in filename:
            continue
        date_str = filename.split("_")[0]
        try:
            file_date = datetime.strptime(date_str, "%Y-%m-%d")
            month_key = (file_date.year, file_date.month)
            if month_key not in monthly_backups or file_date > monthly_backups[month_key][0]:
                monthly_backups[month_key] = (file_date, filename)
        except ValueError:
            continue
    
    # 仅保留最近12个月的备份
    if len(monthly_backups) > 12:
        sorted_months = sorted(monthly_backups.keys(), reverse=True)[:12]
        keep_files = [monthly_backups[key][1] for key in sorted_months]
    else:
        keep_files = [v[1] for v in monthly_backups.values()]
    
    # 删除不在保留列表的文件
    for filename in os.listdir(BACKUP_DIR):
        if "_backup" in filename and filename not in keep_files:
            os.remove(os.path.join(BACKUP_DIR, filename))
            print(f"年末清理已删除: {filename}")

2.3 定时任务调度 #

用schedule库设置定时执行规则，脚本需保持运行状态：

import schedule
import time

def setup_scheduler():
    # 每天9点执行备份
    schedule.every().day.at("09:00").do(save_daily_backup)
    # 每周一9点执行每周清理
    schedule.every().monday.at("09:00").do(weekly_cleanup)
    # 每天检查是否需要执行月/年末清理
    schedule.every().day.at("09:00").do(monthly_start_cleanup)
    schedule.every().day.at("09:00").do(monthly_end_cleanup)
    schedule.every().day.at("09:00").do(yearly_end_cleanup)

    # 启动任务循环
    while True:
        schedule.run_pending()
        time.sleep(60)  # 每分钟检查一次任务

if __name__ == "__main__":
    setup_scheduler()

3. 注意事项 #

• 替换代码中的SOURCE_FILE_PATH和BACKUP_DIR为你的实际路径
• 测试阶段可手动调用函数（如save_daily_backup()）验证功能，避免误删数据
• 若需脚本长期运行，可在Linux用nohup后台启动，或在Windows设置任务计划程序

内容的提问来源于stack exchange，提问作者Fritz Wagner