Popiai-skill/social-media/hilo-promo/scripts/composite.py

#!/usr/bin/env python3
"""
最终合成脚本 - 将所有素材合成为 1080×1920 竖版推广视频

使用方式：
    # 动作模板：渲染字幕 → 拼接视频段 → 叠加尾板 → 混音
    python composite.py action \\
        --dance ./dance_video.mp4 \\
        --tutorial ./tutorial_segment.mp4 \\
        --voiceover ./voiceover.mp3 \\
        --bgm ./bgm.mp3 \\
        --subtitle-json ./voiceover_subtitle.json \\
        --font ./fonts/Poppins-Bold.ttf \\
        --endcard ./assets/endcard_en.mp4 \\
        --output ./final.mp4

    # 图片模板：写真轮播 → 教程预拼 → 底层合成 → 叠加PiP+字幕 → 混音
    python composite.py image \\
        --photos ./photos/photo_01.jpg ./photos/photo_02.jpg ./photos/photo_03.jpg \\
        --tutorial ./tut_combined.mp4 \\
        --pip ./digital_person/pip_clean.mov \\
        --voiceover ./voiceover.mp3 \\
        --subtitle-json ./voiceover_subtitle.json \\
        --font ./fonts/Poppins-Bold.ttf \\
        --endcard ./assets/endcard_en.mp4 \\
        --output ./final.mp4

依赖：
    - render_subtitles.py（字幕渲染）
    - tutorial_synthesis.py（教程段合成）
    - ffmpeg（视频处理）
"""

import argparse
import subprocess
import sys
from pathlib import Path
from typing import List, Optional

# ─────────────────────────── 配置 ───────────────────────────

CANVAS_W = 1080
CANVAS_H = 1920
TARGET_FPS = 30
BGM_VOLUME = 0.15

# 字幕位置
SUBTITLE_Y = 1050

# PiP 参数
PIP_SCALE_W = 520
PIP_SCALE_H = 693
PIP_X = -60
PIP_Y = 1227
PIP_START = 0.5


# ─────────────────────────── 辅助函数 ───────────────────────────


def ensure_dir(path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)


def run_ffmpeg(cmd: List[str], check: bool = True) -> None:
    """执行 ffmpeg 命令"""
    result = subprocess.run(cmd, capture_output=True, text=True)
    if result.returncode != 0:
        print(f"❌ ffmpeg 错误：\n{result.stderr[-800:]}")
        if check:
            sys.exit(1)


def render_subtitles(
    subtitle_json: str,
    font_path: str,
    output_dir: str,
) -> None:
    """调用 render_subtitles.py 渲染字幕 PNG"""
    script_path = Path(__file__).parent / "render_subtitles.py"
    cmd = [
        sys.executable,
        str(script_path),
        "--script",
        subtitle_json,
        "--font",
        font_path,
        "--output",
        output_dir,
        "--width",
        str(CANVAS_W),
        "--height",
        str(CANVAS_H),
    ]
    subprocess.run(cmd, check=True)


def concat_videos(video_paths: List[str], output: str) -> None:
    """用 concat demuxer 拼接多个视频"""
    tmp_dir = Path(output).parent / "_tmp_concat"
    tmp_dir.mkdir(parents=True, exist_ok=True)

    list_file = tmp_dir / "list.txt"
    tmp_files = []

    for i, vp in enumerate(video_paths):
        tmp_out = str(tmp_dir / f"seg_{i:02d}.mp4")
        run_ffmpeg(
            [
                "ffmpeg",
                "-y",
                "-i",
                vp,
                "-vf",
                f"scale={CANVAS_W}:{CANVAS_H},fps={TARGET_FPS},format=yuv420p,setsar=1",
                "-an",
                "-c:v",
                "libx264",
                "-pix_fmt",
                "yuv420p",
                tmp_out,
            ]
        )
        tmp_files.append(tmp_out)

    with open(list_file, "w") as f:
        for tf in tmp_files:
            f.write(f"file '{tf}'\n")

    run_ffmpeg(
        [
            "ffmpeg",
            "-y",
            "-f",
            "concat",
            "-safe",
            "0",
            "-i",
            str(list_file),
            "-c",
            "copy",
            output,
        ]
    )


