视频任务提交、轮询与下载完整流程

更新于 2026-09-28；软件与模型验证范围见正文。

本文使用 Seedance 2.x 的 /api/v3/contents/generations/tasks；不能与 1.x 的 /volc/v1 查询混用。该流程已通过本地状态与下载测试，尚未发起真实视频生成。

第 1 步：选择模型与生成规格

进入 apizt 控制台 → API 令牌 → 新建，选择支持目标模型及协议的分组，设置额度后复制令牌。已有令牌可继续使用；模型目录和权限不是同一回事。查看创建分组令牌的图文步骤。

python3 -m pip install requests
export tokenba_API_KEY="YOUR_tokenba_API_KEY"
export tokenba_VIDEO_MODEL="doubao-seedance-2-0-260128"

此 ID 在当前价格目录列有 /api/v3 协议；确认令牌权限后选择。新版模型可以更换 ID，但 duration/ratio 等仍需按该模型校验。先查看 创建任务参数。

第 2 步：创建、保存 ID、轮询并下载

下载完整 Python 示例 video-workflow.py；这是可运行代码，令牌通过环境变量读取，不是含凭据的配置模板。

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

import requests

BASE = "https://api.apizt.com/api/v3/contents/generations/tasks"


def wait_for_video(task_id, headers, max_seconds=600):
    deadline = time.monotonic() + max_seconds
    url = BASE + "/" + quote(task_id, safe="")
    while time.monotonic() < deadline:
        response = requests.get(url, headers=headers, timeout=(10, 30))
        response.raise_for_status()
        task = response.json()
        status = task.get("status")
        if status == "succeeded":
            video_url = task.get("content", {}).get("video_url")
            if not video_url or urlparse(video_url).scheme != "https":
                raise ValueError("No HTTPS video URL in successful task")
            return video_url
        if status in {"failed", "cancelled", "canceled", "expired"}:
            raise RuntimeError("Video task ended: " + status + " " + str(task.get("error", "")))
        if status not in {"queued", "running"}:
            raise RuntimeError("Unknown task status: " + str(status))
        time.sleep(min(5, max(0, deadline - time.monotonic())))
    raise TimeoutError("Polling timed out; resume querying this task ID instead of creating another")


def download_video(url, target):
    if urlparse(url).scheme != "https":
        raise ValueError("Expected HTTPS download URL")
    target = Path(target)
    temporary = target.with_suffix(target.suffix + ".part")
    try:
        with requests.get(url, stream=True, timeout=(10, 120)) as response:
            response.raise_for_status()
            with temporary.open("wb") as output:
                for chunk in response.iter_content(1024 * 1024):
                    if chunk:
                        output.write(chunk)
        if temporary.stat().st_size == 0:
            raise ValueError("Empty video download")
        temporary.replace(target)
    except Exception:
        temporary.unlink(missing_ok=True)
        raise


def main():
    headers = {"Authorization": "Bearer " + os.environ["tokenba_API_KEY"]}
    response = requests.post(
        BASE,
        headers=headers,
        json={
            "model": os.environ["tokenba_VIDEO_MODEL"],
            "content": [{"type": "text", "text": "A small blue whale swimming slowly in clear water."}],
            "duration": 5,
            "ratio": "16:9",
        },
        timeout=(10, 60),
    )
    response.raise_for_status()
    task_id = response.json().get("id")
    if not task_id:
        raise ValueError("No task ID in creation response")
    Path("video-task-id.txt").write_text(task_id)
    print("task_id:", task_id, flush=True)
    download_video(wait_for_video(task_id, headers), "tokenba-video.mp4")
    print("saved: tokenba-video.mp4")


if __name__ == "__main__":
    main()

第 3 步：理解状态与恢复

创建成功只代表收到任务 ID；必须等成功状态并确认下载文件非空。下载使用独立请求，不向媒体服务器发送 apizt Key。下载中断留下的 .part 文件会清理。

第 4 步：核对费用与接口

单任务查询 · 任务列表 · 视频计费单位。任务重复提交可能重复产生消耗，轮询超时不等于任务失败；以日志和当前分组规则核对。

开始实践：注册并创建 API Key · 已有账号，进入控制台 · 查看当前模型价格。