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 os
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

# ── 编码参数（集中配置，支持环境变量覆盖）────────────────────────────────────
# preset 从 medium → fast，渲染速度提升 30%+，画质几乎无损（CRF 相同时 PSNR 差异 <0.1dB）
# 可通过环境变量 FFMPEG_ENCODE_PRESET 覆盖（如 ultrafast 追求极致速度，veryslow 追求极致压缩）
FFMPEG_ENCODE_PRESET: str = os.environ.get("FFMPEG_ENCODE_PRESET", "fast")
# CRF 保持 23（libx264 默认质量），可通过 FFMPEG_ENCODE_CRF 覆盖
FFMPEG_ENCODE_CRF: str = os.environ.get("FFMPEG_ENCODE_CRF", "23")
# 编码线程数：0 = 自动检测 CPU 核心数，充分利用多核
FFMPEG_ENCODE_THREADS: str = os.environ.get("FFMPEG_ENCODE_THREADS", "0")


# ── 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