【工业化落地】AI 漫剧动态视频全链路流水线｜角色锁定 + 批量图生视频自动化方案

import json
import re
def parse_script(script_text: str):
    """
    解析漫剧分镜文本，输出结构化镜头列表
    输入示例：镜头01｜中景｜女主低头攥紧手掌，眼底含泪｜台词：我从来没有选择｜运镜：缓慢微推｜氛围：黄昏柔光，室内
    """
    shots = []
    pattern = re.compile(r"镜头(\d+)｜(.+?)｜(.+?)｜台词：(.+?)｜运镜：(.+?)｜氛围：(.+)")
    for line in script_text.splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        match = pattern.match(line)
        if match:
            shot_id, frame_type, action, dialogue, camera, atmosphere = match.groups()
            shots.append({
                "shot_id": int(shot_id),
                "frame_type": frame_type,
                "action_prompt": action,
                "dialogue": dialogue,
                "camera": camera,
                "atmosphere": atmosphere,
                "duration": 4,  # 默认单镜头时长s
                "width": 720,
                "height": 1280  # 竖屏漫剧尺寸
            })
    return shots
if __name__ == "__main__":
    raw_script = """
镜头01｜中景｜女主低头攥紧手掌，眼底含泪｜台词：我从来没有选择｜运镜：缓慢微推｜氛围：黄昏柔光，室内
镜头02｜近景｜男主皱眉看向远方，神色落寞｜台词：这条路没得回头｜运镜：固定镜头｜氛围：冷调夜色，窗边
    """
    result = parse_script(raw_script)
    with open("manju_shots.json", "w", encoding="utf-8") as f:
        json.dump(result, f, ensure_ascii=False, indent=2)
    print("✅ 分镜JSON保存完成：manju_shots.json")

{
  "IPAdapter": {
    "model": "ip-adapter-plus-face_sd15.safetensors",
    "reference_image": "./assets/char_heroine_ref.png",
    "weight": 0.68,
    "weight_type": "linear",
    "start_at": 0.0,
    "end_at": 1.0
  },
  "positive_base": "webtoon style, manhwa, clean line art, soft cel shading, detailed face, consistent character, {action}, {atmosphere}, {camera}",
  "negative_base": "blinking distortion, mutated hands, extra limbs, deformed face, flickering, ugly, disfigured, cross-eye, watermark, text"
}

import requests
import json
import os
COMFY_URL = "http://127.0.0.1:8188"
WORKFLOW_PATH = "./workflows/img2video_manju.json"
SHOT_JSON_PATH = "./manju_shots.json"
OUT_DIR = "./output/video_clips"
def queue_prompt(prompt_workflow):
    payload = {"prompt": prompt_workflow}
    resp = requests.post(f"{COMFY_URL}/prompt", json=payload)
    return resp.json()
if __name__ == "__main__":
    os.makedirs(OUT_DIR, exist_ok=True)
    # 加载基础工作流模板
    with open(WORKFLOW_PATH, "r", encoding="utf-8") as f:
        base_workflow = json.load(f)
    # 加载分镜数据
    with open(SHOT_JSON_PATH, "r", encoding="utf-8") as f:
        shot_list = json.load(f)
    for shot in shot_list:
        sid = shot["shot_id"]
        print(f"🎬 开始生成镜头 {sid}")
        # 注入参数到工作流，这里对应你ComfyUI里的节点ID
        base_workflow["6"]["inputs"]["image"] = f"./output/static_frames/shot_{sid}.png"
        base_workflow["12"]["inputs"]["motion_bucket_id"] = 120
        base_workflow["12"]["inputs"]["fps"] = 8
        base_workflow["12"]["inputs"]["num_frames"] = 32
        base_workflow["12"]["inputs"]["seed"] = sid * 888
        res = queue_prompt(base_workflow)
        print(f"镜头{sid}任务已入队，task_id：{res['prompt_id']}")
    print("✅ 全部视频片段任务提交完成，等待ComfyUI渲染")

import edge_tts
import asyncio
import srt
VOICE = "zh-CN-XiaoxiaoNeural"
async def gen_audio_sub(shot):
    text = shot["dialogue"]
    if not text.strip():
        return
    out_mp3 = f"./output/audio/shot_{shot['shot_id']}.mp3"
    communicate = edge_tts.Communicate(text, VOICE, rate="+10%")
    await communicate.save(out_mp3)
    # 简易字幕
    sub = srt.Subtitle(index=shot["shot_id"], start=srt.timedelta(0), end=srt.timedelta(seconds=shot["duration"]), content=text)
    return sub
async def main():
    with open(SHOT_JSON_PATH, "r", encoding="utf-8") as f:
        shots = json.load(f)
        subs = []
        for s in shots:
            sub = await gen_audio_sub(s)
            if sub:
                subs.append(sub)
        srt_content = srt.compose(subs)
        with open("./output/subtitle/all.srt", "w", encoding="utf-8") as f:
            f.write(srt_content)
if __name__ == "__main__":
    asyncio.run(main())