Asset-Referenced Video Generation in Code

import time
from pathlib import Path

import requests

ASSET_KEY = "sk-your-asset-library-KEY"  # icover.ai Settings → Asset Library KEY
APIYI_KEY = "sk-your-APIYI-token"        # api.apiyi.com, SeeDance2 group enabled
IMAGE_PATH = "portrait.jpg"              # local reference image (virtual avatar)
PROMPT = "The character in Image 1 smiles at the camera, slow push-in, natural light"

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",  # required: works around a gateway gzip header mismatch
}


def upload_image(path: str) -> str:
    """1) presign + PUT the local image; returns a public URL (skip if you have one)"""
    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,  # must match the presign
    )
    resp.raise_for_status()
    return data["publicUrl"]


def ingest_asset(image_url: str, label: str = "", group_id: str = None) -> str:
    """2) ingest the asset; returns the asset Id. Real-person assets: pass the verified 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:
    """3) poll the ingest status until Active (about 13 seconds per image, no 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("Ingest failed: check format / aspect ratio 0.4-2.5 / side 300-6000px, then re-upload")
        time.sleep(3)
    raise TimeoutError("Not Active after 90 seconds — investigate and retry")


def create_video_task(asset_id: str) -> str:
    """4) create the generation task referencing asset://. Say "Image 1" in the prompt, never the raw 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:
    """5) poll the task until a terminal state (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:
    """6) download the video. The signed link expires in 24 hours; no Authorization header here"""
    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("saved", out)


if __name__ == "__main__":
    public_url = upload_image(IMAGE_PATH)
    asset_id = ingest_asset(public_url, label="character-A-front")
    print("asset 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("billed tokens:", task["usage"]["completion_tokens"])
        download(task, f"{task_id}.mp4")
    else:
        print("task did not succeed:", task.get("error"))