零基础 AI 漫剧智能量产创作营：从剧本生成到视频渲染的全链路工程实践

┌─────────────────────────────────────────────────────────────┐
│                      用户控制台 (Web/CLI)                   │
└──────────────────────────┬──────────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────────┐
│                   编排引擎 (Celery + Redis)                  │
│   - 任务分解（剧本→分镜→图像→视频→合成）                     │
│   - 状态追踪 & 失败重试                                     │
└──────┬──────────┬──────────┬──────────┬────────────────────┘
       │          │          │          │
       ▼          ▼          ▼          ▼
┌──────────┐┌──────────┐┌──────────┐┌──────────┐
│ 剧本生成 ││ 分镜绘图 ││ 视频生成 ││ 配音合成 │
│ (LLM)   ││ (SDXL)   ││(AnimateDiff)││ (TTS)   │
└──────────┘└──────────┘└──────────┘└──────────┘
       │          │          │          │
       └──────────┴──────────┴──────────┘
                           │
                           ▼
                ┌─────────────────────┐
                │   COS 对象存储      │
                │ (原始素材/成品视频) │
                └─────────────────────┘

import os
import json
from openai import OpenAI  # 也可替换为本地 Qwen 服务

client = OpenAI(
    base_url=os.getenv("LLM_BASE_URL", "http://localhost:8000/v1"),
    api_key="dummy"
)

PROMPT_TEMPLATE = """
你是一个专业漫剧编剧。请根据以下故事梗概，生成 5~8 个分镜画面。
每个分镜需包含：
- scene_desc: 场景视觉描述（中文，50字内，含角色动作、表情、背景）
- dialogue: 该镜头的旁白/对白（中文）
- camera: 镜头运动（固定/推/拉/摇）
- duration: 预计停留秒数（2~4秒）

输出格式为 JSON 数组。
故事梗概：{story}
"""

def generate_script(story: str) -> list[dict]:
    response = client.chat.completions.create(
        model="qwen",
        messages=[{"role": "user", "content": PROMPT_TEMPLATE.format(story=story)}],
        temperature=0.7,
        response_format={"type": "json_object"}  # 确保输出合法 JSON
    )
    data = json.loads(response.choices[0].message.content)
    return data.get("scenes", [])

from diffusers import StableDiffusionXLPipeline, AutoencoderKL
from peft import LoraConfig, get_peft_model
import torch

# 加载基础 SDXL
pipe = StableDiffusionXLPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0",
    torch_dtype=torch.float16
)
# 添加 LoRA 层（秩=16）
lora_config = LoraConfig(
    r=16,
    lora_alpha=32,
    target_modules=["to_q", "to_v", "to_k", "to_out.0"],
    lora_dropout=0.05
)
pipe.unet = get_peft_model(pipe.unet, lora_config)
# ... 训练循环（省略，使用 diffusers 训练脚本）
# 保存 LoRA 权重
pipe.unet.save_pretrained("./lora_character_001")

from diffusers import StableDiffusionXLControlNetPipeline, ControlNetModel
import cv2
import numpy as np

controlnet = ControlNetModel.from_pretrained(
    "diffusers/controlnet-openpose-sdxl-1.0",
    torch_dtype=torch.float16
)
pipe = StableDiffusionXLControlNetPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0",
    controlnet=controlnet,
    torch_dtype=torch.float16
)
pipe.load_lora_weights("./lora_character_001")
pipe.to("cuda")

def generate_scene_image(prompt: str, pose_image_path: str) -> str:
    """生成分镜图，并保存到临时路径"""
    pose_img = cv2.imread(pose_image_path)
    # 这里 pose_image 可由预置动作库提供，或从上一帧自动提取
    image = pipe(
        prompt=prompt + ", best quality, masterpiece",
        negative_prompt="lowres, bad anatomy, worst quality",
        image=pose_img,
        num_inference_steps=25,
        guidance_scale=7.5
    ).images[0]
    out_path = f"/tmp/scene_{hash(prompt)}.png"
    image.save(out_path)
    return out_path

from diffusers import AnimateDiffPipeline, MotionAdapter
from diffusers.utils import export_to_video

adapter = MotionAdapter.from_pretrained("guoyww/animatediff-motion-adapter-v1-5-2")
pipe = AnimateDiffPipeline.from_pretrained(
    "emilianJR/epiCRealism",  # 写实风格 base
    motion_adapter=adapter,
    torch_dtype=torch.float16
)
pipe.scheduler = DDIMScheduler.from_pretrained(
    "emilianJR/epiCRealism",
    subfolder="scheduler",
    clip_sample=False,
    timestep_spacing="linspace",
    beta_schedule="linear"
)
pipe.enable_vae_slicing()
pipe.to("cuda")

def animate_scene(image_path: str, motion_scale: float = 1.0) -> str:
    """将单张图转为 16 帧 GIF/MP4，时长约 2s"""
    init_image = Image.open(image_path).convert("RGB")
    # 生成视频帧（batch size = 16）
    frames = pipe(
        image=init_image,
        prompt="",
        num_frames=16,
        motion_scale=motion_scale,
        decode_chunk_size=4,
    ).frames[0]
    video_path = image_path.replace(".png", ".mp4")
    export_to_video(frames, video_path, fps=8)
    return video_path

import edge_tts
import asyncio
import whisper

model = whisper.load_model("base")

