在 HiAPI 上使用 Kandinsky 6.0 Pro 文本生视频：curl 与 Python 教程

kandinsky-6.0-pro/text-to-video 请求的作用

kandinsky-6.0-pro/text-to-video 可将文本提示词转换为短视频片段。你发送提示词和少量生成参数，模型会生成一个可保存并播放的视频文件。这就是整个产品：输入提示词，输出视频片段。

关键的架构细节是：这不是同步调用。你不会在第一次请求的响应体中直接获得视频。相反，你会创建一个任务，API 返回该任务的标识符，模型随后在后台工作。你的代码需要检查任务状态，直到它报告完成，此时结果可供下载。

这种模式有几个你从一开始就应围绕其设计的后果：

• 客户端必须轮询。单次 POST 无法返回完成的视频，因此任何集成都需要状态轮询循环或类似 webhook 的重新检查。
• 请求在你这边应具有幂等性。如果脚本在提交后崩溃，你不会希望悄悄创建第二个任务。请捕获任务标识符并持久化保存。
• 延迟是变化的。视频生成比图像或文本调用慢，因此等待是正常现象，而非缺陷。

本教程的其余部分按此生命周期顺序展开：准备、用 curl 提交、轮询、下载，然后用 Python 重建相同流程。请将代码视为可工作的骨架——端点路径和字段名应始终对照该模型的 API 参考进行确认，因为这些细节会随平台演进而变化。

发送第一个请求之前

首次调用前需要准备好四件事。

1. HiAPI 账户和 API 密钥

使用从 HiAPI 账户签发的 API 密钥进行认证。请尽可能将密钥排除在源代码管理和 shell 历史记录之外。通常的做法是将其导出为环境变量，并让代码从那里读取：

export HIAPI_API_KEY="your-key-here"

以下每个请求都假设密钥作为 bearer token 放在 Authorization 请求头中发送。如果你尚未创建密钥，请先在账户仪表板中完成此操作。

2. HTTP 客户端

用 curl 足以验证请求是否可用。对于任何需要重复执行的操作，请使用 Python 3 和 requests 库——只需一个依赖，代码可读，且以后易于放入调度器或服务中：

python3 -m pip install requests

3. 存放输出的位置

视频文件比 JSON 响应大得多。请事先决定是写入本地目录还是流式传输到对象存储。本教程中，使用本地 ./output 目录即可：

mkdir -p output

4. 熟悉你的提示词

在编写代码之前写好提示词。一个具体、描述性的提示词——主体、动作、镜头行为、视觉风格——为你提供了一个基线，当出现问题时可用于比较。当提示词是脚本中命名的常量，而非在 shell 提示符下输入的字符串时，一次只改变一个变量会容易得多。

以上就是完整的前置条件列表。无需安装 SDK，也没有构建步骤；平台通过普通 HTTPS 访问。

可用的 curl 请求

第一个目标是获得一条返回任务（而非视频）的命令。在添加任何代码之前先让它正常工作。

请求由三部分组成：模型的端点、Authorization 请求头，以及携带提示词和生成参数的 JSON 请求体。

curl -X POST "https://api.hiapi.ai/v1/models/kandinsky-6.0-pro/text-to-video/jobs" \
  -H "Authorization: Bearer $HIAPI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "A slow drone shot over a misty pine forest at sunrise, soft golden light, cinematic",
    "duration_seconds": 5,
    "aspect_ratio": "16:9"
  }'

仔细阅读响应。你需要寻找：

• 一个任务标识符——你将用于轮询的句柄。它可能命名为 id、job_id 或类似名称。
• 一个初始的状态字段，通常类似于 queued 或 processing。
• 任何回显的参数，可用于确认请求体按你预期的方式被解析。

返回任务标识符和非错误状态的响应意味着请求已被接受。如果该响应中直接包含视频文件，则意味着发生了与本文描述不同的事情。

在基于契约构建之前先确认它

