# app.py import os import oss2 import sys import uuid import shutil import time import gradio as gr import requests import dashscope from dashscope.utils.oss_utils import check_and_upload_local DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY") dashscope.api_key = DASHSCOPE_API_KEY class WanAnimateApp: def __init__(self, url, get_url): self.url = url self.get_url = get_url def predict( self, ref_img, video, model_id, model, ): # Upload files to OSS if needed and get URLs _, image_url = check_and_upload_local(model_id, ref_img, DASHSCOPE_API_KEY) _, video_url = check_and_upload_local(model_id, video, DASHSCOPE_API_KEY) # Prepare the request payload payload = { "model": model_id, "input": { "image_url": image_url, "video_url": video_url }, "parameters": { "check_image": True, "mode": model, } } # Set up headers headers = { "X-DashScope-Async": "enable", "X-DashScope-OssResourceResolve": "enable", "Authorization": f"Bearer {DASHSCOPE_API_KEY}", "Content-Type": "application/json" } # Make the initial API request url = self.url response = requests.post(url, json=payload, headers=headers) # Check if request was successful if response.status_code != 200: raise Exception(f"Initial request failed with status code {response.status_code}: {response.text}") # Get the task ID from response result = response.json() task_id = result.get("output", {}).get("task_id") if not task_id: raise Exception("Failed to get task ID from response") # Poll for results get_url = f"{self.get_url}/{task_id}" headers = { "Authorization": f"Bearer {DASHSCOPE_API_KEY}", "Content-Type": "application/json" } while True: response = requests.get(get_url, headers=headers) if response.status_code != 200: raise Exception(f"Failed to get task status: {response.status_code}: {response.text}") result = response.json() print(result) task_status = result.get("output", {}).get("task_status") if task_status == "SUCCEEDED": # Task completed successfully, return video URL video_url = result["output"]["results"]["video_url"] return video_url, "SUCCEEDED" elif task_status == "FAILED": # Task failed, raise an exception with error message error_msg = result.get("output", {}).get("message", "Unknown error") code_msg = result.get("output", {}).get("code", "Unknown code") print(f"\n\nTask failed: {error_msg} Code: {code_msg} TaskId: {task_id}\n\n") return None, f"Task failed: {error_msg} Code: {code_msg} TaskId: {task_id}" # raise Exception(f"Task failed: {error_msg} TaskId: {task_id}") else: # Task is still running, wait and retry time.sleep(5) # Wait 5 seconds before polling again def start_app(): import argparse parser = argparse.ArgumentParser(description="Wan2.2-Animate 视频生成工具") args = parser.parse_args() url = "https://dashscope.aliyuncs.com/api/v1/services/aigc/image2video/video-synthesis/" # url = "https://poc-dashscope.aliyuncs.com/api/v1/services/aigc/image2video/video-synthesis" get_url = f"https://dashscope.aliyuncs.com/api/v1/tasks/" # get_url = f"https://poc-dashscope.aliyuncs.com/api/v1/tasks" app = WanAnimateApp(url=url, get_url=get_url) with gr.Blocks(title="Wan2.2-Animate 视频生成") as demo: gr.HTML("""

Wan2.2-Animate: Unified Character Animation and Replacement with Holistic Replication

Wan2.2-Animate: 统一的角色动画和视频人物替换模型

Tongyi Lab, Alibaba
📄 Paper 💻 GitHub 🤗 HF Model 🤖 MS Model
🤗 HF Space 🤖 MS Studio
""") gr.HTML("""
‼️Usage (使用说明) Wan-Animate supports two mode: Wan-Animate 支持两种模式: Currently, the following restrictions apply to inputs: 当前，对于输入有以下的限制

Currently, the inference quality has two variants. You can use our open-source code for more flexible configuration.

当前，推理质量有两个变种。 您可以使用我们的开源代码，来进行更灵活的设置。

""") with gr.Row(): with gr.Column(): ref_img = gr.Image( label="Reference Image(参考图像)", type="filepath", sources=["upload"], ) video = gr.Video( label="Template Video(模版视频)", sources=["upload"], ) with gr.Row(): model_id = gr.Dropdown( label="Mode(模式)", choices=["wan2.2-animate-move", "wan2.2-animate-mix"], value="wan2.2-animate-move", info="" ) model = gr.Dropdown( label="推理质量(Inference Quality)", choices=["wan-pro", "wan-std"], value="wan-pro", ) run_button = gr.Button("Generate Video(生成视频)") with gr.Column(): output_video = gr.Video(label="Output Video(输出视频)") output_status = gr.Textbox(label="Status(状态)") run_button.click( fn=app.predict, inputs=[ ref_img, video, model_id, model, ], outputs=[output_video, output_status], ) example_data = [ ['./examples/mov/1/1.jpeg', './examples/mov/1/1.mp4', 'wan2.2-animate-move', 'wan-pro'], ['./examples/mov/2/2.jpeg', './examples/mov/2/2.mp4', 'wan2.2-animate-move', 'wan-pro'], ['./examples/mix/1/1.jpeg', './examples/mix/1/1.mp4', 'wan2.2-animate-mix', 'wan-pro'], ['./examples/mix/2/2.jpeg', './examples/mix/2/2.mp4', 'wan2.2-animate-mix', 'wan-pro'] ] if example_data: gr.Examples( examples=example_data, inputs=[ref_img, video, model_id, model], outputs=[output_video, output_status], fn=app.predict, cache_examples="lazy", ) demo.queue(default_concurrency_limit=100) demo.launch( server_name="0.0.0.0", server_port=7860 ) if __name__ == "__main__": start_app()