cang-ying/video-ai-system/lib/composer.py

# LIB · 成片合成模块
# D180 · 2026-07-10
"""FFmpeg: 多镜头拼接 + 音频合成 + 字幕叠加 → 导出成片"""
import os, subprocess, json, shutil
from lib.models import get_db

FFMPEG = shutil.which("ffmpeg") or os.path.expanduser("~/.workbuddy/binaries/ffmpeg.exe")
PROJECT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

def concat_shots(episode_id, output_path=None, add_fade=True):
    """
    从数据库读取已完成镜头 → FFmpeg 拼接 → 导出成片
    
    Args:
        episode_id: 剧集ID
        output_path: 输出路径（默认 outputs/episodes/EP01-final.mp4）
        add_fade: 镜头间加淡入淡出
    """
    if not os.path.exists(FFMPEG):
        return {"error": f"ffmpeg not found: {FFMPEG}"}
    
    db = get_db()
    shots = db.execute("""
        SELECT id, shot_number, output_path, duration_sec FROM shots 
        WHERE episode_id = ? AND status = 'done' 
        ORDER BY shot_number
    """, (episode_id,)).fetchall()
    
    if not shots:
        return {"error": f"No completed shots for {episode_id}"}
    
    # 检查所有文件存在
    for s in shots:
        if not s["output_path"] or not os.path.exists(os.path.join(PROJECT, s["output_path"])):
            return {"error": f"Missing: {s['shot_number']} → {s['output_path']}"}
    
    if not output_path:
        out_dir = os.path.join(PROJECT, "outputs", "episodes")
        os.makedirs(out_dir, exist_ok=True)
        output_path = os.path.join(out_dir, f"{episode_id}-final.mp4")
    else:
        os.makedirs(os.path.dirname(output_path), exist_ok=True)
    
    # 生成 ffmpeg concat 文件列表（视频自带时长，无需手动指定 duration）
    concat_list = os.path.join(PROJECT, "outputs", "temp_concat.txt")
    with open(concat_list, "w", encoding="utf-8") as f:
        for s in shots:
            abs_path = os.path.join(PROJECT, s["output_path"]).replace("\\", "/")
            f.write(f"file '{abs_path}'\n")
    
    # FFmpeg concat
    cmd = [
        FFMPEG, "-y",
        "-f", "concat",
        "-safe", "0",
        "-i", concat_list,
        "-c:v", "libx264",
        "-preset", "fast",
        "-crf", "23",
        "-c:a", "aac",
        "-pix_fmt", "yuv420p",
        output_path
    ]
    
    r = subprocess.run(cmd, capture_output=True, text=True)
    os.unlink(concat_list)
    
    if r.returncode != 0:
        return {"error": r.stderr[:500]}
    
    return {
        "output_path": output_path,
        "shot_count": len(shots),
        "total_duration": sum(s["duration_sec"] for s in shots),
        "shots": [s["shot_number"] for s in shots]
    }

def add_audio_and_subtitle(video_path, audio_paths, subtitle_texts, output_path):
    """
    视频 + 配音 + 字幕 → 合成
    
    audio_paths: [{"path": "...", "start": 3.5}, ...]  音频文件和起始时间(秒)
    subtitle_texts: [{"text": "...", "start": 3.5, "end": 8.5}, ...]
    """
    if not os.path.exists(FFMPEG):
        return {"error": f"ffmpeg not found: {FFMPEG}"}
    
    asrc = ""  # no original audio
    
    # Build audio inputs
    audio_inputs = []
    audio_filters = []
    for i, a in enumerate(audio_paths):
        audio_inputs.extend(["-i", a["path"]])
        delay_ms = int(a["start"] * 1000)
        audio_filters.append(f"[{i+1}:a]adelay={delay_ms}|{delay_ms}[a{i}]")
    
    # Build subtitle filter
    sub_parts = []
    for s in subtitle_texts:
        esc = s["text"].replace("'", "\\'").replace(":", "\\:")
        sub_parts.append(f"drawtext=text='{esc}':fontsize=28:fontcolor=white:borderw=2:bordercolor=black:x=(w-text_w)/2:y=h-120:enable='between(t,{s['start']},{s['end']})'")
    sub_filter = ",".join(sub_parts) if sub_parts else "null"
    
    # Audio mix
    if audio_inputs:
        n = len(audio_paths)
        amix = "".join(audio_filters) + f"{''.join([f'[a{i}]' for i in range(n)])}amix=inputs={n}:duration=first[aout]"
        filter_complex = f"[0:v]{sub_filter}[vout];{amix}"
    else:
        filter_complex = f"[0:v]{sub_filter}[vout]"
    
    cmd = [FFMPEG, "-y", "-i", video_path] + audio_inputs + [
        "-filter_complex", filter_complex,
        "-map", "[vout]",
    ]
    if audio_inputs:
        cmd += ["-map", "[aout]", "-c:a", "aac"]
    else:
        cmd += ["-an"]
    cmd += [
        "-c:v", "libx264", "-preset", "fast", "-crf", "23",
        "-pix_fmt", "yuv420p",
        output_path
    ]
    
    r = subprocess.run(cmd, capture_output=True, text=True)
    return output_path if r.returncode == 0 else {"error": r.stderr[:500]}

