Wan2.1 图生视频 支持批量生成

Wan2.1 图生视频 支持批量生成

flyfish

综合效果

实现基于 Wan2.1 模型的配置化批量生成功能，支持从prompt.json读取多个 “图像 - 文本提示” 组合（每个任务可关联多图像），通过config.json集中管理模型路径、分辨率、帧数、引导强度等参数，自动根据图像高宽比调整尺寸并适配模型输入要求，利用负向提示词优化生成质量，可批量输出独立视频文件，并提供模型加载时间、单任务耗时等性能统计，具备模块化架构、错误容错和多设备支持能力，适合高效生成多样化视频素材。

综合效果使用的技术如下
AnyText2 在图片里玩文字而且还是所想即所得
Wan2.1 文生视频 支持批量生成、参数化配置和多语言提示词管理
Wan2.1 加速推理方法
Python 实现从 MP4 视频文件中平均提取指定数量的帧
可以生成一段视频，取尾帧之后，利用尾帧再次生成视频，这样就连续起来
Wan2.1 通过首尾帧生成视频
两张图像，从首帧过度到尾帧 例如 “魔塔” 两字 变为 “魔搭” 两字

配置说明

1. config.json（主配置文件）

作用：集中管理所有可配置参数，控制模型加载、生成过程和文件路径。

{
  "model": {
    "id": "/path/to/Wan2___1-I2V-14B-720P-Diffusers/",
    "torch_dtype": "bfloat16",
    "device_map": "balanced"
  },
  "generation": {
    "max_area": 921600,  // 对应720P
    "num_frames": 81,
    "guidance_scale": 5.0,
    "fps": 16,
    "output_prefix": "video_"
  },
  "prompts": {
    "file": "prompt.json",
    "prompt_key": "prompt",
    "image_paths_key": "image_paths"
  },
  "negative_prompt": {
    "file": "negative_prompt.txt",
    "default": "static,blurred,low quality"
  }
}

2. prompt.json（任务配置文件）

作用：定义多个“图像-提示词”任务，支持批量处理。

3. negative_prompt.txt（负向提示词文件）

negative_prompt.txt

Bright tones, overexposed, static, blurred details, subtitles, style, works, paintings, images, static, overall gray, worst quality, low quality, JPEG compression residue, ugly, incomplete, extra fingers, poorly drawn hands, poorly drawn faces, deformed, disfigured, misshapen limbs, fused fingers, still picture, messy background, three legs, many people in the background, walking backwards

作用：定义不希望出现在生成结果中的特征，提高生成质量。

完整代码

import os
import json
import time
import torch
import numpy as np
from PIL import Image
from diffusers import AutoencoderKLWan, WanImageToVideoPipeline
from diffusers.utils import export_to_video
from transformers import CLIPVisionModel

# 设置环境变量，禁用 tokenizers 的并行处理
os.environ["TOKENIZERS_PARALLELISM"] = "false"

# ----------------------
# 读取配置文件
# ----------------------
try:
    with open("config.json", "r", encoding="utf-8") as f:
        config = json.load(f)
        print("已加载配置文件")
except FileNotFoundError:
    print("错误: 未找到config.json文件")
    exit(1)
except json.JSONDecodeError:
    print("错误: config.json格式不正确")
    exit(1)

# ----------------------
# 解析配置参数
# ----------------------
# 模型配置
model_config = config.get("model", {})
model_id = model_config.get("id", "/path/to/default/model")
torch_dtype = getattr(torch, model_config.get("torch_dtype", "bfloat16"))
device_map = model_config.get("device_map", "auto")

# 生成参数
generation_config = config.get("generation", {})
max_area = generation_config.get("max_area", 720 * 1280)
num_frames = generation_config.get("num_frames", 17)
guidance_scale = generation_config.get("guidance_scale", 5.0)
fps = generation_config.get("fps", 16)
output_prefix = generation_config.get("output_prefix", "output_")

# 提示词配置
prompt_config = config.get("prompts", {})
prompt_file = prompt_config.get("file", "prompt.json")
prompt_key = prompt_config.get("prompt_key", "prompt")
image_paths_key = prompt_config.get("image_paths_key", "image_paths")

# 负向提示词配置
negative_config = config.get("negative_prompt", {})
negative_file = negative_config.get("file", "negative_prompt.txt")
default_negative = negative_config.get("default", "static,blurred,low quality")

# ----------------------
# 读取负向提示词
# ----------------------
try:
    with open(negative_file, "r", encoding="utf-8") as f:
        negative_prompt = f.read().strip()
    print(f"已加载负向提示词: {negative_prompt[:30]}...")
except FileNotFoundError:
    print(f"警告: 未找到{negative_file}，使用默认负向提示词")
    negative_prompt = default_negative

# ----------------------
# 读取批量任务配置
# ----------------------
try:
    with open(prompt_file, "r", encoding="utf-8") as f:
        batch_tasks = json.load(f)
    print(f"已加载{len(batch_tasks)}个生成任务")
except FileNotFoundError:
    print(f"错误: 未找到{prompt_file}")
    exit(1)

# ----------------------
# 模型初始化
# ----------------------
start_time = time.time()

# 加载模型组件
print("正在加载模型组件...")
image_encoder = CLIPVisionModel.from_pretrained(
    model_id, 
    subfolder="image_encoder", 
    torch_dtype=torch_dtype
)

vae = AutoencoderKLWan.from_pretrained(
    model_id, 
    subfolder="vae", 
    torch_dtype=torch_dtype
)

pipe = WanImageToVideoPipeline.from_pretrained(
    model_id, 
    vae=vae, 
    image_encoder=image_encoder, 
    torch_dtype=torch_dtype,
    device_map=device_map
)