def photo_carousel(
    photo_paths: List[str],
    each_duration: float,
    transition: str,
    output: str,
) -> None:
    """生成写真图轮播视频（xfade 转场，禁止硬切）"""
    if len(photo_paths) < 2:
        # 单张图直接循环
        run_ffmpeg(
            [
                "ffmpeg",
                "-y",
                "-loop",
                "1",
                "-t",
                str(each_duration),
                "-i",
                photo_paths[0],
                "-vf",
                f"scale={CANVAS_W}:{CANVAS_H},fps={TARGET_FPS},format=yuv420p",
                "-c:v",
                "libx264",
                "-pix_fmt",
                "yuv420p",
                "-t",
                str(each_duration),
                output,
            ]
        )
        return

    # 多张图用 xfade 连接
    tmp_dir = Path(output).parent / "_tmp_carousel"
    tmp_dir.mkdir(parents=True, exist_ok=True)

    inputs = []
    for p in photo_paths:
        inputs += ["-loop", "1", "-t", str(each_duration), "-i", p]

    # 构建 xfade filter chain
    streams = "[0:v]"
    for i in range(1, len(photo_paths)):
        xfade_dur = each_duration * 0.3
        offset = each_duration * i - xfade_dur
        streams += f"[{i}:v]xfade=transition={transition}:duration={xfade_dur}:offset={offset}[v{i}]"
        if i < len(photo_paths) - 1:
            streams += f";[v{i}]"

    total_dur = each_duration * len(photo_paths)

    n = len(photo_paths)
    xfade_dur = each_duration * 0.3

    # 先把每个图片流标准化
    filter_parts = []
    for i in range(n):
        filter_parts.append(
            f"[{i}:v]scale=1080:1920:force_original_aspect_ratio=decrease,pad=1080:1920:(ow-iw)/2:(oh-ih)/2,fps={TARGET_FPS},format=yuv420p,setsar=1[v{i}]"
        )

    # 链式 xfade：[v0][v1]xfade→[x1]; [x1][v2]xfade→[x2]; ...
    prev = "v0"
    for i in range(1, n):
        offset = each_duration * i - xfade_dur
        out_label = f"x{i}" if i < n - 1 else "xout"
        filter_parts.append(
            f"[{prev}][v{i}]xfade=transition={transition}:duration={xfade_dur}:offset={offset}[{out_label}]"
        )
        prev = out_label

    run_ffmpeg(
        [
            "ffmpeg",
            "-y",
            *inputs,
            "-filter_complex",
            ";".join(filter_parts),
            "-map",
            "[xout]",
            "-c:v",
            "libx264",
            "-pix_fmt",
            "yuv420p",
            "-t",
            str(total_dur),
            output,
        ]
    )


def overlay_pip(base_video: str, pip_video: str, start_time: float, output: str) -> str:
    """叠加数字人 PiP，返回叠加后的视频路径"""
    tmp = str(Path(output).parent / "_tmp_pip.mp4")
    run_ffmpeg(
        [
            "ffmpeg",
            "-y",
            "-i",
            base_video,
            "-i",
            pip_video,
            "-filter_complex",
            f"[1:v]scale={PIP_SCALE_W}:{PIP_SCALE_H}[pip];"
            f"[0:v][pip]overlay=x={PIP_X}:y={PIP_Y}:enable='gte(t,{start_time})'",
            "-c:v",
            "libx264",
            "-pix_fmt",
            "yuv420p",
            tmp,
        ]
    )
    return tmp


