Authorization: Bearer <PARTNER_API_KEY>
Idempotency-Key: <业务订单或请求的唯一标识>

export BASE_URL="https://ic.xshow.live/api/partner/v1"

curl --fail-with-body "$BASE_URL/videos" \
  -H "Authorization: Bearer $PARTNER_API_KEY" \
  -H "Idempotency-Key: $ORDER_ID" \
  -F "model=video-face-swap" \
  -F "input_reference[]=@identity.png;type=image/png" \
  -F "input_reference[]=@source.mp4;type=video/mp4" \
  -F "seed=77"

curl --fail-with-body "$BASE_URL/videos" \
  -H "Authorization: Bearer $PARTNER_API_KEY" \
  -H "Idempotency-Key: $ORDER_ID" \
  -F "model=video-face-swap" \
  -F "template_id=missionary-15" \
  -F "input_reference[]=@identity.png;type=image/png"

{"id":"example-task-id","status":"pending","intermediate_image_available":false,"compute":{"kind":"scheduler","phase":"queued"}}

{"id":"example-task-id","status":"completed","intermediate_image_available":true}

{"id":"example-task-id","status":"failed","error":{"message":"任务执行失败的具体原因"}}

import mimetypes
import os
import time
from contextlib import ExitStack
from pathlib import Path

import requests


def media_type(filename):
    value = mimetypes.guess_type(filename)[0]
    if value not in {"image/jpeg", "image/png", "image/bmp", "image/webp", "image/gif", "video/mp4", "video/quicktime"}:
        raise ValueError("无法识别或不支持的媒体类型")
    return value


def checked(response):
    response.raise_for_status()
    return response


def main():
    base = os.environ.get("BASE_URL", "https://ic.xshow.live/api/partner/v1").rstrip("/")
    model = os.environ.get("MODEL", "video-face-swap")
    models = {"video-face-swap", "video-person-swap", "video-undress", "video-head-swap-spicy", "video-pose"}
    if model not in models:
        raise ValueError("不支持的公开模型")
    task_id = os.environ.get("TASK_ID", "")
    with requests.Session() as session:
        session.headers["Authorization"] = "Bearer " + os.environ["PARTNER_API_KEY"]
        if not task_id:
            order_id = os.environ["IDEMPOTENCY_KEY"]
            data = {"model": model, "prompt": os.environ.get("PROMPT", "")}
            if os.environ.get("SEED"):
                data["seed"] = os.environ["SEED"]
            paths = [] if model == "video-undress" else [os.environ.get("FACE_IMAGE", "identity.png")]
            if model == "video-person-swap" and os.environ.get("BODY_IMAGE"):
                paths.append(os.environ["BODY_IMAGE"])
            if os.environ.get("TEMPLATE_ID"):
                raise ValueError("当前 Partner 版本不支持模板模式")
            if model != "video-pose":
                source = os.environ.get("SOURCE_VIDEO", "source.mp4")
                source_mime = media_type(source)
                if source_mime == "image/gif" and model in {"video-head-swap-spicy", "video-face-swap", "video-undress"}:
                    raise ValueError("此视频模式不接收 GIF")
                if source_mime not in {"video/mp4", "video/quicktime", "image/gif"}:
                    raise ValueError("源媒体必须是 MP4/MOV 或该模式支持的 GIF")
                paths.append(source)
            if model == "video-pose":
                if not data["prompt"].strip():
                    raise ValueError("姿势模式必须设置 PROMPT")
                data.update(lora=os.environ["POSE_LORA"], seconds="5", resolution_name="720p", ratio="9:16")
                if os.environ.get("VIDEO_PROMPT"):
                    data["video_prompt"] = os.environ["VIDEO_PROMPT"]
            with ExitStack() as stack:
                files = [("input_reference[]", (Path(p).name, stack.enter_context(open(p, "rb")), media_type(p))) for p in paths]
                result = checked(session.post(base + "/videos", headers={"Idempotency-Key": order_id}, data=data, files=files, timeout=1800)).json()
            task_id = result["id"]
            print("TASK_ID=" + task_id, flush=True)
        deadline = time.monotonic() + 6 * 60 * 60
        intermediate_saved = False
        while time.monotonic() < deadline:
            state = checked(session.post(f"{base}/videos/{task_id}", timeout=60)).json()
            print("status=", state.get("status"), "compute=", state.get("compute"), flush=True)
            if state.get("intermediate_image_available") and not intermediate_saved:
                image = checked(session.get(f"{base}/videos/{task_id}/intermediate", timeout=600))
                Path("intermediate.png").write_bytes(image.content)
                intermediate_saved = True
            if state.get("status") == "completed":
                params = {"format": os.environ["OUTPUT_FORMAT"]} if os.environ.get("OUTPUT_FORMAT") else None
                with checked(session.get(f"{base}/videos/{task_id}/content", params=params, timeout=1800, stream=True)) as video:
                    suffix = ".gif" if video.headers.get("Content-Type", "").split(";")[0] == "image/gif" else ".mp4"
                    with open("result" + suffix, "wb") as output:
                        for chunk in video.iter_content(1024 * 1024):
                            output.write(chunk)
                print("saved: result" + suffix)
                return
            if state.get("status") in {"failed", "cancelled", "canceled", "expired"}:
                raise RuntimeError(str(state.get("error") or state.get("status")))
            time.sleep(5)
        raise TimeoutError("停止客户端轮询；服务端任务不因此取消，请用 TASK_ID 继续查询")


if __name__ == "__main__":
    main()