素材引用生视频实战

import time
import requests

BASE = "https://api.looprouter.com"
HEADERS = {"Authorization": "Bearer sk-你的APIKey", "Accept-Encoding": "identity"}


def asset_action(name, body):
    resp = requests.post(
        f"{BASE}/api/v3/assets?Action={name}&Version=2024-01-01",
        headers=HEADERS, json=body,
    )
    resp.raise_for_status()
    return resp.json()["Result"]


def create_asset(group_id, name, url):
    result = asset_action("CreateAsset", {"GroupId": group_id, "Name": name, "URL": url})
    asset_id = result["Id"]
    deadline = time.time() + 90
    while time.time() < deadline:
        result = asset_action("GetAsset", {"Id": asset_id})
        if result["Status"] == "Active":
            return asset_id
        if result["Status"] == "Failed":
            raise RuntimeError(f"素材处理失败: {result.get('Error', {}).get('Message')}")
        time.sleep(3)
    raise TimeoutError("素材处理超时，请稍后用 GetAsset 复查")


def generate_video(content):
    resp = requests.post(
        f"{BASE}/api/v3/contents/generations/tasks",
        headers=HEADERS,
        json={
            "model": "doubao-seedance-2-0-260128",
            "content": content,
            "resolution": "720p",
            "ratio": "adaptive",
            "duration": 5,
        },
    )
    resp.raise_for_status()
    task_id = resp.json()["id"]
    while True:
        task = requests.get(
            f"{BASE}/api/v3/contents/generations/tasks/{task_id}", headers=HEADERS
        ).json()
        if task["status"] in ("succeeded", "failed", "cancelled", "expired"):
            return task
        time.sleep(20)


if __name__ == "__main__":
    group = asset_action("CreateAssetGroup", {"Name": "短剧角色A", "GroupType": "AIGC"})
    asset_id = create_asset(group["Id"], "角色A-全身", "https://example.com/character-a.png")

    task = generate_video([
        {"type": "text", "text": "图片1中的角色撑着伞走在雨中的巷子里，镜头跟随"},
        {"type": "image_url", "image_url": {"url": f"asset://{asset_id}"}, "role": "reference_image"},
    ])
    if task["status"] != "succeeded":
        raise SystemExit(f"生成失败: {task.get('error')}")

    video_url = task["content"]["video_url"]
    with requests.get(video_url, stream=True) as r:  # 不带 Authorization 头
        with open("output.mp4", "wb") as f:
            for chunk in r.iter_content(chunk_size=1 << 20):
                f.write(chunk)
    print("已保存 output.mp4")

KEY="sk-你的APIKey"
BASE="https://api.looprouter.com"

# 1. 创建素材组
curl -X POST "$BASE/api/v3/assets?Action=CreateAssetGroup&Version=2024-01-01" \
  -H "Authorization: Bearer $KEY" -H "Content-Type: application/json" \
  -d '{"Name": "短剧角色A", "GroupType": "AIGC"}'
# => Result.Id = group-20260709185838-dcnvd

# 2. 注册素材
curl -X POST "$BASE/api/v3/assets?Action=CreateAsset&Version=2024-01-01" \
  -H "Authorization: Bearer $KEY" -H "Content-Type: application/json" \
  -d '{"GroupId": "group-20260709185838-dcnvd", "Name": "角色A-全身", "URL": "https://example.com/character-a.png"}'
# => Result.Id = asset-20260709190143-ntwhb

# 3. 轮询素材状态到 Active
curl -X POST "$BASE/api/v3/assets?Action=GetAsset&Version=2024-01-01" \
  -H "Authorization: Bearer $KEY" -H "Content-Type: application/json" \
  -d '{"Id": "asset-20260709190143-ntwhb"}'

# 4. 引用素材生成视频
curl -X POST "$BASE/api/v3/contents/generations/tasks" \
  -H "Authorization: Bearer $KEY" -H "Content-Type: application/json" \
  -d '{
    "model": "doubao-seedance-2-0-260128",
    "content": [
      {"type": "text", "text": "图片1中的角色撑着伞走在雨中的巷子里，镜头跟随"},
      {"type": "image_url", "image_url": {"url": "asset://asset-20260709190143-ntwhb"}, "role": "reference_image"}
    ],
    "resolution": "720p",
    "duration": 5
  }'
# => {"id": "task_xxxx"}

# 5. 轮询任务
curl "$BASE/api/v3/contents/generations/tasks/task_xxxx" -H "Authorization: Bearer $KEY"

# 6. 下载成片（不带 Authorization 头）
curl -o output.mp4 "<content.video_url>"

const BASE = "https://api.looprouter.com";
const KEY = process.env.LOOPROUTER_API_KEY;
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

async function assetAction(name, body) {
  const res = await fetch(`${BASE}/api/v3/assets?Action=${name}&Version=2024-01-01`, {
    method: "POST",
    headers: { "Content-Type": "application/json", Authorization: `Bearer ${KEY}` },
    body: JSON.stringify(body),
  });
  if (!res.ok) throw new Error(`${name} failed: ${await res.text()}`);
  return (await res.json()).Result;
}

async function main() {
  const group = await assetAction("CreateAssetGroup", { Name: "短剧角色A", GroupType: "AIGC" });
  const { Id: assetId } = await assetAction("CreateAsset", {
    GroupId: group.Id,
    Name: "角色A-全身",
    URL: "https://example.com/character-a.png",
  });

  for (;;) {
    const asset = await assetAction("GetAsset", { Id: assetId });
    if (asset.Status === "Active") break;
    if (asset.Status === "Failed") throw new Error(`素材处理失败: ${asset.Error?.Message}`);
    await sleep(3000);
  }

  const createRes = await fetch(`${BASE}/api/v3/contents/generations/tasks`, {
    method: "POST",
    headers: { "Content-Type": "application/json", Authorization: `Bearer ${KEY}` },
    body: JSON.stringify({
      model: "doubao-seedance-2-0-260128",
      content: [
        { type: "text", text: "图片1中的角色撑着伞走在雨中的巷子里，镜头跟随" },
        { type: "image_url", image_url: { url: `asset://${assetId}` }, role: "reference_image" },
      ],
      resolution: "720p",
      duration: 5,
    }),
  });
  const { id } = await createRes.json();

  for (;;) {
    await sleep(20000);
    const task = await (
      await fetch(`${BASE}/api/v3/contents/generations/tasks/${id}`, {
        headers: { Authorization: `Bearer ${KEY}` },
      })
    ).json();
    if (task.status === "succeeded") return console.log("视频地址:", task.content.video_url);
    if (["failed", "cancelled", "expired"].includes(task.status))
      throw new Error(JSON.stringify(task.error));
  }
}

main();