async def generate_audio_and_subtitle(text: str, output_audio: str):
    """生成音频并返回字幕时间轴"""
    # 1. TTS 生成音频
    communicate = edge_tts.Communicate(text, voice="zh-CN-XiaoxiaoNeural")
    await communicate.save(output_audio)
    # 2. Whisper 对齐（生成带时间戳的转录）
    result = model.transcribe(output_audio, word_timestamps=True)
    # 3. 生成 SRT
    srt_content = ""
    for i, seg in enumerate(result["segments"]):
        start = seg["start"]
        end = seg["end"]
        text_seg = seg["text"]
        srt_content += f"{i+1}\n{format_time(start)} --> {format_time(end)}\n{text_seg}\n\n"
    srt_path = output_audio.replace(".mp3", ".srt")
    with open(srt_path, "w", encoding="utf-8") as f:
        f.write(srt_content)
    return srt_path

def format_time(seconds):
    h = int(seconds // 3600)
    m = int((seconds % 3600) // 60)
    s = int(seconds % 60)
    ms = int((seconds % 1) * 1000)
    return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"

import subprocess
import os

def compose_episode(scene_videos: list, audio_tracks: list, subtitle_paths: list, output_path: str):
    """
    使用 FFmpeg filter 拼接视频、叠加音频和字幕。
    scene_videos: 每幕视频路径
    audio_tracks: 每幕音频路径
    subtitle_paths: 每幕字幕 SRT 路径
    """
    # 1. 先拼接视频流（无音频）
    concat_inputs = "".join([f"-i {v} " for v in scene_videos])
    cmd_concat = f"ffmpeg {concat_inputs} -filter_complex 'concat=n={len(scene_videos)}:v=1:a=0' -c:v libx264 -pix_fmt yuv420p -f mp4 /tmp/video_only.mp4"
    subprocess.run(cmd_concat, shell=True, check=True)
    
    # 2. 拼接音频（使用 concat 协议）
    audio_list = "/tmp/audio_list.txt"
    with open(audio_list, "w") as f:
        for a in audio_tracks:
            f.write(f"file '{a}'\n")
    cmd_audio = f"ffmpeg -f concat -safe 0 -i {audio_list} -c:a aac /tmp/audio_only.m4a"
    subprocess.run(cmd_audio, shell=True, check=True)
    
    # 3. 合并视频和音频，并烧录字幕
    cmd_final = f"""
    ffmpeg -i /tmp/video_only.mp4 -i /tmp/audio_only.m4a \
           -filter_complex "subtitles=/tmp/subtitles_merged.srt:force_style='Fontsize=24,PrimaryColour=&HFFFFFF&'" \
           -c:v libx264 -c:a aac -shortest {output_path}
    """
    # 需要先合并所有 SRT（简单拼接，但需调整时间偏移，这里简化）
    # 实际我们会在步骤中生成全局时间戳
    subprocess.run(cmd_final, shell=True, check=True)

from celery import Celery, chain, group
from celery.result import AsyncResult

app = Celery('manga_pipeline', broker='redis://localhost:6379/0')

@app.task(bind=True)
def generate_script_task(self, story_prompt):
    scenes = generate_script(story_prompt)
    # 保存到 Redis 供后续任务使用
    self.backend.set(self.request.id + "_scenes", json.dumps(scenes))
    return len(scenes)

@app.task
def render_scene_task(scene_data, scene_index, role_lora_path):
    # scene_data: {prompt, pose_ref, dialogue, duration}
    img = generate_scene_image(scene_data['prompt'], scene_data['pose_ref'], role_lora_path)
    video = animate_scene(img, duration=scene_data['duration'])
    audio = asyncio.run(generate_audio_and_subtitle(scene_data['dialogue'], f"/tmp/audio_{scene_index}.mp3"))
    return {"video": video, "audio": audio, "subtitle": subtitle}

@app.task
def compose_all(results, output_key):
    # results 是 list of dict
    scene_videos = [r['video'] for r in results]
    audio_tracks = [r['audio'] for r in results]
    subtitle_paths = [r['subtitle'] for r in results]
    final_path = f"/output/{output_key}.mp4"
    compose_episode(scene_videos, audio_tracks, subtitle_paths, final_path)
    # 上传到腾讯云 COS
    from qcloud_cos import CosConfig, CosS3Client
    config = CosConfig(Region='ap-guangzhou', SecretId=os.getenv('COS_SECRET_ID'), SecretKey=os.getenv('COS_SECRET_KEY'))
    client = CosS3Client(config)
    client.upload_file(Bucket='manga-output', Key=f'episodes/{output_key}.mp4', LocalFilePath=final_path)
    return final_path

def launch_production(story_prompt, project_id):
    # 链式任务
    task_chain = chain(
        generate_script_task.s(story_prompt) |
        group(render_scene_task.s(i) for i in range(8)) |
        compose_all.s(project_id)
    )
    result = task_chain.apply_async()
    return result.id

# terraform/main.tf (片段)
resource "tencentcloud_instance" "gpu_worker" {
  instance_name   = "manga-worker-${count.index}"
  availability_zone = "ap-guangzhou-6"
  image_id        = "img-xxxx"  # GPU 基础镜像（Ubuntu 20.04 + CUDA 12.1）
  instance_type   = "GN7.2XLARGE40"  # A10 * 1
  system_disk_type = "CLOUD_SSD"
  system_disk_size = 100
  count           = 2
  
  user_data = <<-EOF
    #!/bin/bash
    cd /opt
    git clone https://github.com/your-org/manga-pipeline.git
    cd manga-pipeline
    pip install -r requirements.txt
    # 下载预训练模型
    python scripts/download_models.py
    # 启动 Celery Worker
    nohup celery -A tasks worker -l info --concurrency=4 &
  EOF
}