model_load_time = time.time() - start_time
print(f"模型加载完成，耗时: {model_load_time:.2f}秒")

# ----------------------
# 批量处理任务
# ----------------------
total_tasks = 0
success_tasks = 0
total_generation_time = 0

for task_idx, task in enumerate(batch_tasks, 1):
    try:
        prompt = task.get(prompt_key, "")
        image_paths = task.get(image_paths_key, [])
        
        if not prompt:
            print(f"警告: 任务{task_idx}缺少prompt，跳过")
            continue
            
        if not image_paths:
            print(f"警告: 任务{task_idx}缺少image_paths，跳过")
            continue
            
        for img_idx, img_path in enumerate(image_paths, 1):
            total_tasks += 1
            print(f"\n--- 处理任务{task_idx}-图像{img_idx} ---")
            print(f"Prompt: {prompt[:50]}...")
            print(f"Image: {img_path}")
            
            # 打开并预处理图像
            try:
                image = Image.open(img_path).convert("RGB")
            except Exception as e:
                print(f"错误: 无法打开图像{img_path}: {e}")
                continue
                
            # 计算调整后的尺寸
            aspect_ratio = image.height / image.width
            mod_value = pipe.vae_scale_factor_spatial * pipe.transformer.config.patch_size[1]
            height = round(np.sqrt(max_area * aspect_ratio)) // mod_value * mod_value
            width = round(np.sqrt(max_area / aspect_ratio)) // mod_value * mod_value
            image = image.resize((width, height))
            
            print(f"调整尺寸为: {width}x{height}")
            
            # 生成视频
            gen_start_time = time.time()
            
            output = pipe(
                image=image,
                prompt=prompt,
                negative_prompt=negative_prompt,
                height=height,
                width=width,
                num_frames=num_frames,
                guidance_scale=guidance_scale
            ).frames[0]
            
            gen_time = time.time() - gen_start_time
            total_generation_time += gen_time
            
            # 保存视频
            output_path = f"{output_prefix}{task_idx}_{img_idx}.mp4"
            export_to_video(output, output_path, fps=fps)
            
            print(f"✅ 视频已保存至: {output_path}")
            print(f"⏱️ 生成耗时: {gen_time:.2f}秒")
            success_tasks += 1
            
    except Exception as e:
        print(f"错误: 处理任务{task_idx}时发生异常: {e}")
        continue

# ----------------------
# 统计信息
# ----------------------
print("\n==================== 处理完成 ====================")
print(f"总任务数: {total_tasks}")
print(f"成功数: {success_tasks}")
print(f"模型加载时间: {model_load_time:.2f}秒")
print(f"总生成时间: {total_generation_time:.2f}秒")
print(f"平均生成时间: {total_generation_time/max(1, success_tasks):.2f}秒/视频")

生成prompt.json

一个能够遍历指定文件夹，筛选出 jpg、jpeg、png 格式的图像文件，并按照如下JSON 格式输出的 Python 脚本。

[
    {
        "prompt": "内容",
        "image_paths": ["path/to/cat1.jpg"]
    },
    {
        "prompt": "内容",
        "image_paths": ["path/to/dog1.jpg"]
    }
]

代码

import os
import json

def generate_image_json(folder_path, output_file=None, prompt_value="动"):
    """
    遍历指定文件夹，收集图像文件路径并生成JSON格式数据。

    参数:
        folder_path (str): 要遍历的文件夹路径
        output_file (str, optional): JSON输出文件路径。默认为None，直接返回JSON数据。
        prompt_value (str, optional): JSON中使用的prompt值。默认为"动"。

    返回:
        list: 生成的JSON数据
    """
    # 定义支持的图像文件扩展名
    image_extensions = {'.jpg', '.jpeg', '.png'}
    
    # 检查文件夹是否存在
    if not os.path.exists(folder_path):
        raise FileNotFoundError(f"指定的文件夹不存在: {folder_path}")
    
    # 初始化结果列表
    result = []
    
    # 遍历文件夹中的所有文件和子文件夹
    for root, _, files in os.walk(folder_path):
        for file in files:
            # 获取文件扩展名并转换为小写
            file_ext = os.path.splitext(file)[1].lower()
            
            # 检查文件是否为支持的图像格式
            if file_ext in image_extensions:
                # 构建完整的文件路径
                full_path = os.path.join(root, file)
                
                # 添加到结果列表中
                result.append({
                    "prompt": prompt_value,
                    "image_paths": [full_path]
                })
    
    # 如果指定了输出文件，则写入文件
    if output_file:
        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(result, f, ensure_ascii=False, indent=4)
        print(f"JSON数据已写入: {output_file}")
    
    return result

if __name__ == "__main__":
    # 示例用法
    folder_to_scan = input("请输入要扫描的文件夹路径: ").strip()
    
    if not folder_to_scan:
        print("未输入文件夹路径，使用当前目录作为示例。")
        folder_to_scan = os.getcwd()
    
    try:
        # 生成JSON数据并打印
        json_data = generate_image_json(folder_to_scan)
        
        # 打印生成的JSON数据
        print(json.dumps(json_data, ensure_ascii=False, indent=4))
        
        # 可选: 将JSON数据保存到文件
        save_option = input("是否将结果保存到文件？(y/n): ").strip().lower()
        if save_option == 'y':
            output_path = input("请输入输出文件路径 (例如: images.json): ").strip()
            if output_path:
                generate_image_json(folder_to_scan, output_path)
    except Exception as e:
        print(f"发生错误: {e}")