jiangchang-platform-kit/src/screencast/composer.py

"""FFmpeg 合成：帧序列 + SRT 字幕 + 背景音乐 → MP4。"""
from __future__ import annotations

import subprocess
import sys
from dataclasses import dataclass, field
from pathlib import Path

from jiangchang_skill_core.media_assets import (
    background_music_issue,
    pick_background_music,
    resolve_ffmpeg,
)

# 匠厂统一字幕样式 —— 所有 skill 录屏共用，做品牌一致性
_JIANGCHANG_SUBTITLE_STYLE = (
    "FontName=Arial,FontSize=14,"
    "PrimaryColour=&H00FFFFFF,OutlineColour=&H00000000,"
    "Outline=2,Shadow=1,Alignment=2"
)


@dataclass
class ComposeVideoResult:
    output_path: str
    warnings: list[str] = field(default_factory=list)


def _escape_srt_path(path: str) -> str:
    return path.replace("\\", "/").replace(":", "\\:")


def _run_ffmpeg(ffmpeg_exe: Path, cmd_tail: list[str]) -> tuple[bool, str]:
    cmd = [str(ffmpeg_exe), *cmd_tail]
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
        )
    except FileNotFoundError:
        return False, "ffmpeg not found"
    if result.returncode != 0:
        err = (result.stderr or result.stdout or "").strip()
        return False, err[:500] if err else f"exit {result.returncode}"
    return True, ""


def _media_assets_env(
    media_assets_root: str | Path | None,
    media_assets_env: dict[str, str] | None,
) -> dict[str, str] | None:
    if media_assets_root is None and media_assets_env is None:
        return None
    env = dict(media_assets_env or {})
    if media_assets_root is not None:
        env["MEDIA_ASSETS_ROOT"] = str(media_assets_root)
    return env


def compose_video(
    frames_dir: str,
    fps: int,
    subtitle_path: str,
    output_path: str,
    *,
    music_volume: float = 0.15,
    allow_no_music: bool = True,
    media_assets_root: str | Path | None = None,
    media_assets_env: dict[str, str] | None = None,
) -> ComposeVideoResult:
    """合成 MP4；ffmpeg 与背景音乐均走 jiangchang_skill_core.media_assets。"""
    frames_dir_path = Path(frames_dir)
    subtitle_path_obj = Path(subtitle_path)
    output_path_obj = Path(output_path)
    output_path_obj.parent.mkdir(parents=True, exist_ok=True)
    warnings: list[str] = []

    assets_env = _media_assets_env(media_assets_root, media_assets_env)

    ffmpeg_exe = resolve_ffmpeg(env=assets_env)
    if ffmpeg_exe is None or not Path(ffmpeg_exe).is_file():
        raise RuntimeError("ffmpeg_not_found")

    ffmpeg_path = Path(ffmpeg_exe)
    music_file = pick_background_music(env=assets_env)
    if music_file is None:
        issue = background_music_issue(env=assets_env)
        if issue:
            warnings.append(issue)
        else:
            warnings.append("background_music_missing")
        if not allow_no_music:
            raise FileNotFoundError(warnings[-1])

    srt_esc = _escape_srt_path(str(subtitle_path_obj.resolve()))
    subtitle_filter = (
        f"subtitles='{srt_esc}'"
        f":force_style='{_JIANGCHANG_SUBTITLE_STYLE}'"
    )

    if music_file and music_file.is_file():
        filter_complex = (
            f"[0:v]{subtitle_filter}[vout];"
            f"[1:a]volume={music_volume},apad[aout]"
        )
        cmd_tail = [
            "-y",
            "-framerate",
            str(fps),
            "-i",
            str(frames_dir_path / "frame_%06d.png"),
            "-i",
            str(music_file),
            "-filter_complex",
            filter_complex,
            "-map",
            "[vout]",
            "-map",
            "[aout]",
            "-c:v",
            "libx264",
            "-preset",
            "fast",
            "-crf",
            "23",
            "-c:a",
            "aac",
            "-b:a",
            "128k",
            "-shortest",
            "-pix_fmt",
            "yuv420p",
            str(output_path_obj),
        ]
        ok, err = _run_ffmpeg(ffmpeg_path, cmd_tail)
        if not ok:
            warnings.append(f"ffmpeg_with_music_failed: {err}")
            if not allow_no_music:
                raise RuntimeError(f"ffmpeg_compose_failed: {err}")
            music_file = None
        else:
            print(f"[screencast] 背景音乐: {music_file.name}")
            print(f"[screencast] 输出: {output_path_obj}")
            return ComposeVideoResult(str(output_path_obj.resolve()), warnings)

    cmd_tail = [
        "-y",
        "-framerate",
        str(fps),
        "-i",
        str(frames_dir_path / "frame_%06d.png"),
        "-vf",
        subtitle_filter,
        "-c:v",
        "libx264",
        "-preset",
        "fast",
        "-crf",
        "23",
        "-pix_fmt",
        "yuv420p",
        str(output_path_obj),
    ]
    print("[screencast] 运行 FFmpeg 合成（无背景音乐）...")
    ok, err = _run_ffmpeg(ffmpeg_path, cmd_tail)
    if not ok:
        print(err, file=sys.stderr)
        raise RuntimeError(f"ffmpeg_compose_failed: {err}")

    print(f"[screencast] 输出: {output_path_obj}")
    return ComposeVideoResult(str(output_path_obj.resolve()), warnings)