使用Python schedule库实现工作日OAID定时下载并规避重复执行咨询

1. 定时规则错误修复 #

你当前的schedule配置写法不符合库的API规范，无法实现「周一到周五0点到18点执行」的需求：

• 原写法monday.to(5).days的链式调用不被schedule支持，正确的工作日配置可以直接给周一到周五分别注册任务，或者在任务入口增加工作日判断
• 原写法to(15).hours会导致15点后就停止执行，和你要求的18点截止不符
• 没有配置任务触发间隔，会导致默认每分钟触发一次任务，产生大量重复执行

优化后的基础依赖与全局配置示例：

import os
import time
import schedule
import datetime
from more_itertools import chunked
from concurrent.futures import ProcessPoolExecutor

# 提前定义全局参数，避免作用域报错
TARGET_AZ_CONTAINER = "你的容器名"
AZ_SUBFOLDER = "你的子文件夹路径"
# 已完成任务持久化存储，小批量场景用文本文件即可，也可替换为sqlite/redis
FINISHED_OAID_PATH = "./finished_oaids.txt"

2. 新增去重逻辑，避免重复下载 #

原代码没有任何已完成任务的标记能力，每次触发都会重新下载全量ID的文件，需要增加持久化的成功状态记录：

def load_finished_oaids():
    if not os.path.exists(FINISHED_OAID_PATH):
        return set()
    with open(FINISHED_OAID_PATH, "r", encoding="utf-8") as f:
        return set(line.strip() for line in f if line.strip())

def save_finished_oaid(oaid):
    with open(FINISHED_OAID_PATH, "a", encoding="utf-8") as f:
        f.write(f"{oaid}\n")

3. 优化下载任务逻辑，补全异常处理与结果校验 #

原代码没有处理任务执行结果，也没有异常捕获，无法判断下载是否成功，也没法更新完成状态：

def download_and_upload_wrapper(oaid, container, subfolder):
    try:
        # 调用你原有下载逻辑
        download_and_upload(oaid, container, subfolder)
        # 下载成功后标记为已完成
        save_finished_oaid(oaid)
        return True, oaid
    except Exception as e:
        print(f"下载{oaid}失败，错误信息：{str(e)}")
        return False, oaid

def do_download_job():
    # 先过滤已完成的ID，只处理待执行的任务
    all_oaids = ['id1', 'id2', 'others..']
    finished_oaids = load_finished_oaids()
    pending_oaids = [oid for oid in all_oaids if oid not in finished_oaids]
    if not pending_oaids:
        print("所有任务已完成，停止本次执行")
        # 全量完成后取消定时任务，避免后续空跑
        return schedule.CancelJob
    
    # 进程池大小按需调整，不需要占满全部CPU，避免IO等待浪费资源
    max_workers = max(2, os.cpu_count() // 2)
    with ProcessPoolExecutor(max_workers=max_workers) as ex:
        # 按进程数分块提交任务
        for chunk in chunked(pending_oaids, max_workers):
            tasks = [ex.submit(download_and_upload_wrapper, oid, TARGET_AZ_CONTAINER, AZ_SUBFOLDER) for oid in chunk]
            # 等待当前块所有任务完成
            for task in tasks:
                task.result()

4. 修正定时任务注册逻辑 #

def main():
    # 配置周一到周五，0点到18点之间每2小时执行一次，可按需调整间隔
    for weekday in (schedule.every().monday, schedule.every().tuesday, schedule.every().wednesday, schedule.every().thursday, schedule.every().friday):
        weekday.hour.at(":00").do(do_download_job).tag("download_job")
    
    # 启动调度
    while True:
        # 非窗口期跳过执行
        now = datetime.datetime.now()
        if 0 <= now.hour < 18:
            schedule.run_pending()
        # 每分钟检查一次调度
        time.sleep(60)

额外优化建议 #

• 如果ID数量很大，可以把完成状态存在redis或者sqlite里，性能比纯文本文件好很多
• 可以增加失败重试机制，给每个ID设置最大重试次数，避免单次网络波动导致永久失败
• 如果你不需要窗口期内多次执行，只是每天0点触发一次全量下载，直接把定时规则改成每天0点执行，加上工作日判断即可，不需要配置间隔执行
• 注意进程池的任务不要传不可序列化的参数，否则会触发pickle报错

内容的提问来源于stack exchange，提问作者3nomis