如何按每小时一次调度Python脚本以分批拉取API数据

针对你的需求，有两种常用实现方式，按需选择：

方式一：脚本内自带休眠调度（一次性启动，自动跑完6批） #

这种方式让脚本自主控制批次间隔，无需外部调度工具，适合希望一次性启动就完成全部任务的场景。核心思路是：每次拉取150条数据后休眠1小时，重复6次；同时建议记录拉取进度，避免进程意外中断后重复拉取。

示例代码：

import time
import json
from your_api_module import fetch_data  # 替换成你实际的API拉取函数

# 进度记录文件，跟踪已完成的批次
PROGRESS_FILE = "pull_progress.json"

def load_progress():
    try:
        with open(PROGRESS_FILE, "r") as f:
            return json.load(f)
    except FileNotFoundError:
        return {"completed_batches": 0}

def save_progress(completed):
    with open(PROGRESS_FILE, "w") as f:
        json.dump({"completed_batches": completed}, f)

def main():
    progress = load_progress()
    completed = progress["completed_batches"]
    total_batches = 6
    batch_size = 150

    while completed < total_batches:
        print(f"开始拉取第 {completed+1} 批数据（共{total_batches}批）")
        # 替换为你实际拉取150条日期数据的逻辑
        fetch_data(batch_size=batch_size, batch_num=completed+1)
        
        completed += 1
        save_progress(completed)
        
        if completed < total_batches:
            print(f"第 {completed} 批拉取完成，休眠1小时...")
            time.sleep(3600)  # 休眠3600秒=1小时

    print("所有批次数据拉取完成！")
    # 可选：完成后删除进度文件
    # import os
    # os.remove(PROGRESS_FILE)

if __name__ == "__main__":
    main()

注意：

• 替换your_api_module和fetch_data为你实际的API调用模块与函数
• 进度文件可避免进程意外终止后重复拉取已完成批次
• 建议在fetch_data中加入错误重试逻辑（比如用tenacity库），避免单次API请求失败导致批次中断

方式二：系统级定时任务（更可靠，适合长期调度） #

如果担心脚本休眠期间进程被意外终止，推荐用系统自带的定时任务工具，每小时启动一次脚本，每次拉取150条数据，共执行6次。

Linux/macOS（用cron） #

1. 编辑crontab：crontab -e
2. 添加定时规则，比如从9点开始每小时执行一次，共6次：

# 让脚本通过进度文件自主判断批次，无需手动指定
0 */1 * * * /usr/bin/python3 /path/to/your_script.py

脚本内通过进度文件判断是否还有未完成批次，没有则直接退出。

Windows（任务计划程序） #

1. 打开「任务计划程序」，创建基本任务
2. 设置触发器为「每天」，起始时间设为你需要的时刻，重复间隔1小时，重复次数5次（加上首次共6次）
3. 设置操作为「启动程序」，选择Python.exe路径，添加参数为你的脚本路径
4. 同样建议脚本内加入进度记录逻辑，避免重复拉取

关键注意事项 #

• API限流校验：拉取前可检查当前小时内的请求次数，避免触发限流被封禁
• 日志记录：将每批次拉取的结果（成功/失败、拉取数量）写入日志文件，方便排查问题
• 幂等性：确保API拉取接口是幂等的，即使重复拉取同一批次也不会产生重复数据或错误

内容的提问来源于stack exchange，提问作者nYuker_98 D