def overlay_subtitles(
    video: str, subtitle_dir: str, subtitle_json: str, output: str
) -> None:
    """按时间戳叠加字幕 PNG"""
    import json

    with open(subtitle_json) as f:
        data = json.load(f)
    segments = data.get("segments", data)
    if not segments:
        print("⚠️ 未找到字幕 segments，直接复制视频")
        subprocess.run(["cp", video, output], check=True)
        return

    # 构建字幕 overlay filter
    filters = []
    import os

    png_files: list[str] = [
        str(p) for p in sorted(Path(subtitle_dir).glob("sub_*.png"))
    ]
    for i, (seg, png) in enumerate(zip(segments, png_files)):
        start = seg["start"]
        end = seg["end"]
        filters.append(
            f"[{i}:v]format=rgba,overlay=x=0:y={SUBTITLE_Y}:enable='between(t,{start},{end})'[s{i}]"
        )
        if i == 0:
            base = f"-i{os.fspath(png)}"
        else:
            base = f"-i{os.fspath(png)}"

    # 简化：用 ffmpeg 的 subtitles filter 或逐个 overlay
    # 逐个 overlay
    cmd = ["ffmpeg", "-y", "-i", video]
    for png in png_files:
        cmd.extend(["-i", str(png)])

    filter_parts = []
    prev = "0:v"
    for i in range(len(png_files)):
        seg = segments[i]
        out_label = f"vs{i}" if i < len(png_files) - 1 else "vout"
        filter_parts.append(
            f"[{prev}][{i + 1}:v]overlay=0:{SUBTITLE_Y}:enable='between(t,{seg['start']},{seg['end']})'[{out_label}]"
        )
        prev = out_label
    filter_complex = ";".join(filter_parts)

    cmd += [
        "-filter_complex",
        filter_complex,
        "-map",
        "[vout]",
        "-c:v",
        "libx264",
        output,
    ]
    run_ffmpeg(cmd)


def mix_audio(
    video: str,
    voiceover: str,
    bgm: Optional[str],
    output: str,
) -> None:
    """混音：口播 + BGM + 尾板自带音频"""
    if bgm:
        run_ffmpeg(
            [
                "ffmpeg",
                "-y",
                "-i",
                video,
                "-i",
                voiceover,
                "-i",
                bgm,
                "-filter_complex",
                f"[1:a]volume=1.0[vo];[2:a]volume={BGM_VOLUME}[bg];"
                "[vo][bg]amix=inputs=2:duration=first[aout]",
                "-map",
                "0:v",
                "-map",
                "[aout]",
                "-c:v",
                "copy",
                "-c:a",
                "aac",
                "-shortest",
                output,
            ]
        )
    else:
        run_ffmpeg(
            [
                "ffmpeg",
                "-y",
                "-i",
                video,
                "-i",
                voiceover,
                "-filter_complex",
                "[1:a]volume=1.0[aout]",
                "-map",
                "0:v",
                "-map",
                "[aout]",
                "-c:v",
                "copy",
                "-c:a",
                "aac",
                "-shortest",
                output,
            ]
        )


# ─────────────────────────── 子命令 ───────────────────────────


def cmd_action(args: argparse.Namespace) -> None:
    """动作模板合成"""
    print("🎬 动作模板合成...")
    tmp_dir = Path(args.output).parent / "_tmp_action"
    tmp_dir.mkdir(parents=True, exist_ok=True)

    # 1. 渲染字幕
    subtitle_dir = str(tmp_dir / "subtitles")
    if args.subtitle_json:
        print("📝 渲染字幕...")
        render_subtitles(args.subtitle_json, args.font, subtitle_dir)

    # 2. 拼接底层视频（舞蹈 + 教程段）
    segments = [args.dance]
    if args.tutorial:
        segments.append(args.tutorial)
    base_video = str(tmp_dir / "base_concat.mp4")
    concat_videos(segments, base_video)
    print(f"   ✅ 底层拼接完成：{base_video}")

    # 3. 叠加字幕
    if args.subtitle_json:
        print("📝 叠加字幕...")
        with_subtitle = str(tmp_dir / "with_subtitle.mp4")
        overlay_subtitles(base_video, subtitle_dir, args.subtitle_json, with_subtitle)
        base_video = with_subtitle

    # 4. 拼接尾板
    print("🔗 拼接尾板...")
    with_endcard = str(tmp_dir / "with_endcard.mp4")
    concat_videos([base_video, args.endcard], with_endcard)

    # 5. 混音
    print("🎵 混音...")
    mix_audio(with_endcard, args.voiceover, args.bgm, args.output)

    print(f"✅ 完成：{args.output}")


