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,
)

# xfade 转场纯逻辑已抽离到 domain 层，这里 re-export 保持向后兼容
from packages.domain.xfade_builder import DEFAULT_TRANSITION_DURATION as _default_transition_duration_base  # noqa: F401
from packages.domain.xfade_builder import (
    SUPPORTED_TRANSITIONS,
    XFADE_TRANSITION_MAP,
    XFade_TRANSITION_NAMES,
)
from packages.domain.xfade_builder import build_xfade_filter_chain as _build_xfade_filter_chain_base
from packages.domain.xfade_builder import chain_filters as _chain_filters_base
from packages.domain.xfade_builder import resolve_xfade_transition as _resolve_xfade_transition_base

logger = logging.getLogger(__name__)

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

DEFAULT_OUTPUT_WIDTH = 1280
DEFAULT_OUTPUT_HEIGHT = 720
DEFAULT_FPS = 25

# 向后兼容：DEFAULT_TRANSITION_DURATION 从 domain 层导出
DEFAULT_TRANSITION_DURATION = _default_transition_duration_base


# 向后兼容：薄包装函数
def chain_filters(filters: list[str], output_label: str, *, input_label: str = "0:v") -> str:
    return _chain_filters_base(filters, output_label, input_label=input_label)


def resolve_xfade_transition(transition_name: Any) -> str:
    return _resolve_xfade_transition_base(transition_name)


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]:
    return _build_xfade_filter_chain_base(
        clip_durations,
        clip_video_labels,
        transitions,
        transition_duration=transition_duration,
        output_label=output_label,
    )


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


def run_ffprobe(
    command: list[str],
    *,
    capture_output: bool = True,
    timeout: int = 30,
) -> tuple[str, str]:
    """执行 FFprobe 命令。

    Args:
        command: 完整的 ffprobe 命令列表（含 "ffprobe" 本身）
        capture_output: 是否捕获 stdout/stderr
        timeout: 超时时间（秒），默认 30s；None 表示不设超时

    Returns:
        (stdout, stderr) 元组

    Raises:
        subprocess.CalledProcessError: 命令执行失败时抛出
        subprocess.TimeoutExpired: 超时未完成时抛出
    """
    try:
        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,
            timeout=timeout,
        )
        return (result.stdout or "", result.stderr or "")
    except subprocess.TimeoutExpired:
        logger.error(
            "FFprobe 命令超时 (%ds): command=%s",
            timeout or -1,
            " ".join(str(c) for c in command[:20]),
        )
        raise
    except subprocess.CalledProcessError as e:
        stderr_text = (e.stderr or "").strip()
        logger.error(
            "FFprobe 命令失败: exit_code=%d command=%s\nstderr:\n%s",
            e.returncode,
            " ".join(str(c) for c in command[:20]),
            stderr_text[:5000],
        )
        raise


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}