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

"""FFmpeg 工具函数 — Worker 层.

业务相关的滤镜构建、视频探测、视频标准化等能力放在这里；
底层原语（run_ffmpeg / 二进制路径 / 默认超时）已下沉到 packages/shared/ffmpeg_utils.py，
本模块 re-export 保持向后兼容。
"""

from __future__ import annotations

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

# 底层原语从 shared 层导入，application 层和 worker 层共用同一份实现
from shared.ffmpeg_utils import (  # noqa: F401
    DEFAULT_FFMPEG_TIMEOUT,
    FFMPEG_BIN,
    FFPROBE_BIN,
    run_ffmpeg,
)

logger = logging.getLogger(__name__)

# ── 常量（Worker 层业务相关） ────────────────────────────────────────────────

DEFAULT_OUTPUT_WIDTH = 1280
DEFAULT_OUTPUT_HEIGHT = 720
DEFAULT_FPS = 25

# xfade 转场映射：transition_effect 名称 → FFmpeg xfade transition 名称
# 键同时支持 TransitionEffect 枚举值和字符串名称（向后兼容）
# "cut" 为特殊值：硬切，不使用 xfade（由调用方特殊处理）
XFADE_TRANSITION_MAP: dict[str, str] = {
    # 基础
    "fade": "fade",
    "dissolve": "dissolve",
    "crossfade": "dissolve",
    "crossdissolve": "dissolve",
    # 滑入系列
    "slideleft": "slideleft",
    "slide_left": "slideleft",
    "slideright": "slideright",
    "slide_right": "slideright",
    "slideup": "slideup",
    "slide_up": "slideup",
    "slidedown": "slidedown",
    "slide_down": "slidedown",
    "slide": "slideleft",  # 默认向左滑
    # 缩放
    "zoom": "zoomin",
    "zoomin": "zoomin",
    "zoomout": "zoomout",
    # 擦除系列
    "wipe": "wipeleft",  # 默认向左擦
    "wipeleft": "wipeleft",
    "wiperight": "wiperight",
    "wipeup": "wipeup",
    "wipedown": "wipedown",
    # 特殊效果
    "circlecrop": "circlecrop",
    "circle": "circlecrop",
    "rectcrop": "rectcrop",
    "rect": "rectcrop",
}

DEFAULT_TRANSITION_DURATION = 0.5


# ── FFprobe 探测 ──────────────────────────────────────────────────────────────


def probe_has_audio(local_path: str | Path) -> bool:
    """探测文件是否包含音频流。

    Args:
        local_path: 本地文件路径

    Returns:
        True 表示有音频流（或探测失败保守返回），False 表示确认无音频流
    """
    try:
        result = subprocess.run(  # nosec B603
            [
                FFPROBE_BIN,
                "-v",
                "error",
                "-select_streams",
                "a:0",
                "-show_entries",
                "stream=codec_type",
                "-of",
                "default=noprint_wrappers=1:nokey=1",
                str(local_path),
            ],
            check=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            timeout=10,
        )
        return result.stdout.strip() == "audio"
    except Exception:
        # 探测失败保守返回 True，让 FFmpeg 自己处理（避免误删音频）
        return True


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,
            "video_codec": str, "audio_codec": str, "pix_fmt": str,
            "has_audio": bool,
        }
        失败时返回默认值。
    """
    try:
        result = subprocess.run(  # nosec B603
            [
                FFPROBE_BIN,
                "-v",
                "error",
                "-show_entries",
                "stream=width,height,r_frame_rate,duration,codec_name,codec_type,pix_fmt",
                "-show_entries",
                "format=duration",
                "-of",
                "json",
                video_path,
            ],
            check=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            timeout=15,
        )

        import json

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

        video_stream = next((s for s in streams if s.get("codec_type") == "video"), {})
        audio_stream = next((s for s in streams if s.get("codec_type") == "audio"), {})

        width = int(video_stream.get("width", DEFAULT_OUTPUT_WIDTH))
        height = int(video_stream.get("height", DEFAULT_OUTPUT_HEIGHT))
        video_codec = video_stream.get("codec_name", "") or ""
        pix_fmt = video_stream.get("pix_fmt", "") or ""

        # 解析帧率
        fps_str = video_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(video_stream.get("duration", 0))

        has_audio = bool(audio_stream)
        audio_codec = audio_stream.get("codec_name", "") or ""

        return {
            "width": width,
            "height": height,
            "duration": duration,
            "fps": round(fps, 2),
            "video_codec": video_codec,
            "audio_codec": audio_codec,
            "pix_fmt": pix_fmt,
            "has_audio": has_audio,
        }
    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,
            "video_codec": "",
            "audio_codec": "",
            "pix_fmt": "",
            "has_audio": True,
        }


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 转场滤镜链。

    对每步 xfade 自动钳制 transition duration，确保
    ``offset + td ≤ first_input_duration``，避免 FFmpeg exit 234。

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

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

    if n == 0:
        return "", 0.0

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

    # xfade 链 — 每步动态钳制 td，防止 offset + td > first_input_duration
    cumulative = 0.0
    prev_label = clip_video_labels[0]
    total_transition = 0.0  # 累计已使用的转场时长

    for i in range(1, n):
        cumulative += clip_durations[i - 1]

        # 当前 xfade 的第一个输入时长
        if i == 1:
            first_input_dur = clip_durations[0]
        else:
            first_input_dur = cumulative - total_transition

        # 原始 offset 计算
        offset = max(0.0, cumulative - transition_duration * i)

        # 安全钳制：offset + td 不能超过第一个输入的时长
        available = max(0.0, first_input_dur - offset)
        safe_td = min(transition_duration, available)

        # 同时不能超过剩余总时长
        remaining = max(0.0, sum(clip_durations) - cumulative)
        safe_td = min(safe_td, remaining)
        # 同时不能超过当前第二个输入（单个片段）的时长
        safe_td = min(safe_td, clip_durations[i])
        safe_td = max(0.001, safe_td)  # 至少 1ms，避免 td=0

        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={safe_td:.3f}"
            f":offset={offset:.3f}"
            f"[{out_label}]"
        )
        prev_label = out_label
        total_transition += safe_td

    # 总时长减去转场重叠部分
    total_duration = sum(clip_durations) - total_transition
    return ";".join(parts), max(0.0, total_duration)