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  # noqa: F401
from packages.domain.xfade_builder import XFADE_TRANSITION_MAP  # noqa: F401
from packages.domain.xfade_builder import XFade_TRANSITION_NAMES  # noqa: F401
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}


def random_edge_crop(
    input_path: str | Path,
    output_path: str | Path | None = None,
    *,
    min_crop_pct: float = 0.02,
    max_crop_pct: float = 0.05,
) -> Path:
    """对视频四边做随机裁剪再缩放回原分辨率，用于改变 pHash 指纹。

    Args:
        input_path: 输入视频路径
        output_path: 输出路径；为 None 时写入 input_path 同目录的临时文件，
                     成功后覆盖原文件
        min_crop_pct: 每边最小裁剪比例（默认 2%）
        max_crop_pct: 每边最大裁剪比例（默认 5%）

    Returns:
        输出文件路径（Path 对象）

    Raises:
        subprocess.CalledProcessError: ffmpeg 执行失败时抛出
    """
    import random
    import shutil
    import tempfile

    input_path = Path(input_path)

    # 获取原始分辨率
    info = probe_video_info(str(input_path))
    W = info["width"]
    H = info["height"]

    if W <= 0 or H <= 0:
        logger.warning("无法获取视频分辨率 (W=%d H=%d)，跳过裁剪: %s", W, H, input_path)
        return input_path

    # 四边各自随机裁剪 2%~5%
    crop_top = int(H * random.uniform(min_crop_pct, max_crop_pct))
    crop_bottom = int(H * random.uniform(min_crop_pct, max_crop_pct))
    crop_left = int(W * random.uniform(min_crop_pct, max_crop_pct))
    crop_right = int(W * random.uniform(min_crop_pct, max_crop_pct))

    # 裁剪后尺寸（确保至少 2 像素）
    new_w = max(W - crop_left - crop_right, 2)
    new_h = max(H - crop_top - crop_bottom, 2)
    x_offset = crop_left
    y_offset = crop_top

    # 确保裁剪尺寸为偶数（ffmpeg 编码器常要求偶数尺寸）
    new_w = new_w if new_w % 2 == 0 else new_w - 1
    new_h = new_h if new_h % 2 == 0 else new_h - 1
    if new_w < 2:
        new_w = 2
    if new_h < 2:
        new_h = 2

    # 输出分辨率必须与原始一致
    out_w = W if W % 2 == 0 else W + 1
    out_h = H if H % 2 == 0 else H + 1

    vf = f"crop={new_w}:{new_h}:{x_offset}:{y_offset},scale={out_w}:{out_h}"

    logger.info(
        "随机边缘裁剪: %s → crop(%d,%d,%d,%d)=%dx%d scale→%dx%d",
        input_path.name,
        crop_top,
        crop_bottom,
        crop_left,
        crop_right,
        new_w,
        new_h,
        out_w,
        out_h,
    )

    # 确定输出路径
    if output_path is None:
        temp_fd, temp_path = tempfile.mkstemp(suffix=".mp4", dir=input_path.parent)
        import os

        os.close(temp_fd)
        temp_output = Path(temp_path)
        replace_original = True
    else:
        temp_output = Path(output_path)
        replace_original = False

    command = [
        FFMPEG_BIN,
        "-y",
        "-i",
        str(input_path),
        "-vf",
        vf,
        "-c:v",
        "libx264",
        "-preset",
        "fast",
        "-crf",
        "18",
        "-c:a",
        "copy",
        "-movflags",
        "+faststart",
        str(temp_output),
    ]

    try:
        run_ffmpeg(command)
    except Exception:
        # 裁剪失败时清理临时文件
        if temp_output.exists() and replace_original:
            temp_output.unlink(missing_ok=True)
        raise

    # 成功 → 覆盖原文件
    if replace_original:
        shutil.move(str(temp_output), str(input_path))
        return input_path

    return temp_output