Creatify Boreal API 怎么用？从提交任务到下载视频

TL;DR

• Creatify Boreal 已在 HiAPI 上线，可把提示词或参考图变成带配音和音效的视频，打开模型页即可开始。
• 720p 每秒 $0.0145，8 秒 $0.116；1080p 每秒 $0.043，5 秒 $0.215。查看分档价格与预算。
• 使用 HiAPI /v1/tasks 接口：提交任务拿 data.taskId，查询 data.status，成功后从 data.output 选择 type=video 的地址下载。

Creatify Boreal 能做什么？

• 纯文字生成商品特写或 UGC 口播，把镜头、台词和声音写进同一段提示词。
• 用 image_url 提供商品或人物参考图，再描述光线、动作和台词。
• 选择 720p、1080p 或 2K，时长为 1–20 秒整数；模型还支持 audio_url 输入自己的旁白。

查看完整视频

Vertical selfie-style UGC video: a young woman in a bright bedroom holds up a pastel lip balm and says to camera: "Okay, this is the only lip balm that survived my winter." Natural room tone.

用这个提示词生成 →

creatify-boreal，720p，aspect_ratio: "9:16"，请求 8 秒，$0.116。成片 720×1280，容器时长 8.04 秒，含非静音 AAC 音轨。

人物把润唇膏举到镜头附近，可以观察商品位置和自拍构图。音轨只验证为非静音，台词与口型尚未做听审。预填链接带入提示词；分辨率、比例和时长仍需选择。

从网页或 API 怎么开始？

网页路线

1. 打开模型页，粘贴上面的完整英文提示词。
2. 选择 720p、9:16、8 秒，查看预计费用。
3. 提交后查看成片，再决定是否调整镜头或台词。

API 路线

1. 创建 API Key，把它设为环境变量 HIAPI_API_KEY。
2. 向 POST /v1/tasks 提交下面的请求，立即保存 data.taskId。
3. 查询 GET /v1/tasks/{taskId}，等待终态，再下载视频。

curl：如何提交第一条视频？

Bash 示例：将固定业务键设为 BOREAL_IDEMPOTENCY_KEY，同一请求复用；720p、8 秒为 $0.116。

curl --fail-with-body --show-error --request POST 'https://api.hiapi.ai/v1/tasks' \
  --header "Authorization: Bearer $HIAPI_API_KEY" \
  --header "Idempotency-Key: ${BOREAL_IDEMPOTENCY_KEY:?Set a stable key for this job}" \
  --header 'Content-Type: application/json' \
  --data '{
  "model": "creatify-boreal",
  "input": {
    "prompt": "Vertical selfie-style UGC video: a young woman in a bright bedroom holds up a pastel lip balm and says to camera: \"Okay, this is the only lip balm that survived my winter.\" Natural room tone.",
    "resolution": "720p",
    "aspect_ratio": "9:16",
    "duration": 8
  }
}'

保存响应的 data.taskId，设为 TASK_ID 后查询：

curl --fail-with-body --show-error \
  "https://api.hiapi.ai/v1/tasks/$TASK_ID" \
  --header "Authorization: Bearer $HIAPI_API_KEY"

检查 data.status：success 时从 data.output 选择 type=video 的地址；fail 时处理 data.error；其他状态继续等待。HTTP 200 不表示任务完成。

Python：如何保留任务 ID、轮询并下载？

准备 Python 与 requests，设置 HIAPI_API_KEY 和 BOREAL_JOB_KEY。后者是你自己的稳定业务编号：同一条业务请求重启后继续使用同一个值；换提示词或参数时使用新编号。

代码把业务编号映射为固定的 Idempotency-Key，并保存请求指纹和 taskId。成功拿到 ID 后先立即打印，再尝试写文件；保存失败时保留终端里的 ID，设置 BOREAL_TASK_ID 后可恢复查询。

import hashlib
import json
import os
import sys
import time
from pathlib import Path
from urllib.parse import quote

import requests

