fix: 渲染引擎全链路安全加固 P0+P1 (#317)

apps/worker/video_processing/ffmpeg_utils.py

@@ -119,6 +119,54 @@ def run_ffmpeg(
         raise
 
 
+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:
     """探测文件是否包含音频流。