You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

使用schedule库:脚本启动滞后或错过计划时立即执行任务

解决方案

核心思路

通过本地文件持久化记录上次执行时间,结合启动时的时间校验逻辑,同时满足两个需求:

  1. 非计划时间启动脚本时立即执行main()
  2. 因断电等情况重启后,检测到错过的计划任务时立即执行main()

完整代码实现

import schedule
import time
import datetime
import os

# 定义所有计划执行的时间点
SCHEDULE_TIMES = ["00:00", "03:00", "06:00", "09:00", "12:00", "15:00", "18:00", "21:00"]

def get_last_run_time():
    """读取上次执行main的时间,文件不存在则返回一个初始时间"""
    if os.path.exists("last_run.txt"):
        with open("last_run.txt", "r") as f:
            try:
                return datetime.datetime.strptime(f.read().strip(), "%Y-%m-%d %H:%M:%S")
            except:
                # 文件内容无效时返回很早的初始时间
                return datetime.datetime(2000, 1, 1)
    else:
        return datetime.datetime(2000, 1, 1)

def update_last_run_time():
    """更新上次执行main的时间到本地文件"""
    with open("last_run.txt", "w") as f:
        f.write(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

def main():
    # 这里编写你的业务逻辑
    print(f"执行main函数:{datetime.datetime.now()}")
    # 执行完成后更新上次执行时间
    update_last_run_time()

def check_and_run_missed_or_immediate():
    """启动时检查是否需要立即执行main"""
    now = datetime.datetime.now()
    last_run = get_last_run_time()
    
    # 生成今天和昨天的所有计划时间,处理跨天错过任务的情况
    today_schedules = [datetime.datetime.combine(now.date(), datetime.datetime.strptime(t, "%H:%M").time()) for t in SCHEDULE_TIMES]
    yesterday_schedules = [dt - datetime.timedelta(days=1) for dt in today_schedules]
    all_schedules = yesterday_schedules + today_schedules
    
    # 判断是否有错过的任务:计划时间在最后一次执行之后、当前时间之前
    has_missed_tasks = any(last_run < schedule_time < now for schedule_time in all_schedules)
    # 判断是否在非计划时间点启动
    is_in_non_schedule_time = now.strftime("%H:%M") not in SCHEDULE_TIMES
    
    if has_missed_tasks or is_in_non_schedule_time:
        main()

if __name__ == "__main__":
    # 启动时先执行检查逻辑
    check_and_run_missed_or_immediate()
    
    # 批量创建计划任务
    for time_str in SCHEDULE_TIMES:
        schedule.every().day.at(time_str).do(main)
    
    # 持续运行调度器
    while True:
        schedule.run_pending()
        time.sleep(1)

代码说明

  • 持久化时间记录:用last_run.txt存储上次执行main的时间,脚本重启后能读取历史记录,判断是否存在错过的任务。
  • 启动检查逻辑:
    1. 同时生成今天和昨天的计划时间,避免跨天的错过任务被忽略(比如昨天21:00的任务,今天00:30重启脚本)。
    2. 两种触发立即执行的场景:当前时间不在计划时间点内、存在未执行的计划任务。
  • 批量设置计划任务:用列表统一管理所有时间点,循环创建任务,避免重复冗余代码。

内容的提问来源于stack exchange,提问作者Saeed

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.16 17:54:56