def render_episode(episode_id):
    """
    一键成片：数据库 → 拼接所有已完成镜头 → 导出
    """
    print(f"🎬 合成 {episode_id}...")
    
    # Step 1: Concat all done shots
    result = concat_shots(episode_id)
    if "error" in result:
        return result
    
    print(f"✅ 拼接完成: {result['shot_count']}镜 {result['total_duration']}s")
    print(f"📁 {result['output_path']}")
    return result


def render_episode_full(episode_id, audio_config=None, subtitle_config=None, color_grade=None):
    """
    完整成片：拼接 → 调用 video-editor.js 做配音+BGM+字幕+调色+运镜 → 导出
    
    这是对 engines/video-editor.js 的 Python 封装。
    当需要完整成片（不只是拼接）时使用此函数。
    
    Args:
        episode_id: 剧集ID
        audio_config: {"voice": "path/to/voice.wav", "bgm": "path/to/bgm.mp3", "sfx": [...]}
        subtitle_config: {"srt": "path/to/subtitles.srt"}
        color_grade: {"brightness": 0, "contrast": 1.1, "saturation": 1.05}
    """
    import json as _json
    
    # Step 1: 先拼接所有已完成镜头
    concat_result = concat_shots(episode_id)
    if "error" in concat_result:
        return concat_result
    
    # Step 2: 构造 video-editor.js 的输入
    editor_input = {
        "timeline": [
            {
                "shot": concat_result["output_path"],
                "transition": "cut"
            }
        ],
        "resolution": {"w": 1080, "h": 1920},  # 竖屏9:16
        "fps": 24,
        "output": concat_result["output_path"].replace("-final.mp4", "-final-full.mp4")
    }
    
    if audio_config:
        editor_input["audio"] = audio_config
    if subtitle_config:
        editor_input["subtitle"] = subtitle_config.get("srt")
    if color_grade:
        editor_input["colorGrade"] = color_grade
    
    # Step 3: 调用 video-editor.js
    input_file = concat_result["output_path"] + ".editor-input.json"
    engines_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "engines")
    editor_js = os.path.join(engines_dir, "video-editor.js")
    
    if not os.path.exists(editor_js):
        print(f"⚠️ video-editor.js not found at {editor_js}, falling back to simple concat")
        return concat_result
    
    with open(input_file, "w", encoding="utf-8") as f:
        _json.dump(editor_input, f, ensure_ascii=False, indent=2)
    
    try:
        r = subprocess.run(
            ["node", "-e", f"const ed=require('{editor_js.replace(chr(92),'/')}');"
             f"const cfg=require('{input_file.replace(chr(92),'/')}');"
             f"ed.edit(cfg).then(r=>{{console.log(JSON.stringify(r));process.exit(0);}})"
             f".catch(e=>{{console.error(e.message);process.exit(1);}})"],
            capture_output=True, text=True, timeout=300
        )
        os.unlink(input_file)
        if r.returncode == 0:
            result = _json.loads(r.stdout)
            return {
                "output_path": result.get("outputPath", editor_input["output"]),
                "shot_count": concat_result["shot_count"],
                "total_duration": concat_result["total_duration"],
                "editor_used": True
            }
        return {"error": r.stderr[:500], "fallback": concat_result}
    except Exception as e:
        if os.path.exists(input_file): os.unlink(input_file)
        return {"error": str(e), "fallback": concat_result}

if __name__ == "__main__":
    import sys
    if len(sys.argv) < 2:
        print("Usage: python composer.py concat <episode_id>")
        print("       python composer.py render <episode_id>")
        sys.exit(1)
    cmd = sys.argv[1]
    ep = sys.argv[2] if len(sys.argv) > 2 else "DSV-EP01"
    if cmd == "concat":
        print(json.dumps(concat_shots(ep), indent=2, ensure_ascii=False))
    elif cmd == "render":
        print(json.dumps(render_episode(ep), indent=2, ensure_ascii=False))