Generate a video with Seedance 2.0 Fast in Python

"""Generate a video with Seedance 2.0 Fast through FishToken, wait for it, save the result URL.

    pip install requests
    export FISHTOKEN_API_KEY=your-key-here
    python3 seedance_fast.py
"""
import os
import sys
import time

import requests

API = "https://fishtoken.ai/v1"
MODEL = "doubao-seedance-2-0-fast-260128"
HEADERS = {"Authorization": f"Bearer {os.environ['FISHTOKEN_API_KEY']}"}
PROMPT = ("Vertical shot: a woven rattan handbag turning slowly on a sunlit wooden table, "
          "soft shadows, shallow depth of field")

POLL_EVERY = 5      # seconds between status checks
MAX_WAIT = 10 * 60  # give up after ten minutes


def submit(prompt: str) -> str:
    """Submit a generation task and return its task ID."""
    r = requests.post(f"{API}/video/generations", headers=HEADERS,
                      json={"model": MODEL, "prompt": prompt}, timeout=60)
    r.raise_for_status()
    data = r.json()
    # PLACEHOLDER: the field that holds the task ID is defined in https://fishtoken.ai/docs
    # (commonly "id" or "task_id"). Print data once and adjust the line below if needed.
    task_id = data.get("id") or data.get("task_id")
    if not task_id:
        sys.exit(f"no task ID in response: {data}")
    return task_id


def wait_for(task_id: str) -> dict:
    """Poll the task until it finishes or MAX_WAIT passes."""
    deadline = time.time() + MAX_WAIT
    while time.time() < deadline:
        r = requests.get(f"{API}/video/generations/{task_id}", headers=HEADERS, timeout=60)
        r.raise_for_status()
        task = r.json()
        # PLACEHOLDER: check the task endpoint in https://fishtoken.ai/docs for the exact
        # status values and the field that carries the video URL, then adjust these two lines.
        status = str(task.get("status", "")).lower()
        if status in ("succeeded", "completed", "success", "done"):
            return task
        if status in ("failed", "error", "cancelled"):
            sys.exit(f"task {task_id} failed: {task}")
        print(f"  {status or 'pending'} ... checking again in {POLL_EVERY}s")
        time.sleep(POLL_EVERY)
    sys.exit(f"task {task_id} not finished after {MAX_WAIT}s")


def video_url(task: dict) -> str:
    """Find the first http(s) URL in the finished task, whatever key it sits under."""
    stack = [task]
    while stack:
        item = stack.pop()
        if isinstance(item, dict):
            stack.extend(item.values())
        elif isinstance(item, list):
            stack.extend(item)
        elif isinstance(item, str) and item.startswith("http"):
            return item
    sys.exit(f"no URL in finished task: {task}")


if __name__ == "__main__":
    task_id = submit(PROMPT)
    print(f"submitted task {task_id}")
    finished = wait_for(task_id)
    url = video_url(finished)
    with open("seedance_result.txt", "a", encoding="utf8") as f:
        f.write(f"{task_id}\t{url}\n")
    print(f"video ready: {url}")
    print("saved to seedance_result.txt")

    # Optional: download the file. Result URLs may expire, so do this soon after the task finishes.
    mp4 = requests.get(url, timeout=300)
    mp4.raise_for_status()
    with open(f"{task_id}.mp4", "wb") as f:
        f.write(mp4.content)
    print(f"downloaded {task_id}.mp4 ({len(mp4.content) // 1024} KB)")

$ python3 seedance_fast.py
submitted task TASK_ID
  pending ... checking again in 5s
  pending ... checking again in 5s
video ready: https://.../TASK_ID.mp4
saved to seedance_result.txt
downloaded TASK_ID.mp4 (... KB)