omn​​i-fast-v2v实测：4SAPI视频转视频Python接入

你的应用 → 中转入口(4sapi) → omni-fast-v2v
  ① 提交任务 → 拿到 task_id
  ② 轮询 task_id → 查状态/进度
  ③ 状态 completed → 拿下载地址 → 下载成片

pip install pillow            # 仅测试2的参考图生成需要，其余用例可不装
export API_BASE=https://4sapi.com
export API_KEY=你的密钥        # 千万别把真实 Key 写进代码或提交到仓库
export OUTPUT_DIR=./output

import base64, json, os, time
from pathlib import Path
from urllib.request import Request, urlopen
from urllib.error import HTTPError

API_BASE = os.getenv("API_BASE", "https://4sapi.com").rstrip("/")
API_KEY = os.getenv("API_KEY", "")          # 从环境变量读取，不写死
OUTPUT_DIR = Path(os.getenv("OUTPUT_DIR", "./output"))
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

def api_request(method, path, body=None, timeout=60):
    url = f"{API_BASE}{path}"
    headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
    data = json.dumps(body, ensure_ascii=False).encode() if body else None
    req = Request(url, data=data, headers=headers, method=method)
    try:
        with urlopen(req, timeout=timeout) as resp:
            raw = resp.read().decode("utf-8", "replace")
            return json.loads(raw) if raw else {}
    except HTTPError as exc:
        print(f"HTTP {exc.code}: {exc.read().decode('utf-8', 'replace')[:500]}")
        raise

def submit_video(payload, timeout=120):
    return api_request("POST", "/v1/videos", payload, timeout=timeout)

def poll_video(task_id, interval=8, timeout=600):
    deadline = time.time() + timeout
    while time.time() < deadline:
        last = api_request("GET", f"/v1/videos/{task_id}")
        status = str(last.get("status", "")).lower()
        print(f"  状态: {status} | 进度: {last.get('progress', 0)}%")
        if status in {"completed", "succeeded", "success"}:
            return last
        if status in {"failed", "cancelled", "error"}:
            raise RuntimeError(f"任务失败: {last.get('error')}")
        time.sleep(interval)
    raise TimeoutError("任务超时")

MAX_VIDEO_SIZE = 5 * 1024 * 1024  # 5MB

def make_data_video(path: Path) -> str:
    size = path.stat().st_size
    if size > MAX_VIDEO_SIZE:
        raise ValueError(f"视频过大: {size/1024/1024:.1f}MB（限制 5MB）")
    return "data:video/mp4;base64," + base64.b64encode(path.read_bytes()).decode("ascii")

payload = {
    "model": "omni-fast-v2v",
    "prompt": "Transform this video into a rainy cinematic night scene with neon "
              "reflections and dramatic lighting. Keep the original composition.",
    "video": make_data_video(Path("source.mp4")),
    "seconds": "8",
    "aspect_ratio": "16:9",
    "resolution": "720p",
}
submitted = submit_video(payload)
task_id = submitted.get("id") or submitted.get("task_id")
result = poll_video(task_id, timeout=600)
# result 里取下载链接，存成 mp4

def download_file(url, save_path, timeout=300):
    if url.startswith("/"):
        url = f"{API_BASE}{url}"
    headers = {}
    if url.startswith(API_BASE):                 # 同域才带 Key
        headers["Authorization"] = f"Bearer {API_KEY}"
    with urlopen(Request(url, headers=headers), timeout=timeout) as resp:
        data = resp.read()
    save_path.write_bytes(data)
    return len(data)