素材引用生视频实战

import time
from pathlib import Path

import requests

ASSET_KEY = "sk-你的素材库KEY"    # icover.ai「设置 → 素材库 KEY」
APIYI_KEY = "sk-你的APIYI令牌"    # api.apiyi.com，须勾选 SeeDance2 分组
IMAGE_PATH = "portrait.jpg"       # 本地参考图（虚拟人像）
PROMPT = "图片1中的人物正面微笑，镜头缓慢推近，自然光"

ICOVER = "https://icover.ai/api"
SEEDANCE = "https://api.apiyi.com/seedance/api/v3/contents/generations/tasks"
ASSET_HEADERS = {"Authorization": f"Bearer {ASSET_KEY}"}
APIYI_HEADERS = {
    "Authorization": f"Bearer {APIYI_KEY}",
    "Content-Type": "application/json",
    "Accept-Encoding": "identity",  # 必加：规避网关 gzip 头与实际编码不符
}


def upload_image(path: str) -> str:
    """① presign + PUT 上传本地图片，返回公网 URL（已有公网 URL 可跳过）"""
    ext = Path(path).suffix.lstrip(".").lower() or "jpg"
    content_type = f"image/{'jpeg' if ext in ('jpg', 'jpeg') else ext}"
    data = requests.post(
        f"{ICOVER}/storage/presign", headers=ASSET_HEADERS,
        json={"ext": ext, "contentType": content_type}, timeout=30,
    ).json()["data"]
    resp = requests.put(
        data["uploadUrl"], data=Path(path).read_bytes(),
        headers={"Content-Type": content_type}, timeout=120,  # 与 presign 一致
    )
    resp.raise_for_status()
    return data["publicUrl"]


def ingest_asset(image_url: str, label: str = "", group_id: str = None) -> str:
    """② 素材入库，返回 asset Id。真人素材：带上真人认证拿到的 group_id"""
    body = {"imageUrl": image_url, "label": label}
    if group_id:
        body["groupId"] = group_id
    r = requests.post(
        f"{ICOVER}/asset-library/assets", headers=ASSET_HEADERS,
        json=body, timeout=60,
    ).json()
    return r["Result"]["Id"]


def wait_asset_active(asset_id: str, timeout: int = 90) -> None:
    """③ 轮询入库状态到 Active（单图约 13 秒，无 SLA）"""
    deadline = time.time() + timeout
    while time.time() < deadline:
        status = requests.get(
            f"{ICOVER}/asset-library/assets/{asset_id}",
            headers=ASSET_HEADERS, timeout=30,
        ).json()["Result"]["Status"]
        print("asset status:", status)
        if status == "Active":
            return
        if status == "Failed":
            raise RuntimeError("素材入库失败：检查格式 / 宽高比 0.4-2.5 / 边长 300-6000px 后重传")
        time.sleep(3)
    raise TimeoutError("90 秒未 Active，请排查后重试")


def create_video_task(asset_id: str) -> str:
    """④ 引用 asset:// 创建生成任务。提示词用「图片1」指代，不要写 asset ID 原文"""
    body = {
        "model": "doubao-seedance-2-0-260128",
        "content": [
            {"type": "text", "text": PROMPT},
            {"type": "image_url",
             "image_url": {"url": f"asset://{asset_id}"},
             "role": "reference_image"},
        ],
        "resolution": "720p", "ratio": "16:9", "duration": 5,
    }
    return requests.post(SEEDANCE, headers=APIYI_HEADERS, json=body, timeout=60).json()["id"]


def wait_video(task_id: str) -> dict:
    """⑤ 轮询任务到终态（succeeded / failed / expired）"""
    while True:
        time.sleep(20)
        task = requests.get(f"{SEEDANCE}/{task_id}", headers=APIYI_HEADERS, timeout=30).json()
        print("task status:", task.get("status"))
        if task.get("status") in ("succeeded", "failed", "expired"):
            return task


def download(task: dict, out: str) -> None:
    """⑥ 下载视频。直链 24 小时过期，立即转存；下载不要带 Authorization 头"""
    with requests.get(task["content"]["video_url"], stream=True, timeout=300) as r:
        r.raise_for_status()
        with open(out, "wb") as f:
            for chunk in r.iter_content(chunk_size=1 << 20):
                f.write(chunk)
    print("已保存", out)


if __name__ == "__main__":
    public_url = upload_image(IMAGE_PATH)
    asset_id = ingest_asset(public_url, label="角色A-正面")
    print("素材 ID:", f"asset://{asset_id}")
    wait_asset_active(asset_id)

    task_id = create_video_task(asset_id)
    print("task_id:", task_id)
    task = wait_video(task_id)
    if task["status"] == "succeeded":
        print("计费 tokens:", task["usage"]["completion_tokens"])
        download(task, f"{task_id}.mp4")
    else:
        print("任务未成功:", task.get("error"))