Vidu Python Polling and Download

pip install requests
export MODELSELL_BASE_URL="https://api.modelsell.com"
export MODELSELL_API_KEY="<your-modelsell-api-key>"

import os
import time
from pathlib import Path
from urllib.parse import quote, urlparse

import requests

BASE = os.environ.get("MODELSELL_BASE_URL", "https://api.modelsell.com").rstrip("/")
KEY = os.environ["MODELSELL_API_KEY"]
HEADERS = {"Authorization": f"Bearer {KEY}"}
PROMPT = "A red ceramic teapot on a wooden table in warm morning light."


def create_task(path, body, family="native"):
    response = requests.post(BASE + path, headers=HEADERS, json=body, timeout=300)
    response.raise_for_status()
    payload = response.json()
    task_id = payload.get("task_id") if family == "native" else payload.get("id")
    if not isinstance(task_id, str) or not task_id.startswith("task_"):
        raise RuntimeError("Creation returned no ModelSell public task ID")
    print(f"Save this task ID: {task_id}", flush=True)
    return task_id


def wait_for_task(task_id, family="native", max_wait=1800, interval=12):
    escaped_id = quote(task_id, safe="")
    path = (f"/ent/v2/tasks/{escaped_id}/creations"
            if family == "native" else f"/v1/videos/{escaped_id}")
    waiting = ({"created", "queueing", "processing"} if family == "native"
               else {"queued", "in_progress"})
    success = "success" if family == "native" else "completed"
    deadline = time.monotonic() + max_wait
    while time.monotonic() < deadline:
        remaining = deadline - time.monotonic()
        response = requests.get(
            BASE + path, headers=HEADERS,
            timeout=min(120, max(0.1, remaining)),
        )
        response.raise_for_status()
        task = response.json()
        state = task.get("state" if family == "native" else "status")
        if state == success:
            return task
        if state == "failed":
            detail = task.get("err_code") if family == "native" else task.get("error")
            raise RuntimeError(f"Task {task_id} failed: {detail}")
        if state not in waiting:
            raise RuntimeError(f"Unexpected state {state!r}; retain ID {task_id}")
        print(f"{task_id}: {state}", flush=True)
        time.sleep(min(interval, max(0, deadline - time.monotonic())))
    raise TimeoutError(f"Local wait expired; query this task later: {task_id}")


def result_url(task, family="native"):
    if family == "native":
        creations = task.get("creations") or []
        url = creations[0].get("url") if creations else None
    else:
        url = task.get("video_url") or (task.get("metadata") or {}).get("url")
    if not isinstance(url, str) or urlparse(url).scheme not in {"http", "https"}:
        raise RuntimeError("Successful task has no downloadable video URL")
    return url


def download_video(url, destination):
    target = Path(destination)
    temporary = target.with_suffix(target.suffix + ".part")
    try:
        # Result URLs may use a CDN: do not send the ModelSell API Key.
        with requests.get(url, stream=True, timeout=300) as response:
            response.raise_for_status()
            with temporary.open("wb") as output:
                for chunk in response.iter_content(65536):
                    if chunk:
                        output.write(chunk)
        temporary.replace(target)
    except Exception:
        temporary.unlink(missing_ok=True)
        raise


if __name__ == "__main__":
    # Set MODELSELL_TASK_ID to resume polling without creating another task.
    task_id = os.environ.get("MODELSELL_TASK_ID")
    if not task_id:
        task_id = create_task("/ent/v2/text2video", {
            "model": "viduq3-turbo", "prompt": PROMPT,
            "duration": 5, "resolution": "720p", "aspect_ratio": "16:9",
            "audio": False, "is_rec": False, "watermark": False,
        })
    task = wait_for_task(task_id)
    download_video(result_url(task), "vidu-output.mp4")

export MODELSELL_TASK_ID="<previously-returned-task-id>"
python vidu_example.py

task_id = create_task("/v1/videos", {
    "model": "viduq3-turbo", "prompt": PROMPT, "seconds": "5", "size": "720p",
    "metadata": {
        "vidu_operation": "text2video", "aspect_ratio": "16:9",
        "audio": False, "is_rec": False, "watermark": False,
    },
}, "generic")
task = wait_for_task(task_id, "generic")
download_video(result_url(task, "generic"), "vidu-generic-output.mp4")