#!/usr/bin/env python3 """ Wan 2.6 完整 Demo 支持功能：文生视频(t2v)、图生视频(i2v/i2v-flash)、有声/无声/自定义音频 视频相关参数全覆盖 用法： NEWAPI_BASE_URL=https://your-gateway.example.com/v1 \ NEWAPI_API_KEY=sk-xxx \ python3 demo_wan26.py 可选参数： --skip-t2v 跳过文生视频 --skip-i2v 跳过图生视频 --only-t2v 只测文生视频 --only-i2v 只测图生视频 """ import json import os import sys import time import urllib.error import urllib.request from pathlib import Path BASE_URL = os.environ.get("NEWAPI_BASE_URL", "") API_KEY = os.environ.get("NEWAPI_API_KEY", "") OUTPUT_DIR = Path(os.environ.get("NEWAPI_OUTPUT_DIR", "wan26_output")) def api_post(path, data): req = urllib.request.Request( f"{BASE_URL}{path}", data=json.dumps(data).encode("utf-8"), headers={ "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json", }, method="POST", ) with urllib.request.urlopen(req, timeout=120) as resp: return json.loads(resp.read().decode("utf-8")) def api_get(path): req = urllib.request.Request( f"{BASE_URL}{path}", headers={"Authorization": f"Bearer {API_KEY}"}, method="GET", ) with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read().decode("utf-8")) def download(url, dest): with urllib.request.urlopen(url, timeout=60) as r: data = r.read() Path(dest).write_bytes(data) return len(data) def poll_video(task_id, interval=10, timeout=300): """轮询视频任务直到完成""" deadline = time.time() + timeout while time.time() < deadline: data = api_get(f"/videos/{task_id}") status = data.get("status", "unknown") print(f" 轮询: {status}") if status == "completed": url = data.get("metadata", {}).get("url", "") return url if status == "failed": err = data.get("error", {}).get("message", "unknown error") raise RuntimeError(f"任务失败: {err}") time.sleep(interval) raise TimeoutError(f"轮询超时 ({timeout}s)") # ============================================================ # 文生视频 (wan2.6-t2v) # ============================================================ def test_t2v_basic(): """5. 文生视频：基础""" print("\n=== 5. 文生视频：基础 (720P, 5s) ===") resp = api_post("/video/generations", { "model": "wan2.6-t2v", "prompt": "一只橘猫在海边奔跑，慢动作，电影画质", "resolution": "720P", "duration": 5, }) task_id = resp["id"] print(f" 任务: {task_id}") url = poll_video(task_id) size = download(url, OUTPUT_DIR / "5_t2v_basic.mp4") print(f" ✅ {size} bytes → 5_t2v_basic.mp4") def test_t2v_params(): """6. 文生视频：全参数""" print("\n=== 6. 文生视频：全参数 (1080P + negative_prompt + seed) ===") resp = api_post("/video/generations", { "model": "wan2.6-t2v", "prompt": "城市夜景延时摄影，车流光轨", "negative_prompt": "模糊, 抖动, 低质量", "resolution": "1080P", "duration": 5, "prompt_extend": True, "watermark": False, "seed": 12345, }) task_id = resp["id"] print(f" 任务: {task_id}") url = poll_video(task_id, timeout=600) size = download(url, OUTPUT_DIR / "6_t2v_params.mp4") print(f" ✅ {size} bytes → 6_t2v_params.mp4") # ============================================================ # 图生视频 (wan2.6-i2v / wan2.6-i2v-flash) # ============================================================ def test_i2v_silent(img_url): """7. 图生视频：无声 (i2v-flash)""" print("\n=== 7. 图生视频：无声 (i2v-flash, 720P, 5s) ===") resp = api_post("/video/generations", { "model": "wan2.6-i2v-flash", "prompt": "猫开始奔跑", "input_reference": img_url, "resolution": "720P", "duration": 5, # audio defaults to false }) task_id = resp["id"] print(f" 任务: {task_id}") url = poll_video(task_id) size = download(url, OUTPUT_DIR / "7_i2v_silent.mp4") print(f" ✅ {size} bytes → 7_i2v_silent.mp4") def test_i2v_audio(img_url): """8. 图生视频：有声 (i2v-flash)""" print("\n=== 8. 图生视频：有声 (i2v-flash, 720P, 5s, audio=true) ===") resp = api_post("/video/generations", { "model": "wan2.6-i2v-flash", "prompt": "街头吉他手弹奏，城市环境音", "input_reference": img_url, "resolution": "720P", "duration": 5, "audio": True, }) task_id = resp["id"] print(f" 任务: {task_id}") url = poll_video(task_id) size = download(url, OUTPUT_DIR / "8_i2v_audio.mp4") print(f" ✅ {size} bytes → 8_i2v_audio.mp4") def test_i2v_custom_audio(img_url, audio_url): """9. 图生视频：自定义音频""" print("\n=== 9. 图生视频：自定义音频 (audio_url) ===") resp = api_post("/video/generations", { "model": "wan2.6-i2v-flash", "prompt": "音乐家演奏", "input_reference": img_url, "resolution": "720P", "duration": 5, "audio": True, "audio_url": audio_url, }) task_id = resp["id"] print(f" 任务: {task_id}") url = poll_video(task_id) size = download(url, OUTPUT_DIR / "9_i2v_custom_audio.mp4") print(f" ✅ {size} bytes → 9_i2v_custom_audio.mp4") def test_i2v_multishot(img_url): """10. 图生视频：多镜头 (shot_type=multi)""" print("\n=== 10. 图生视频：多镜头 (i2v-flash, shot_type=multi) ===") resp = api_post("/video/generations", { "model": "wan2.6-i2v-flash", "prompt": "猫在花园探索，多角度镜头切换", "input_reference": img_url, "resolution": "720P", "duration": 5, "shot_type": "multi", "prompt_extend": True, }) task_id = resp["id"] print(f" 任务: {task_id}") url = poll_video(task_id, timeout=600) size = download(url, OUTPUT_DIR / "10_i2v_multishot.mp4") print(f" ✅ {size} bytes → 10_i2v_multishot.mp4") def test_i2v_hq(img_url): """11. 图生视频：高质量 (wan2.6-i2v)""" print("\n=== 11. 图生视频：高质量 (wan2.6-i2v, 1080P) ===") resp = api_post("/video/generations", { "model": "wan2.6-i2v", "prompt": "猫优雅地走过花园小径", "negative_prompt": "模糊, 抖动", "input_reference": img_url, "resolution": "1080P", "duration": 5, "watermark": False, "seed": 42, }) task_id = resp["id"] print(f" 任务: {task_id}") url = poll_video(task_id, timeout=600) size = download(url, OUTPUT_DIR / "11_i2v_hq.mp4") print(f" ✅ {size} bytes → 11_i2v_hq.mp4") # 注意: 首尾帧生视频(first_frame_url + last_frame_url)仅 wan2.7 支持 # wan2.6 只支持单图(img_url)生视频 # ============================================================ # 主流程 # ============================================================ def main(): if not API_KEY or not BASE_URL: print("请先设置环境变量 NEWAPI_BASE_URL 和 NEWAPI_API_KEY", file=sys.stderr) print( "示例: NEWAPI_BASE_URL='https://your-gateway.example.com/v1' " "NEWAPI_API_KEY='sk-xxx' python3 demo_wan26.py", file=sys.stderr, ) sys.exit(1) args = sys.argv[1:] skip_t2v = "--skip-t2v" in args skip_i2v = "--skip-i2v" in args only_t2v = "--only-t2v" in args only_i2v = "--only-i2v" in args if only_t2v: skip_i2v = True elif only_i2v: skip_t2v = True OUTPUT_DIR.mkdir(exist_ok=True) print(f"Base URL: {BASE_URL}") print(f"输出目录: {OUTPUT_DIR}") results = {} img_url = "" # 生成源图用于 i2v 测试 if not skip_i2v: print("\n=== 生成源图用于 i2v 测试 ===") try: resp = api_post("/images/generations", { "model": "gpt-image-2", "prompt": "橘猫，摄影风格", "size": "1024x1024", "n": 1, }) img_url = resp["data"][0]["url"] print(f" ✅ {img_url[:60]}...") except Exception as e: print(f" ❌ 无法生成源图: {e}") # --- 文生视频 --- if not skip_t2v: for name, fn in [ ("5_t2v_basic", test_t2v_basic), ("6_t2v_params", test_t2v_params), ]: try: fn() results[name] = "✅" except Exception as e: results[name] = f"❌ {e}" # --- 图生视频 --- if not skip_i2v and img_url: for name, fn, extra in [ ("7_i2v_silent", test_i2v_silent, (img_url,)), ("8_i2v_audio", test_i2v_audio, (img_url,)), ("10_i2v_multishot", test_i2v_multishot, (img_url,)), ("11_i2v_hq", test_i2v_hq, (img_url,)), ]: try: fn(*extra) results[name] = "✅" except Exception as e: results[name] = f"❌ {e}" # 自定义音频（需要用户提供 audio_url，跳过） audio_url = os.environ.get("NEWAPI_AUDIO_URL", "") if audio_url: try: test_i2v_custom_audio(img_url, audio_url) results["9_i2v_custom_audio"] = "✅" except Exception as e: results["9_i2v_custom_audio"] = f"❌ {e}" else: results["9_i2v_custom_audio"] = "⏭️ 跳过（设 NEWAPI_AUDIO_URL 启用）" elif not skip_i2v: results["i2v"] = "⏭️ 跳过（无源图）" # 汇总 print("\n" + "=" * 50) print(" 测试结果") print("=" * 50) passed = sum(1 for v in results.values() if v == "✅") for k, v in sorted(results.items()): print(f" {k}: {v}") print(f"\n 通过: {passed}/{len(results)}") if __name__ == "__main__": main()