Seedance 视频生成示例

import hashlib
import hmac
import json
import os
import time
from datetime import datetime, timezone
from urllib.error import HTTPError
from urllib.request import Request, urlopen

API_KEY = os.environ["APIGO_API_KEY"]
HOST = "api-vip.apigo.ai"
MODEL = "dreamina-seedance-2-0-260128"
IMAGE_URL = "https://ark-project.tos-cn-beijing.volces.com/doc_image/r2v_tea_pic1.jpg"


def send(request):
    try:
        with urlopen(request, timeout=60) as resp:
            return json.loads(resp.read())
    except HTTPError as e:
        raise SystemExit(f"HTTP {e.code}: {e.read().decode('utf-8', 'replace')}")


def asset_api(action, body):
    """素材库管理接口：HMAC-SHA256 签名。"""
    payload = json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
    x_date = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    day = x_date[:8]
    query = f"Action={action}&Version=2024-01-01"
    body_hash = hashlib.sha256(payload).hexdigest()
    signed_headers = "content-type;host;x-content-sha256;x-date"
    canonical_request = "\n".join([
        "POST", "/", query,
        f"content-type:application/json\nhost:{HOST}\n"
        f"x-content-sha256:{body_hash}\nx-date:{x_date}\n",
        signed_headers, body_hash,
    ])
    scope = f"{day}/ap-southeast-1/ark/request"
    string_to_sign = "\n".join([
        "HMAC-SHA256", x_date, scope,
        hashlib.sha256(canonical_request.encode("utf-8")).hexdigest(),
    ])
    signing_key = API_KEY.encode("utf-8")
    for part in (day, "ap-southeast-1", "ark", "request"):
        signing_key = hmac.new(signing_key, part.encode("utf-8"), hashlib.sha256).digest()
    signature = hmac.new(signing_key, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()

    return send(Request(f"https://{HOST}/?{query}", data=payload, method="POST", headers={
        "Content-Type": "application/json",
        "X-Date": x_date,
        "X-Content-Sha256": body_hash,
        "Authorization": f"HMAC-SHA256 Credential={API_KEY}/{scope}, "
                         f"SignedHeaders={signed_headers}, Signature={signature}",
    }))["Result"]


def task_api(method, path, body=None):
    """视频生成接口：Bearer 鉴权。"""
    data = json.dumps(body).encode("utf-8") if body is not None else None
    return send(Request(f"https://{HOST}{path}", data=data, method=method, headers={
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }))


# 1. 创建素材组
group_id = asset_api("CreateAssetGroup", {"Name": "demo-assets", "GroupType": "AIGC"})["Id"]
print("group:", group_id)

# 2. 导入图片素材
asset_id = asset_api("CreateAsset", {
    "GroupId": group_id,
    "URL": IMAGE_URL,
    "AssetType": "Image",
    "Name": "tea-reference",
})["Id"]
print("asset:", asset_id)

# 3. 等待素材就绪
while True:
    asset = asset_api("GetAsset", {"Id": asset_id})
    print("asset status:", asset["Status"])
    if asset["Status"] == "Active":
        break
    if asset["Status"] == "Failed":
        raise SystemExit(f"素材处理失败: {asset.get('Error')}")
    time.sleep(5)

# 4. 创建视频生成任务
task_id = task_api("POST", "/api/v3/contents/generations/tasks", {
    "model": MODEL,
    "content": [
        {"type": "text", "text": "让图片1中的茶杯缓缓升起热气，镜头轻微推近。"},
        {"type": "image_url", "role": "reference_image", "image_url": {"url": f"asset://{asset_id}"}},
    ],
    "duration": 5,
    "resolution": "720p",
    "ratio": "16:9",
    "generate_audio": False,
    "watermark": False,
})["id"]
print("task:", task_id)

# 5. 轮询任务结果
while True:
    task = task_api("GET", f"/api/v3/contents/generations/tasks/{task_id}")
    print("task status:", task["status"])
    if task["status"] not in ("queued", "running"):
        break
    time.sleep(10)

if task["status"] != "succeeded":
    raise SystemExit(f"任务结束: {task['status']} {task.get('error')}")

with urlopen(task["content"]["video_url"], timeout=300) as resp, open("output.mp4", "wb") as f:
    f.write(resp.read())
print("saved: output.mp4")