BASE = "https://api.hiapi.ai/v1/tasks"
JOB_KEY = os.environ["BOREAL_JOB_KEY"]  # Keep the same value for the same business job.
KEY = hashlib.sha256(JOB_KEY.encode()).hexdigest()
HEADERS = {"Authorization": f"Bearer {os.environ['HIAPI_API_KEY']}"}
STATE = Path(f"boreal-{KEY}.json")
PAYLOAD = {
    "model": "creatify-boreal",
    "input": {
        "prompt": "Vertical selfie-style UGC video: a young woman in a bright bedroom holds up a pastel lip balm and says to camera: \"Okay, this is the only lip balm that survived my winter.\" Natural room tone.",
        "resolution": "720p",
        "aspect_ratio": "9:16",
        "duration": 8
    }
}
FINGERPRINT = hashlib.sha256(
    json.dumps(PAYLOAD, sort_keys=True).encode()
).hexdigest()

class TaskStateError(RuntimeError):
    def __init__(self, task_id):
        self.task_id = task_id

def save_state(task_id):
    temporary = STATE.with_suffix(".tmp")
    temporary.write_text(json.dumps({
        "fingerprint": FINGERPRINT, "taskId": task_id
    }), encoding="utf-8")
    temporary.replace(STATE)

def get_task_id():
    saved = None
    if STATE.exists():
        saved = json.loads(STATE.read_text(encoding="utf-8"))
        if saved["fingerprint"] != FINGERPRINT:
            raise RuntimeError("This job key belongs to a different request.")
    recovery_id = os.environ.get("BOREAL_TASK_ID")
    if saved and saved.get("taskId"):
        task_id = saved["taskId"]
        if recovery_id and recovery_id != task_id:
            raise RuntimeError(f"Recovery ID differs; keep taskId={task_id}")
        print(f"taskId={task_id}", flush=True)
        return task_id
    if recovery_id:
        print(f"taskId={recovery_id}", flush=True)
        try:
            save_state(recovery_id)
        except OSError:
            raise TaskStateError(recovery_id) from None
        return recovery_id
    if saved is not None:
        raise RuntimeError("Submission outcome unknown. Recover the original task ID before continuing.")
    save_state(None)  # Persist intent before the request; do not blindly submit again.
    try:
        response = requests.post(
            BASE, headers={**HEADERS, "Idempotency-Key": KEY},
            json=PAYLOAD, timeout=30,
        )
        response.raise_for_status()
        created = response.json()
        if created.get("code") != 200:
            raise ValueError("Task creation was rejected")
        task_id = created["data"]["taskId"]
        if not isinstance(task_id, str) or not task_id:
            raise ValueError("Missing task ID")
    except (requests.RequestException, ValueError, KeyError, TypeError):
        raise RuntimeError("No task ID captured. Keep the job state and recover the submission before retrying.") from None
    print(f"taskId={task_id}", flush=True)
    try:
        save_state(task_id)
    except OSError:
        raise TaskStateError(task_id) from None
    return task_id

def wait_for_video(task_id):
    deadline = time.monotonic() + 900
    url = f"{BASE}/{quote(task_id, safe='')}"
    while time.monotonic() < deadline:
        try:
            response = requests.get(url, headers=HEADERS, timeout=30)
        except requests.RequestException:
            time.sleep(10)
            continue
        if response.status_code == 429 or response.status_code >= 500:
            time.sleep(10)
            continue
        response.raise_for_status()
        result = response.json()
        if result.get("code") != 200:
            raise RuntimeError("Task query returned an application error")
        data = result["data"]
        if data["status"] == "success":
            video_url = next(
                (item["url"] for item in data.get("output", [])
                 if item.get("type") == "video" and item.get("url")), None,
            )
            if not video_url:
                raise RuntimeError("Successful task has no video URL")
            return video_url
        if data["status"] in ("fail", "failed"):
            raise RuntimeError(f"Task failed; keep taskId={task_id}; inspect data.error in a protected log.")
        time.sleep(10)
    raise TimeoutError(f"Polling limit reached; resume taskId={task_id}")

def download(video_url):
    target = Path(f"boreal-{KEY}.mp4")
    temporary = target.with_suffix(".part")
    with requests.get(video_url, stream=True, timeout=(10, 60)) as response:
        response.raise_for_status()
        media_type = response.headers.get("Content-Type", "").split(";", 1)[0].lower()
        if media_type not in ("video/mp4", "application/octet-stream"):
            raise RuntimeError("Unexpected download content type")
        prefix = bytearray()
        with temporary.open("wb") as output:
            for chunk in response.iter_content(1024 * 1024):
                if chunk:
                    prefix.extend(chunk[:max(0, 12 - len(prefix))])
                    output.write(chunk)
        if len(prefix) < 12 or prefix[4:8] != b"ftyp":
            raise RuntimeError("Downloaded file is not an MP4")
    temporary.replace(target)
    return target

