一文带你深入了解Python中的自动化任务调度

pip install schedule

import schedule
import time # 用于暂停程序，让调度器有机会运行
import os
import datetime

# 假设这个函数就是你之前写的清理函数，这里为了演示简化为打印日志
def perform_cleanup_job(folder_path, days_threshold):
    """
    模拟一个文件清理任务。
    在实际应用中，这里会调用你之前编写的 auto_clean_folder_once 函数。
    """
    print(f"⏰ [{datetime.datetime.now().strftime('%h:%m:%s')}] 正在执行定时清理任务...")
    print(f"   - 清理目录: {folder_path}")
    print(f"   - 删除 {days_threshold} 天前的过期文件。")
    # 在实际代码中，这里会调用 auto_clean_folder_once(folder_path, days_threshold)
    print("   ✨ 本次清理任务完成！")

if __name__ == "__main__":
    # 请替换为你要定时清理的文件夹路径
    target_folder_for_cleanup = os.path.expanduser("~/desktop/每日自动清理") 
    os.makedirs(target_folder_for_cleanup, exist_ok=true) # 确保文件夹存在

    print("🚀 正在启动python自动化任务调度器 (schedule库演示)...")
    print("--- 请注意：此脚本需要持续运行才能保持调度！ ---")

    # 示例1：每隔5秒执行一次任务
    schedule.every(5).seconds.do(perform_cleanup_job, target_folder_for_cleanup, 7) # 模拟清理7天前文件
    print(f"任务已设置：每5秒执行一次清理任务，目标文件夹：{target_folder_for_cleanup}")

    # 示例2：每隔1分钟执行一次
    # schedule.every(1).minutes.do(perform_cleanup_job, target_folder_for_cleanup, 14)
    # print("任务已设置：每1分钟执行一次清理任务。")

    # 示例3：每天的特定时间执行任务 (例如每天凌晨1点)
    # schedule.every().day.at("01:00").do(perform_cleanup_job, target_folder_for_cleanup, 30)
    # print("任务已设置：每天凌晨01:00执行清理任务。")

    # 示例4：每周三的下午14:30执行
    # schedule.every().wednesday.at("14:30").do(perform_cleanup_job, target_folder_for_cleanup, 60)
    # print("任务已设置：每周三14:30执行清理任务。")

    # 主循环：持续检查是否有待执行的任务
    while true:
        schedule.run_pending() # 运行所有待执行的任务
        time.sleep(1) # 等待1秒，避免cpu占用过高

pip install apscheduler

from apscheduler.schedulers.background import backgroundscheduler # 后台调度器
from apscheduler.triggers.interval import intervaltrigger # 间隔触发器
from apscheduler.triggers.date import datetrigger # 日期触发器
from apscheduler.triggers.cron import crontrigger # cron表达式触发器
import os
import datetime
import time

def perform_data_sync_job():
    """
    模拟一个数据同步任务。
    实际中，这里可以调用你的其他自动化脚本。
    """
    print(f"🔄 [{datetime.datetime.now().strftime('%h:%m:%s')}] 正在执行数据同步任务...")
    # 模拟任务耗时
    time.sleep(2)
    print("   ✨ 数据同步任务完成！")

def perform_weekly_report_job():
    """模拟一个每周生成报告的任务"""
    print(f"📊 [{datetime.datetime.now().strftime('%h:%m:%s')}] 正在生成每周报告...")
    time.sleep(3)
    print("   ✨ 每周报告生成完成！")

if __name__ == "__main__":
    scheduler = backgroundscheduler() # 创建一个后台调度器实例
    
    print("🚀 正在启动apscheduler任务调度器...")

    # 1. 间隔触发器 (intervaltrigger)：每隔5秒执行一次
    scheduler.add_job(
        perform_data_sync_job, 
        intervaltrigger(seconds=5), 
        id='data_sync_job' # 为作业指定唯一id
    )
    print("任务：'data_sync_job' 已设置，每5秒执行一次。")

    # 2. 日期触发器 (datetrigger)：在指定日期和时间执行一次性任务
    # run_date = datetime.datetime.now() + datetime.timedelta(seconds=10)
    # scheduler.add_job(
    #     perform_weekly_report_job, 
    #     datetrigger(run_date=run_date), 
    #     id='one_time_report', 
    #     name='一次性报告任务'
    # )
    # print(f"任务：'one_time_report' 已设置，将在 {run_date.strftime('%h:%m:%s')} 执行。")

    # 3. cron触发器 (crontrigger)：使用cron表达式进行复杂调度
    # 每天凌晨2点0分执行
    scheduler.add_job(
        perform_weekly_report_job, 
        crontrigger(hour=2, minute=0), 
        id='daily_cleanup_cron', 
        name='每日凌晨清理任务'
    )
    print("任务：'daily_cleanup_cron' 已设置，每天凌晨2点执行。")

    # 启动调度器
    scheduler.start()
    print("调度器已启动。程序将持续运行以保持调度。按 ctrl+c 停止。")

    # 保持主线程运行，以便后台调度器能够工作
    try:
        while true:
            time.sleep(1)
    except (keyboardinterrupt, systemexit):
        scheduler.shutdown() # 安全关闭调度器
        print("调度器已停止。")

from apscheduler.schedulers.background import backgroundscheduler
from apscheduler.jobstores.sqlalchemy import sqlalchemyjobstore # 用于持久化
from apscheduler.triggers.interval import intervaltrigger
import os
import datetime
import time

# 定义一个会失败的模拟任务
def failing_job():
    print(f"🔴 [{datetime.datetime.now().strftime('%h:%m:%s')}] 尝试执行一个可能失败的任务...")
    if datetime.datetime.now().second % 10 == 0: # 每10秒模拟失败一次
        raise valueerror("模拟任务执行失败！")
    print("   ✅ 任务成功完成！")

# 定义一个错误监听器
def my_listener(event):
    if event.exception:
        print(f"🔥 作业 '{event.job_id}' 执行失败：{event.exception}")
    else:
        print(f"✨ 作业 '{event.job_id}' 正常完成。")

if __name__ == "__main__":
    # 配置作业存储：这里使用sqlite作为示例数据库
    jobstores = {
        'default': sqlalchemyjobstore(url='sqlite:///jobs.sqlite')
    }
    # 配置执行器 (默认是threadpoolexecutor)
    executors = {
        'default': {'type': 'threadpool', 'max_workers': 20}
    }

    scheduler = backgroundscheduler(jobstores=jobstores, executors=executors)
    
    print("🚀 正在启动apscheduler调度器 (持久化与异常处理演示)...")

    # 添加一个每5秒执行的任务
    scheduler.add_job(
        failing_job, 
        intervaltrigger(seconds=5), 
        id='resilient_task',
        name='弹性任务'
    )
    print("任务：'resilient_task' 已设置，每5秒执行一次。")

    # 注册错误监听器
    from apscheduler.events import event_job_error, event_job_executed
    scheduler.add_listener(my_listener, event_job_error | event_job_executed)

    # 启动调度器
    scheduler.start()
    print("调度器已启动。可以尝试 ctrl+c 中断程序，再重新运行，任务会从上次中断的地方继续。")

    try:
        while true:
            time.sleep(1)
    except (keyboardinterrupt, systemexit):
        scheduler.shutdown()
        print("调度器已停止。")