Seedance 2.5 · Token · API

import json
import os
import time
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

BASE_URL = "https://nezhagate.com"
API_KEY = os.environ["NEZHAGATE_API_KEY"]


def request(path, payload=None):
    data = json.dumps(payload).encode() if payload is not None else None
    req = Request(BASE_URL + path, data=data, headers={
        "Authorization": "Bearer " + API_KEY,
        "Content-Type": "application/json",
    })
    with urlopen(req, timeout=60) as response:
        return json.load(response)


def submit(payload):
    # Do not automatically retry a POST: its outcome may be unknown.
    job = request("/v1/videos/generations", payload)
    job_id = job["id"]
    print("Submitted:", job_id, flush=True)
    return job_id


def wait_for(job_id):
    deadline = time.monotonic() + 1800
    while time.monotonic() < deadline:
        try:
            job = request("/v1/videos/jobs/" + job_id)
        except HTTPError as exc:
            if exc.code != 429 and exc.code < 500:
                raise
            time.sleep(10)
            continue
        except (URLError, TimeoutError):
            time.sleep(10)
            continue
        progress = job.get("progress", {})
        print(job_id, job["status"], progress.get("phase", ""), flush=True)
        if job["status"] == "succeeded":
            return job
        if job["status"] == "failed":
            raise RuntimeError(json.dumps(job.get("error"), ensure_ascii=False))
        time.sleep(5)
    raise TimeoutError("Keep polling this job; do not resubmit: " + job_id)


if __name__ == "__main__":
    # Two separately billed tasks. Save the printed IDs to resume polling.
    draft_id = submit({
        "model": "seedance-2.5-token-draft",
        "prompt": "A sunrise over the sea",
        "duration": 5,
        "resolution": "480p",
    })
    draft = wait_for(draft_id)
    print("Draft video:", draft["data"][0]["url"])

    final_id = submit({
        "model": "seedance-2.5-token",
        "draft_task_id": draft_id,
        "resolution": "1080p",
    })
    final = wait_for(final_id)
    print("Final video:", final["data"][0]["url"])