xiaoxia-saas/packages/shared/ffmpeg_utils.py

"""FFmpeg 共享工具 — packages/shared 层.

仅包含与业务无关的底层原语：FFmpeg/FFprobe 二进制路径、run_ffmpeg 执行器。
业务相关的滤镜构建、视频探测等留在 apps/worker/video_processing/ffmpeg_utils.py。

application 层和 worker 层都可以引用本模块，避免跨层依赖。
"""

from __future__ import annotations

import logging
import shutil
import subprocess  # nosec B404

logger = logging.getLogger(__name__)

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

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

# FFmpeg 执行默认超时（秒），防止 FFmpeg hang 住导致进程永久阻塞
# 默认 30 分钟，足够处理大部分短视频渲染；超长视频可单独传参覆盖
DEFAULT_FFMPEG_TIMEOUT = 1800


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


def run_ffmpeg(
    command: list[str],
    *,
    capture_output: bool = True,
    timeout: int | None = DEFAULT_FFMPEG_TIMEOUT,
) -> tuple[str, str]:
    """执行 FFmpeg 命令（统一入口）。

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

    Returns:
        (stdout, stderr) 元组

    Raises:
        subprocess.CalledProcessError: 命令执行失败时抛出，
            异常信息包含完整 stderr 以便排查。
        subprocess.TimeoutExpired: 超时未完成时抛出，FFmpeg 进程会被 kill。
    """
    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(
            "FFmpeg 命令超时 (%ds): command=%s",
            timeout or -1,
            " ".join(str(c) for c in command[:20]),
        )
        raise
    except subprocess.CalledProcessError as e:
        # 把完整 stderr 打到日志，方便排查 exit code 183 等问题
        stderr_text = (e.stderr or "").strip()
        logger.error(
            "FFmpeg 命令失败: exit_code=%d command=%s\nstderr:\n%s",
            e.returncode,
            " ".join(str(c) for c in command[:20]),  # 截断过长的命令
            stderr_text[:5000],  # 截断过长的 stderr
        )
        raise