使用 ONE 任務 API 生成影片

POST https://www.getapi.one/v1/videos
GET https://www.getapi.one/v1/videos/{task_id}
GET https://www.getapi.one/v1/videos/{task_id}/content

curl --fail-with-body https://www.getapi.one/v1/videos \
  -H "Authorization: Bearer $GETAPI_ONE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "<OPENAI_VIDEO_MODEL_ID_FROM_CURRENT_CATALOG>",
    "prompt": "A five-second locked-camera product turntable on a neutral background"
  }' > video-submit.json

import json
import os
import time
from pathlib import Path
from urllib.parse import quote
from urllib.request import Request, urlopen

base_url = "https://www.getapi.one"
api_key = os.environ["GETAPI_ONE_API_KEY"]
submitted = json.loads(Path("video-submit.json").read_text(encoding="utf-8"))
task_id = submitted.get("id") or submitted.get("task_id")
if not task_id:
    raise ValueError("Submit response did not contain id or task_id")

safe_id = quote(task_id, safe="")
headers = {"Authorization": f"Bearer {api_key}"}
for attempt in range(30):
    request = Request(f"{base_url}/v1/videos/{safe_id}", headers=headers)
    with urlopen(request, timeout=30) as response:
        task = json.load(response)
    status = task.get("status")
    if status == "completed":
        break
    if status == "failed":
        raise RuntimeError(f"video task failed: {task.get('error')}")
    if status not in {"queued", "in_progress"}:
        raise RuntimeError(f"unknown non-success status: {status}")
    if attempt == 29:
        raise TimeoutError("video was not complete after 30 status checks")
    time.sleep(10)

content_request = Request(
    f"{base_url}/v1/videos/{safe_id}/content",
    headers=headers,
)
MAX_VIDEO_BYTES = 512 * 1024 * 1024
CHUNK_BYTES = 1024 * 1024
output_path = Path("result.mp4")
created_output = False
with urlopen(content_request, timeout=60) as response:
    content_type = response.headers.get_content_type()
    if not content_type.startswith("video/"):
        raise ValueError(f"unexpected content type: {content_type}")
    try:
        output = output_path.open("xb")
        created_output = True
        with output:
            total_bytes = 0
            while chunk := response.read(CHUNK_BYTES):
                total_bytes += len(chunk)
                if total_bytes > MAX_VIDEO_BYTES:
                    raise ValueError("video exceeds the local 512 MiB limit")
                output.write(chunk)
    except BaseException:
        if created_output:
            output_path.unlink(missing_ok=True)
        raise