上述字段名——端点路径、duration_seconds、aspect_ratio、标识符键——是最可能因模型和时间而变化的细节。可靠的真相来源是该模型的 API 参考页面。检查一次，将确切的名称作为常量复制到脚本中，并调整代码结构，使名称变更成为一行编辑，而非在整个代码库中搜索。

如果收到 4xx，错误响应体比状态码提供更多信息。401 表示密钥错误或缺失；422 类验证错误通常会指出有问题的字段。

提交并轮询任务直至完成

提交调用可用后，生命周期为：提交、捕获标识符、轮询、在终态停止。

提交

上一节的 POST 创建任务。无论代码接下来做什么，都应将返回的标识符存储在持久位置——脚本中的变量、任务表中的一行、队列中的一条消息。如果进程在收到提交响应后、持久化标识符之前死亡，你将失去轮询该任务的能力。

轮询

轮询是对任务资源发起 GET 请求，使用你捕获的标识符：

curl "https://api.hiapi.ai/v1/models/kandinsky-6.0-pro/text-to-video/jobs/$JOB_ID" \
  -H "Authorization: Bearer $HIAPI_API_KEY"

响应会报告任务当前状态。实际使用中你会看到少量状态：排队或待处理、处理中、已完成，以及某种形式的失败或取消。请围绕这些类别而非确切的字符串设计循环——将 API 返回的任何内容映射到你代码中的 running、done、failed，这样重命名不会破坏你的重试逻辑。

决定等待多久

两个数字很重要：轮询间隔和整体超时。

• 间隔。每隔几秒轮询一次，不要紧密循环。紧密循环不会带来额外好处，还可能触发速率限制。对于视频任务，请求之间间隔数秒是合理的。
• 超时。设置上限。视频生成需要时间，但“永远等待”不是生产脚本可接受的答案。选择一个最大的实际耗时预算，并将超出视为可记录并有意识地重试的失败。

一种能经受实际使用的模式

在生产环境中，轮询步骤通常与提交步骤解耦，而非放在一个休眠的进程中。工作进程提交、记录任务并退出；定期的重新检查会拾取未完成的任务并轮询它们。这对重启更健壮，也更容易观察。对于本教程，单个进程中的简单循环即可——只需将它组织为函数而非内联代码，以后将其移到定期调度的工作进程就会很简单。

下载生成的视频片段

完成的任务还不是你磁盘上的文件。已完成任务的状态响应会指向输出——通常是一个 URL，有时是包含 URL 的嵌套对象。读取该字段并将其作为下载步骤的输入。

curl -L "$OUTPUT_URL" -o output/clip.mp4

请检查以下几点以确认下载确实成功：

• 跟随重定向。输出 URL 经常通过重定向提供；curl 用 -L 处理，Python 用 allow_redirects=True 处理。
• 验证内容类型。完成的任务应产生视频载荷。如果保存为 clip.mp4 的却是 JSON 或 HTML 错误页面，说明下载出了问题，而非生成过程。
• 检查大小。零字节或可疑的小文件是下载失败。如果存在响应头，将字节数与响应头声明的大小进行比较。
• 考虑过期问题。输出 URL 通常有时间限制。完成后立即下载，如果需要持久性，请将文件复制到你自己的存储中，而不是将 URL 作为永久引用保存。

对于教程之外的任何用途，请将下载流式传输到目标位置，而非将整个文件加载到内存中，并先写入临时文件名再重命名——这样未完成的下载永远不会对下游代码看起来像完整文件。

用 Python 实现同样的请求

以下是将 curl 流程重建为一个可调用函数：提交、轮询、下载。需要调整的名称已收集在顶部作为常量。

import os
import time
import requests

API_BASE = "https://api.hiapi.ai/v1"
MODEL_PATH = "/models/kandinsky-6.0-pro/text-to-video"
API_KEY = os.environ["HIAPI_API_KEY"]

HEADERS = {"Authorization": f"Bearer {API_KEY}"}

RUNNING_STATES = {"queued", "pending", "processing", "running"}
DONE_STATES = {"completed", "succeeded", "done"}
FAILED_STATES = {"failed", "cancelled", "error"}


