完整轮询示例¶

#!/usr/bin/env bash
set -euo pipefail

API_BASE='https://ai.shanhui.tv'
: "${SHANHUI_API_KEY:?请先设置 SHANHUI_API_KEY}"

SUBMIT_RESPONSE=$(curl -fsS -X POST "$API_BASE/v1/video/generations" \
  -H "Authorization: Bearer $SHANHUI_API_KEY" \
  -H 'Content-Type: application/json' \
  -d '{
    "model": "doubao-seedance-2-0-fast-260128",
    "prompt": "清晨的湖面，薄雾散开，镜头缓慢向前",
    "metadata": {
      "duration": 5,
      "ratio": "16:9",
      "resolution": "720p",
      "watermark": false
    }
  }')

TASK_ID=$(printf '%s' "$SUBMIT_RESPONSE" | python3 -c \
  'import json,sys; d=json.load(sys.stdin); print(d.get("task_id") or d.get("id") or "")')

if [ -z "$TASK_ID" ]; then
  printf '%s\n' "$SUBMIT_RESPONSE" >&2
  exit 1
fi

echo "Task ID: $TASK_ID"

for attempt in $(seq 1 90); do
  RESULT=$(curl -fsS "$API_BASE/v1/video/generations/$TASK_ID" \
    -H "Authorization: Bearer $SHANHUI_API_KEY")
  STATUS=$(printf '%s' "$RESULT" | python3 -c \
    'import json,sys; print(json.load(sys.stdin).get("status", ""))')
  echo "[$attempt] $STATUS"

  if [ "$STATUS" = 'completed' ] || [ "$STATUS" = 'failed' ]; then
    printf '%s' "$RESULT" | python3 -m json.tool
    exit 0
  fi
  sleep 10
done

echo '轮询超时；任务可能仍在上游执行，请保留 task_id 后续查询。' >&2
exit 2

import os
import time
import requests

BASE_URL = "https://ai.shanhui.tv"
API_KEY = os.environ["SHANHUI_API_KEY"]
HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

payload = {
    "model": "doubao-seedance-2-0-fast-260128",
    "prompt": "清晨的湖面，薄雾散开，镜头缓慢向前",
    "metadata": {
        "duration": 5,
        "ratio": "16:9",
        "resolution": "720p",
        "watermark": False,
    },
}

response = requests.post(
    f"{BASE_URL}/v1/video/generations",
    headers=HEADERS,
    json=payload,
    timeout=60,
)
response.raise_for_status()
submitted = response.json()
task_id = submitted.get("task_id") or submitted.get("id")
if not task_id:
    raise RuntimeError(f"响应中没有任务 ID: {submitted}")

deadline = time.monotonic() + 15 * 60
while time.monotonic() < deadline:
    response = requests.get(
        f"{BASE_URL}/v1/video/generations/{task_id}",
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=30,
    )
    response.raise_for_status()
    result = response.json()
    status = result.get("status")

    if status == "completed":
        print(result["metadata"]["url"])
        break
    if status == "failed":
        raise RuntimeError(result.get("error") or result)

    time.sleep(10)
else:
    raise TimeoutError(f"任务仍未结束，请稍后继续查询：{task_id}")

const baseURL = "https://ai.shanhui.tv";
const apiKey = process.env.SHANHUI_API_KEY;

const submitted = await fetch(`${baseURL}/v1/video/generations`, {
  method: "POST",
  headers: {
    Authorization: `Bearer ${apiKey}`,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    model: "doubao-seedance-2-0-fast-260128",
    prompt: "清晨的湖面，薄雾散开，镜头缓慢向前",
    metadata: {
      duration: 5,
      ratio: "16:9",
      resolution: "720p",
      watermark: false,
    },
  }),
}).then(async (r) => {
  if (!r.ok) throw new Error(await r.text());
  return r.json();
});

const taskId = submitted.task_id || submitted.id;
if (!taskId) throw new Error("响应中没有任务 ID");

for (let attempt = 0; attempt < 90; attempt += 1) {
  const result = await fetch(
    `${baseURL}/v1/video/generations/${encodeURIComponent(taskId)}`,
    { headers: { Authorization: `Bearer ${apiKey}` } },
  ).then(async (r) => {
    if (!r.ok) throw new Error(await r.text());
    return r.json();
  });

  if (result.status === "completed") {
    console.log(result.metadata?.url);
    break;
  }
  if (result.status === "failed") throw new Error(JSON.stringify(result.error));
  await new Promise((resolve) => setTimeout(resolve, 10_000));
}