视频生成 API 怎么用 Python 轮询?间隔、超时、一直 PENDING 与不重复扣费(2026)

import json, time, pathlib, requests

API = "https://cocodot.co/api/ai/video/generations"
H = {"Authorization": "Bearer cocodot-..."}
LOG = pathlib.Path("video_tasks.jsonl")  # 每提交成功一条就追加一行,脚本断了也能续查

def submit(body):
    # 同一账户的提交逐条处理,单次可能要等较久:超时给足,超时后不要自动重投
    r = requests.post(API, json=body, headers=H, timeout=120)
    if not r.ok:
        raise RuntimeError(f"{r.status_code} {r.text}")  # 各类报错的处理见 /hub/video-api-error-codes-fix
    tid = r.json()["id"]
    with LOG.open("a", encoding="utf-8") as f:
        f.write(json.dumps({"id": tid, "model": body["model"], "at": time.time()}, ensure_ascii=False) + "\n")
    return tid

def wait(tid, interval=15, deadline_s=3600):
    end = time.time() + deadline_s
    while time.time() < end:
        time.sleep(interval)  # 同一任务 10 秒内重复查不会拿到新状态
        try:
            r = requests.get(f"{API}/{tid}", headers=H, timeout=90)
            r.raise_for_status()
            t = r.json()
        except requests.RequestException:
            continue  # 网络抖动:接着查同一个任务,绝不重新提交
        if t["status"] == "SUCCEEDED":
            return t
        if t["status"] == "FAILED":
            raise RuntimeError(f"{tid} FAILED: {t.get('error')}")
    raise TimeoutError(f"{tid} 到截止时间仍是 PENDING:稍后用 resume() 补查,不要重投")

def download(tid, path):
    for _ in range(2):
        response = requests.get(f"{API}/{tid}", headers=H, timeout=90)
        response.raise_for_status()
        t = response.json()  # 每次下载前重新取链接
        r = requests.get(t["videoUrl"], stream=True, timeout=300)
        if r.ok:
            with open(path, "wb") as f:
                for chunk in r.iter_content(1 << 20):
                    f.write(chunk)
            return path
        time.sleep(5)
    raise RuntimeError(f"{tid} 下载失败,记下任务 ID 稍后再试")

def resume():
    """脚本重启或提交超时后:先对账,只查不重投。"""
    response = requests.get(API, headers=H, timeout=60)
    response.raise_for_status()
    recent = response.json()  # 最近 50 条,并非完整历史
    for x in recent:
        if x["status"] == "PENDING":
            print("仍在生成:", x["id"], x["model"], x["createdAt"], (x.get("prompt") or "")[:40])
    known = {json.loads(l)["id"] for l in LOG.read_text(encoding="utf-8").splitlines()} if LOG.exists() else set()
    for x in recent:
        if x["id"] not in known:
            print("本地没记、服务端有:", x["id"], x["status"], x["createdAt"])