def cmd_image(args: argparse.Namespace) -> None:
    """图片模板合成"""
    print("🖼️ 图片模板合成...")
    tmp_dir = Path(args.output).parent / "_tmp_image"
    tmp_dir.mkdir(parents=True, exist_ok=True)

    # 0. 渲染字幕
    subtitle_dir = str(tmp_dir / "subtitles")
    if args.subtitle_json:
        print("📝 渲染字幕...")
        render_subtitles(args.subtitle_json, args.font, subtitle_dir)

    # 1. 写真轮播
    print("📸 生成写真轮播...")
    carousel = str(tmp_dir / "carousel.mp4")
    photo_carousel(
        args.photos, each_duration=2.0, transition="slideleft", output=carousel
    )

    # 2. 教程预拼（调用 tutorial_synthesis.py pipeline）
    #    tut_combined.mp4 由 tutorial_synthesis.py pipeline 阶段生成，此处接收的是已拼接好的 tut_combined.mp4
    #    将 tut_combined.mp4 叠加到写真图3 上
    print("🎬 合成教程 overlay...")
    with_tutorial = str(tmp_dir / "with_tutorial.mp4")
    run_ffmpeg(
        [
            "ffmpeg",
            "-y",
            "-i",
            carousel,  # 底层：写真图1+2 xfade
            "-i",
            args.tutorial,  # tut_combined
            "-filter_complex",
            "[1:v]scale=1080:1920,fps=30,format=yuv420p,setsar=1[tt];"
            "[0:v][tt]overlay=0:0:enable='between(t,4.0,15.0)'[vout]",
            "-map",
            "[vout]",
            "-c:v",
            "libx264",
            "-pix_fmt",
            "yuv420p",
            with_tutorial,
        ]
    )

    # 3. 叠加 PiP
    if args.pip:
        print("🤳 叠加数字人 PiP...")
        with_pip = str(tmp_dir / "with_pip.mp4")
        pip_result = overlay_pip(
            with_tutorial, args.pip, start_time=PIP_START, output=with_pip
        )
        with_pip = pip_result  # overlay_pip 返回实际输出路径
    else:
        with_pip = with_tutorial

    # 4. 叠加字幕
    if args.subtitle_json:
        print("📝 叠加字幕...")
        with_subtitle = str(tmp_dir / "with_subtitle.mp4")
        overlay_subtitles(with_pip, subtitle_dir, args.subtitle_json, with_subtitle)
        with_pip = with_subtitle

    # 5. 拼接尾板
    print("🔗 拼接尾板...")
    with_endcard = str(tmp_dir / "with_endcard.mp4")
    concat_videos([with_pip, args.endcard], with_endcard)

    # 6. 混音（使用数字人视频自带音频 + BGM）
    print("🎵 混音...")
    mix_audio(with_endcard, args.voiceover, args.bgm, args.output)

    print(f"✅ 完成：{args.output}")


# ─────────────────────────── 主入口 ───────────────────────────


def main():
    parser = argparse.ArgumentParser(
        description="最终合成：动作模板 / 图片模板 → 1080×1920 竖版推广视频",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    subparsers = parser.add_subparsers(dest="cmd", required=True)

    # action 子命令
    p = subparsers.add_parser("action", help="动作模板合成")
    p.add_argument("--dance", required=True, help="舞蹈视频路径")
    p.add_argument("--tutorial", default=None, help="教程段视频路径（可选）")
    p.add_argument("--voiceover", required=True, help="口播音频路径")
    p.add_argument("--bgm", default=None, help="BGM 路径（可选）")
    p.add_argument("--subtitle-json", required=True, help="字幕时间戳 JSON")
    p.add_argument("--font", required=True, help="Poppins-Bold.ttf 字体路径")
    p.add_argument("--endcard", required=True, help="尾板视频路径")
    p.add_argument("--output", required=True, help="输出最终视频路径")

    # image 子命令
    p = subparsers.add_parser("image", help="图片模板合成")
    p.add_argument("--photos", nargs="+", required=True, help="写真图路径列表")
    p.add_argument(
        "--tutorial", required=True, help="教程预拼视频 tut_combined.mp4 路径"
    )
    p.add_argument("--pip", default=None, help="数字人抠像视频 .mov 路径（可选）")
    p.add_argument("--voiceover", required=True, help="口播音频路径")
    p.add_argument("--bgm", default=None, help="BGM 路径（可选）")
    p.add_argument("--subtitle-json", required=True, help="字幕时间戳 JSON")
    p.add_argument("--font", required=True, help="Poppins-Bold.ttf 字体路径")
    p.add_argument("--endcard", required=True, help="尾板视频路径")
    p.add_argument("--output", required=True, help="输出最终视频路径")

    args = parser.parse_args()

    if args.cmd == "action":
        cmd_action(args)
    elif args.cmd == "image":
        cmd_image(args)


if __name__ == "__main__":
    main()