def generate_clip(prompt, out_path, timeout_s=600, interval_s=5):
    # 1. submit
    resp = requests.post(
        API_BASE + MODEL_PATH + "/jobs",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={"prompt": prompt, "duration_seconds": 5, "aspect_ratio": "16:9"},
        timeout=30,
    )
    resp.raise_for_status()
    job = resp.json()
    job_id = job["id"]
    print(f"submitted job {job_id}")

    # 2. poll
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        status_resp = requests.get(
            f"{API_BASE}{MODEL_PATH}/jobs/{job_id}",
            headers=HEADERS,
            timeout=30,
        )
        status_resp.raise_for_status()
        data = status_resp.json()
        state = str(data.get("status", "")).lower()

        if state in DONE_STATES:
            break
        if state in FAILED_STATES:
            raise RuntimeError(f"job {job_id} ended in state {state}: {data}")
        time.sleep(interval_s)
    else:
        raise TimeoutError(f"job {job_id} did not finish within {timeout_s}s")

    # 3. download
    url = data["output"]["url"] if isinstance(data.get("output"), dict) else data["output"]
    with requests.get(url, stream=True, allow_redirects=True, timeout=60) as r:
        r.raise_for_status()
        tmp = out_path + ".part"
        with open(tmp, "wb") as f:
            for chunk in r.iter_content(chunk_size=1 << 16):
                f.write(chunk)
        os.replace(tmp, out_path)
    print(f"saved {out_path}")
    return job_id


if __name__ == "__main__":
    generate_clip(
        "A slow drone shot over a misty pine forest at sunrise, soft golden light, cinematic",
        "output/clip.mp4",
    )

为什么这样写

• 用状态集合而非精确匹配。将返回的状态映射到 running/done/failed，意味着新的中间状态不会破坏循环。
• 单调截止时间。time.monotonic() 不受时钟调整影响，因此超时是准确的。
• 原子写入。写入 .part 再重命名，意味着崩溃的下载永远不会留下看起来有效的截断文件。
• 返回任务 ID。调用者可以记录、存储，并在需要时再次轮询——这是将轮询移出此进程的第一步。

请调整端点路径、请求字段名和响应键，以匹配模型的 API 参考。其余一切——循环、超时、下载——都是通用的。

错误状态与调试清单

当运行失败时，先隔离阶段再改变任何东西。每个阶段都有不同的失败特征。

认证失败

提交时出现 401 或 403 意味着密钥缺失、错误或未发送。检查环境变量在运行代码的 shell 或进程中是否确实已设置——在一个终端导出的密钥对 cron 任务是不可见的。打印密钥的前几个字符以确认其已解析，绝不要打印完整密钥。

请求格式错误

提交时出现 400 或 422 类错误意味着请求体未通过验证。响应体通常会指出字段。常见原因：参数拼写错误、时长超出支持范围，或提示词为空。逐字段将你的 JSON 与 API 参考进行比较；单个重命名的键是最常见的原因。

任务卡在运行状态

如果轮询永远不到达终态，请确认三件事：你轮询的标识符是你提交的那个；轮询间隔不是过于频繁以致被速率限制；超时对模型来说足够长。在每次轮询时记录原始状态响应，以便查看状态是在推进还是冻结。

任务完成但下载失败

任务完成但下载失败是文件处理问题，而非生成问题。检查是否跟随重定向、URL 是否未过期，以及内容类型是视频而非 JSON。用 curl -L 单独尝试该 URL，将其与脚本分开排查。

同一提示词反复失败

如果一个提示词持续失败而其他成功，先怀疑内容政策或提示词长度，而非基础设施。改变一个变量——缩短提示词、简化场景描述——然后重试。

一条通用规则

为每个任务记录四件事：任务标识符、最终状态、实际耗时，以及下载的 HTTP 状态。这四个值能在不到一分钟内将大多数“它没工作”的报告转化为具体、可修复的阶段。