def main():
    task_id = None
    try:
        task_id = get_task_id()
        path = download(wait_for_video(task_id))
        print(f"Saved {path.name}; taskId={task_id}")
    except Exception as error:
        task_id = task_id or getattr(error, "task_id", None)
        # Do not print raw response bodies, API keys or signed media URLs.
        if task_id:
            print(f"Stopped ({type(error).__name__}); keep taskId={task_id}; resume the same job.", file=sys.stderr)
        else:
            print("Stopped; retain the job state and any taskId printed above. Recover before resubmitting.", file=sys.stderr)
        raise SystemExit(1) from None

if __name__ == "__main__":
    main()

此例每隔 10 秒查询，900 秒是本地等待上限，不是生成耗时承诺；超时或下载失败后，复用同一业务编号和 ID 继续。请求已发出但响应丢失时，保留任务状态并先找回 ID；本地状态记录用于阻止盲目重复提交，不依赖请求头单独保证去重。

下载不把API认证头转发给成片存储地址；文件须通过内容类型和MP4文件头检查，再改为正式文件名。示例用于单进程，多进程服务还需为业务编号设置唯一约束与原子认领。

如何把商品图变成视频？

将下面的精华瓶参考图地址传给 image_url，保留完整提示词。aspect_ratio: "auto" 跟随参考图；此案例的成片是 9:16、720×1280，请求 720p、8 秒，费用 $0.116。

{
  "model": "creatify-boreal",
  "input": {
    "prompt": "Soft light moves across the serum bottle as a drop falls from the gold dropper and ripples; a gentle voice says: \"Calm skin, all day.\" Light airy ambience.",
    "resolution": "720p",
    "aspect_ratio": "auto",
    "duration": 8,
    "image_url": "https://static.hiapi.ai/gallery/2026/10/3a69e44a41585cb7.webp"
  }
}

在提示词指南查看输入图与完整成片。预填提示词后仍需手动上传参考图或填写 image_url。

哪些参数需要显式填写？

auto 在有参考图时跟随图片比例，没有参考图时使用 16:9；指定固定比例会对参考图居中裁切。传入音频时保留前 duration 秒，音频较短则补静音。

提交、查询或下载失败后怎么办？

• 提交返回 HTTP 400 时，检查必填字段、取值和媒体地址，再修正请求；不要把查询响应的 HTTP 200 当作视频已完成。
• data.status 为 fail 时，读取 data.error 并保留 taskId，修正原因后再发起新任务。
• 查询超时、HTTP 429 或服务端暂时错误时，等待后查询同一 taskId。
• 下载中断时，继续查询原任务并重新下载，完成后再把 .part 文件改为 .mp4。

接入业务前要处理什么？

• 每次提交显式填写 resolution、duration 和 aspect_ratio，让画幅与预算可预测。
• 服务端可在请求顶层配置 callback.url 与 callback.when: "final"，按 taskId 对 success、fail 通知去重；漏收通知时补查原任务。
• 成功后及时下载并转存到自己的存储，业务页面使用你保存的长期地址。
• 将业务编号、请求内容与 taskId 一起保存，重启或下载失败时继续原任务。

常见问题

文生视频和图生视频需要不同模型名吗？

不需要，都使用 creatify-boreal；图生视频在 input 中增加 image_url。有自己的旁白时还可使用 audio_url。

8 秒视频为什么显示 8.04 秒？

这里的示例请求时长是 8 秒，视频容器显示 8.04 秒；按请求时长计费，720p 为 $0.116。

怎样降低尝试提示词的预算？

先用 720p、8 秒确认方向。按每条 $0.116 计算，$10 可覆盖 86 条，总计 $9.976；最终选择的画面再考虑更高分辨率。

程序中断后是否需要重新提交？

优先复用已保存的 taskId。保存文件失败时，使用终端输出的 ID 设置 BOREAL_TASK_ID；如果提交结果不明，先查找原任务再继续。

接着看什么？

• 提示词指南：5 个真实视频场景
• 价格与预算：720p、1080p 与 2K 档位
• Creatify Boreal 参数与请求文档

复制润唇膏案例的提示词，用这个提示词生成 →；选择 720p、9:16、8 秒后提交。