xiaoxia-saas/apps/worker/video_processing/ffmpeg_utils.py

"""FFmpeg 工具函数 — 从 editing_modes.py / video_compose_service.py 提取的共享原语.

提供 FFmpeg / FFprobe 调用、视频信息探测、视频标准化、xfade 转场滤镜构建
等底层能力，供 EditingModeProcessor、VideoComposeService、UnifiedRenderService
共同复用。
"""

from __future__ import annotations

import logging
import shutil
import subprocess  # nosec B404
from pathlib import Path
from typing import Any

logger = logging.getLogger(__name__)

# ── 常量 ──────────────────────────────────────────────────────────────────────

FFMPEG_BIN: str = shutil.which("ffmpeg") or "ffmpeg"
FFPROBE_BIN: str = shutil.which("ffprobe") or "ffprobe"

DEFAULT_OUTPUT_WIDTH = 1280
DEFAULT_OUTPUT_HEIGHT = 720
DEFAULT_FPS = 25

# xfade 转场映射：transition_effect 名称 → FFmpeg xfade transition 名称
# 键同时支持 TransitionEffect 枚举值和字符串名称（向后兼容）
XFADE_TRANSITION_MAP: dict[str, str] = {
    "fade": "fade",
    "slideleft": "slideleft",
    "slide_left": "slideleft",
    "slideright": "slideright",
    "slide_right": "slideright",
    "dissolve": "dissolve",
    "wipe": "wipeleft",
    "wipeleft": "wipeleft",
}

DEFAULT_TRANSITION_DURATION = 0.5


# ── FFmpeg 执行 ───────────────────────────────────────────────────────────────


def run_ffmpeg(
    command: list[str],
    *,
    capture_output: bool = True,
) -> tuple[str, str]:
    """执行 FFmpeg 命令。

    Args:
        command: 完整的 ffmpeg 命令列表（含 "ffmpeg" 本身）
        capture_output: 是否捕获 stdout/stderr

    Returns:
        (stdout, stderr) 元组

    Raises:
        subprocess.CalledProcessError: 命令执行失败时抛出
    """
    result = subprocess.run(  # nosec B603
        command,
        check=True,
        stdout=subprocess.PIPE if capture_output else None,
        stderr=subprocess.PIPE if capture_output else None,
        text=True,
    )
    return (result.stdout or "", result.stderr or "")


def probe_duration(local_path: str | Path) -> float:
    """用 ffprobe 获取视频时长（秒）。

    失败时返回默认值 5.0 秒。
    """
    try:
        result = subprocess.run(  # nosec B603
            [
                FFPROBE_BIN,
                "-v",
                "error",
                "-show_entries",
                "format=duration",
                "-of",
                "default=noprint_wrappers=1:nokey=1",
                str(local_path),
            ],
            check=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )
        return round(float(result.stdout.strip()), 3)
    except Exception:
        return 5.0


def probe_video_info(video_path: str) -> dict[str, Any]:
    """获取视频信息（宽、高、时长、fps）。

    Returns:
        {"width": int, "height": int, "duration": float, "fps": float}
        失败时返回默认值。
    """
    try:
        result = subprocess.run(  # nosec B603
            [
                FFPROBE_BIN,
                "-v",
                "error",
                "-select_streams",
                "v:0",
                "-show_entries",
                "stream=width,height,r_frame_rate,duration",
                "-show_entries",
                "format=duration",
                "-of",
                "json",
                video_path,
            ],
            check=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )

        import json

        info = json.loads(result.stdout)
        stream = info.get("streams", [{}])[0]
        fmt = info.get("format", {})

        width = int(stream.get("width", DEFAULT_OUTPUT_WIDTH))
        height = int(stream.get("height", DEFAULT_OUTPUT_HEIGHT))

        # 解析帧率
        fps_str = stream.get("r_frame_rate", "25/1")
        if "/" in fps_str:
            num, den = fps_str.split("/")
            fps = float(num) / float(den) if float(den) > 0 else DEFAULT_FPS
        else:
            fps = float(fps_str) if fps_str else DEFAULT_FPS

        # 时长
        duration = float(fmt.get("duration", 0)) or float(stream.get("duration", 0))

        return {
            "width": width,
            "height": height,
            "duration": duration,
            "fps": round(fps, 2),
        }
    except Exception as e:
        logger.warning("获取视频信息失败: %s, error: %s", video_path, e)
        return {
            "width": DEFAULT_OUTPUT_WIDTH,
            "height": DEFAULT_OUTPUT_HEIGHT,
            "duration": 0.0,
            "fps": DEFAULT_FPS,
        }


