Python代码如何实现定时任务 Python代码使用Schedule模块的配置

import schedule
import time

def job():
    print("我正在执行一个定时任务...")

# 每天的10:30执行任务
schedule.every().day.at("10:30").do(job)

# 每10分钟执行一次任务
schedule.every(10).minutes.do(job)

# 每周一执行任务
schedule.every().monday.do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

import schedule
import time

def job():
    try:
        print("我正在尝试执行一个可能失败的任务...")
        # 模拟一个可能出错的操作
        result = 1 / 0
        print("任务成功完成！")
    except Exception as e:
        print(f"任务执行失败，错误信息：{e}")

schedule.every(5).seconds.do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

import schedule
import time
import threading

def job(task_id):
    print(f"任务 {task_id} 正在执行...")
    time.sleep(5)  # 模拟耗时操作
    print(f"任务 {task_id} 完成！")

def run_threaded(job_func, task_id):
    job_thread = threading.Thread(target=job_func, args=(task_id,))
    job_thread.start()

schedule.every(10).seconds.do(run_threaded, job, 1)
schedule.every(15).seconds.do(run_threaded, job, 2)

while True:
    schedule.run_pending()
    time.sleep(1)

import schedule
import time
import json
import os

CONFIG_FILE = "schedule_config.json"

def job(message):
    print(message)

def save_schedule():
    tasks = []
    for task in schedule.jobs:
        tasks.append({
            "interval": task.interval,
            "unit": task.unit,
            "next_run": task.next_run.isoformat(),
            "job_func": task.job_func.__name__,  # 仅保存函数名
            "args": task.args,
            "kwargs": task.kwargs
        })
    with open(CONFIG_FILE, "w") as f:
        json.dump(tasks, f)

def load_schedule():
    if not os.path.exists(CONFIG_FILE):
        return

    with open(CONFIG_FILE, "r") as f:
        tasks = json.load(f)

    for task in tasks:
        # 注意：这里需要根据job_func的名字重新注册任务
        if task["job_func"] == "job":
            getattr(schedule.every(task["interval"], task["unit"]), task["unit"]).do(job, *task["args"], **task["kwargs"])
            schedule.jobs[-1].next_run = datetime.fromisoformat(task["next_run"]) # 恢复下次执行时间
        # 可以添加更多job_func的处理

# 加载配置
load_schedule()

# 添加一些任务
schedule.every(10).seconds.do(job, "每10秒执行一次")
schedule.every().day.at("12:00").do(job, "每天中午12点执行")

# 保存配置
save_schedule()

while True:
    schedule.run_pending()
    time.sleep(1)