Replace only the person in the source video with the person from the reference image. Preserve the reference person's facial identity, hairstyle, body shape, proportions, and clothing. Preserve the source video's original motion, pose sequence, facial motion, background, environment, objects, lighting, camera movement, framing, composition, timing, and duration. Do not modify or regenerate the scene. Keep the replacement person's identity, body shape, and clothing consistent across all frames.

curl -X POST "$BASE_URL/api/proxy/v1/videos" \
  -H "Cookie: ic_session=$IC_SESSION" \
  -H "X-IC-Operation-ID: $(uuidgen)" \
  -F "model=video-person-swap" \
  -F "prompt=Replace only the person in the source video with the person from the reference image. Preserve the reference person's facial identity, hairstyle, body shape, proportions, and clothing. Preserve the source video's original motion, pose sequence, facial motion, background, environment, objects, lighting, camera movement, framing, composition, timing, and duration. Do not modify or regenerate the scene. Keep the replacement person's identity, body shape, and clothing consistent across all frames." \
  -F "seed=12345" \
  -F "input_reference[]=@face.png" \
  -F "input_reference[]=@body.png" \
  -F "input_reference[]=@source.mp4"

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

import requests

BASE_URL = os.environ.get("BASE_URL", "https://ic.xshow.live").rstrip("/")
IC_SESSION = os.environ["IC_SESSION"]
MODEL = "video-person-swap"
POLL_SECONDS = 5
TIMEOUT_SECONDS = 6 * 60 * 60
DEFAULT_PROMPT = "Replace only the person in the source video with the person from the reference image. Preserve the reference person's facial identity, hairstyle, body shape, proportions, and clothing. Preserve the source video's original motion, pose sequence, facial motion, background, environment, objects, lighting, camera movement, framing, composition, timing, and duration. Do not modify or regenerate the scene. Keep the replacement person's identity, body shape, and clothing consistent across all frames."

session = requests.Session()
session.cookies.set("ic_session", IC_SESSION)


def mime(path: str) -> str:
    value = mimetypes.guess_type(path)[0]
    if not value:
        raise ValueError(f"无法识别媒体 MIME: {path}")
    return value


def checked(response: requests.Response) -> requests.Response:
    if not response.ok:
        raise RuntimeError(f"HTTP {response.status_code}: {response.text[:1000]}")
    return response


with ExitStack() as stack:
    files = [
        ("input_reference[]", (Path("face.png").name, stack.enter_context(open("face.png", "rb")), mime("face.png"))),
        ("input_reference[]", (Path("body.png").name, stack.enter_context(open("body.png", "rb")), mime("body.png"))),
        ("input_reference[]", (Path("source.mp4").name, stack.enter_context(open("source.mp4", "rb")), mime("source.mp4")))
    ]
    created = checked(session.post(
        f"{BASE_URL}/api/proxy/v1/videos",
        headers={"X-IC-Operation-ID": str(uuid.uuid4())},
        data={
            "model": MODEL,
            "prompt": DEFAULT_PROMPT,
            "seed": "12345",
        },
        files=files,
        timeout=1800,
    )).json()

task_id = created.get("id") or (created.get("data") or {}).get("id")
if not task_id:
    raise RuntimeError(f"创建响应缺少任务 ID: {created}")
print("task_id:", task_id)

deadline = time.time() + TIMEOUT_SECONDS
while time.time() < deadline:
    state = checked(session.post(
        f"{BASE_URL}/api/proxy/v1/videos/{task_id}",
        timeout=60,
    )).json()
    status = str(state.get("status") or (state.get("data") or {}).get("status") or "pending").lower()
    print("status:", status)
    if status == "completed":
        video = checked(session.get(
            f"{BASE_URL}/api/proxy/v1/videos/{task_id}/content",
            timeout=1800,
        ))
        Path("result.mp4").write_bytes(video.content)
        print("saved: result.mp4")

        payload = state.get("data") if isinstance(state.get("data"), dict) else state
        if payload.get("intermediate_image_available") is True:
            intermediate = checked(session.get(
                f"{BASE_URL}/api/proxy/v1/videos/{task_id}/intermediate",
                timeout=600,
            ))
            Path("intermediate.png").write_bytes(intermediate.content)
            print("saved: intermediate.png")
        break
    if status in {"failed", "cancelled", "canceled", "expired"}:
        raise RuntimeError(f"任务失败: {state}")
    time.sleep(POLL_SECONDS)
else:
    raise TimeoutError(f"任务 {task_id} 超过 {TIMEOUT_SECONDS} 秒仍未完成")