def normalize_video(
    input_path: str,
    output_path: str,
    *,
    width: int = DEFAULT_OUTPUT_WIDTH,
    height: int = DEFAULT_OUTPUT_HEIGHT,
    fps: int = DEFAULT_FPS,
) -> dict[str, Any]:
    """标准化视频（缩放 + 恒定帧率）。

    使用 scale + pad 保持宽高比，黑边填充到目标分辨率。

    Returns:
        {"width": int, "height": int, "path": str}
    """
    command = [
        FFMPEG_BIN,
        "-y",
        "-i",
        input_path,
        "-vf",
        f"scale={width}:{height}:force_original_aspect_ratio=decrease,"
        f"pad={width}:{height}:(ow-iw)/2:(oh-ih)/2:black,"
        f"fps={fps}",
        "-c:v",
        "libx264",
        "-crf",
        "23",
        "-preset",
        "medium",
        "-c:a",
        "aac",
        "-b:a",
        "128k",
        "-movflags",
        "+faststart",
        output_path,
    ]
    run_ffmpeg(command)
    return {"width": width, "height": height, "path": output_path}


# ── xfade / concat 滤镜构建 ──────────────────────────────────────────────────


def chain_filters(filters: list[str], output_label: str, *, input_label: str = "0:v") -> str:
    """将滤镜列表串联为 FFmpeg 滤镜字符串。

    例：chain_filters(["scale=1280:720", "fps=25"], "v0")
    → "[0:v]scale=1280:720,fps=25[v0]"
    """
    filter_body = ",".join(filters)
    return f"[{input_label}]{filter_body}[{output_label}]"


def resolve_xfade_transition(transition_name: str) -> str:
    """将转场效果名称映射为 FFmpeg xfade transition 名称。

    支持 TransitionEffect 枚举值和字符串名称，未知值回退到 "fade"。
    """
    # 兼容 TransitionEffect 枚举（有 .value 属性）
    if hasattr(transition_name, "value"):
        transition_name = transition_name.value
    return XFADE_TRANSITION_MAP.get(transition_name, "fade")


def build_xfade_filter_chain(
    clip_durations: list[float],
    clip_video_labels: list[str],
    transitions: list[str],
    *,
    transition_duration: float = DEFAULT_TRANSITION_DURATION,
    output_label: str = "outv",
) -> tuple[str, float]:
    """构建 xfade 转场滤镜链。

    Args:
        clip_durations: 每个片段的时长
        clip_video_labels: 每个片段的视频流标签（如 "v0", "v1"）
        transitions: 每个片段对应的转场效果（第一个片段的转场被忽略）
        transition_duration: 转场时长（秒）
        output_label: 最终输出标签

    Returns:
        (filter_string, estimated_total_duration)
    """
    n = len(clip_durations)
    parts: list[str] = []
    total_duration = sum(clip_durations)

    if n == 0:
        return "", 0.0

    if n == 1:
        parts.append(f"[{clip_video_labels[0]}]copy[{output_label}]")
        return ";".join(parts), total_duration

    # xfade 链
    cumulative = 0.0
    prev_label = clip_video_labels[0]

    for i in range(1, n):
        cumulative += clip_durations[i - 1]
        offset = max(0.0, cumulative - transition_duration * i)

        transition = transitions[i] if i < len(transitions) else "cut"
        xfade_transition = resolve_xfade_transition(transition)

        if i == n - 1:
            out_label = output_label
        else:
            out_label = f"xf{i}"

        parts.append(
            f"[{prev_label}][{clip_video_labels[i]}]"
            f"xfade=transition={xfade_transition}"
            f":duration={transition_duration}"
            f":offset={offset:.3f}"
            f"[{out_label}]"
        )
        prev_label = out_label

    # 总时长减去转场重叠部分
    total_duration -= transition_duration * (n - 1)
    return ";".join(parts), max(0.0, total_duration)