pip install schedule

import schedule
import time # 用于暫停程序，讓調度器有機會運行
import os
import datetime

# 假設這個函數(shù)就是你之前寫的清理函數(shù)，這里為了演示簡化為打印日志
def perform_cleanup_job(folder_path, days_threshold):
    """
    模擬一個文件清理任務。
    在實際應用中，這里會調用你之前編寫的 auto_clean_folder_once 函數(shù)。
    """
    print(f"? [{datetime.datetime.now().strftime('%H:%M:%S')}] 正在執(zhí)行定時清理任務...")
    print(f"   - 清理目錄: {folder_path}")
    print(f"   - 刪除 {days_threshold} 天前的過期文件。")
    # 在實際代碼中，這里會調用 auto_clean_folder_once(folder_path, days_threshold)
    print("   ? 本次清理任務完成！")

if __name__ == "__main__":
    # 請?zhí)鎿Q為你要定時清理的文件夾路徑
    target_folder_for_cleanup = os.path.expanduser("~/Desktop/每日自動清理") 
    os.makedirs(target_folder_for_cleanup, exist_ok=True) # 確保文件夾存在

    print("?? 正在啟動Python自動化任務調度器 (schedule庫演示)...")
    print("--- 請注意：此腳本需要持續(xù)運行才能保持調度！ ---")

    # 示例1：每隔5秒執(zhí)行一次任務
    schedule.every(5).seconds.do(perform_cleanup_job, target_folder_for_cleanup, 7) # 模擬清理7天前文件
    print(f"任務已設置：每5秒執(zhí)行一次清理任務，目標文件夾：{target_folder_for_cleanup}")

    # 示例2：每隔1分鐘執(zhí)行一次
    # schedule.every(1).minutes.do(perform_cleanup_job, target_folder_for_cleanup, 14)
    # print("任務已設置：每1分鐘執(zhí)行一次清理任務。")

    # 示例3：每天的特定時間執(zhí)行任務 (例如每天凌晨1點)
    # schedule.every().day.at("01:00").do(perform_cleanup_job, target_folder_for_cleanup, 30)
    # print("任務已設置：每天凌晨01:00執(zhí)行清理任務。")

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

    # 主循環(huán)：持續(xù)檢查是否有待執(zhí)行的任務
    while True:
        schedule.run_pending() # 運行所有待執(zhí)行的任務
        time.sleep(1) # 等待1秒，避免CPU占用過高

pip install APScheduler

from apscheduler.schedulers.background import BackgroundScheduler # 后臺調度器
from apscheduler.triggers.interval import IntervalTrigger # 間隔觸發(fā)器
from apscheduler.triggers.date import DateTrigger # 日期觸發(fā)器
from apscheduler.triggers.cron import CronTrigger # Cron表達式觸發(fā)器
import os
import datetime
import time

def perform_data_sync_job():
    """
    模擬一個數(shù)據(jù)同步任務。
    實際中，這里可以調用你的其他自動化腳本。
    """
    print(f"?? [{datetime.datetime.now().strftime('%H:%M:%S')}] 正在執(zhí)行數(shù)據(jù)同步任務...")
    # 模擬任務耗時
    time.sleep(2)
    print("   ? 數(shù)據(jù)同步任務完成！")

def perform_weekly_report_job():
    """模擬一個每周生成報告的任務"""
    print(f"?? [{datetime.datetime.now().strftime('%H:%M:%S')}] 正在生成每周報告...")
    time.sleep(3)
    print("   ? 每周報告生成完成！")

if __name__ == "__main__":
    scheduler = BackgroundScheduler() # 創(chuàng)建一個后臺調度器實例
    
    print("?? 正在啟動APScheduler任務調度器...")

    # 1. 間隔觸發(fā)器 (IntervalTrigger)：每隔5秒執(zhí)行一次
    scheduler.add_job(
        perform_data_sync_job, 
        IntervalTrigger(seconds=5), 
        id='data_sync_job' # 為作業(yè)指定唯一ID
    )
    print("任務：'data_sync_job' 已設置，每5秒執(zhí)行一次。")

    # 2. 日期觸發(fā)器 (DateTrigger)：在指定日期和時間執(zhí)行一次性任務
    # 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')} 執(zhí)行。")

    # 3. Cron觸發(fā)器 (CronTrigger)：使用Cron表達式進行復雜調度
    # 每天凌晨2點0分執(zhí)行
    scheduler.add_job(
        perform_weekly_report_job, 
        CronTrigger(hour=2, minute=0), 
        id='daily_cleanup_cron', 
        name='每日凌晨清理任務'
    )
    print("任務：'daily_cleanup_cron' 已設置，每天凌晨2點執(zhí)行。")

    # 啟動調度器
    scheduler.start()
    print("調度器已啟動。程序將持續(xù)運行以保持調度。按 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')}] 嘗試執(zhí)行一個可能失敗的任務...")
    if datetime.datetime.now().second % 10 == 0: # 每10秒模擬失敗一次
        raise ValueError("模擬任務執(zhí)行失?。?)
    print("   ? 任務成功完成！")

# 定義一個錯誤監(jiān)聽器
def my_listener(event):
    if event.exception:
        print(f"?? 作業(yè) '{event.job_id}' 執(zhí)行失敗：{event.exception}")
    else:
        print(f"? 作業(yè) '{event.job_id}' 正常完成。")

if __name__ == "__main__":
    # 配置作業(yè)存儲：這里使用SQLite作為示例數(shù)據(jù)庫
    jobstores = {
        'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
    }
    # 配置執(zhí)行器 (默認是ThreadPoolExecutor)
    executors = {
        'default': {'type': 'threadpool', 'max_workers': 20}
    }

    scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors)
    
    print("?? 正在啟動APScheduler調度器 (持久化與異常處理演示)...")

    # 添加一個每5秒執(zhí)行的任務
    scheduler.add_job(
        failing_job, 
        IntervalTrigger(seconds=5), 
        id='resilient_task',
        name='彈性任務'
    )
    print("任務：'resilient_task' 已設置，每5秒執(zhí)行一次。")

    # 注冊錯誤監(jiān)聽器
    from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_EXECUTED
    scheduler.add_listener(my_listener, EVENT_JOB_ERROR | EVENT_JOB_EXECUTED)

    # 啟動調度器
    scheduler.start()
    print("調度器已啟動?？梢試L試 Ctrl+C 中斷程序，再重新運行，任務會從上次中斷的地方繼續(xù)。")

    try:
        while True:
            time.sleep(1)
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown()
        print("調度器已停止。")