| 1 | """音频抽取管线。在调用转录 API 之前把视频抽成低带宽 mp3。 |
| 2 | |
| 3 | ffmpeg 二进制通过 ``imageio-ffmpeg`` PyPI 包提供并由 PyInstaller 打入 |
| 4 | sidecar onefile；运行时不依赖系统 ``PATH`` 上的 ffmpeg。 |
| 5 | |
| 6 | 设计参考： ``.kiro/specs/transcript-audio-extract-and-ui/design.md`` |
| 7 | """ |
| 8 | from __future__ import annotations |
| 9 | |
| 10 | import asyncio |
| 11 | import collections |
| 12 | import os |
| 13 | import time |
| 14 | from pathlib import Path |
| 15 | from typing import Optional |
| 16 | |
| 17 | from utils.logger import setup_logger |
| 18 | |
| 19 | logger = setup_logger("AudioExtraction") |
| 20 | |
| 21 | |
| 22 | # --------------------------------------------------------------------------- |
| 23 | # Public exceptions |
| 24 | # --------------------------------------------------------------------------- |
| 25 | |
| 26 | |
| 27 | class AudioExtractError(Exception): |
| 28 | """所有抽音失败的基类。 |
| 29 | |
| 30 | ``str(exc)`` 总以 ``audio_extract_failed: <cause>`` 开头，与 |
| 31 | requirements R6.2 / R6.3 字面契约一致；上层 |
| 32 | :class:`core.transcript_manager.TranscriptManager` 不再二次包装。 |
| 33 | """ |
| 34 | |
| 35 | cause: str = "unknown" |
| 36 | |
| 37 | def __init__(self, detail: str = "") -> None: |
| 38 | prefix = f"audio_extract_failed: {self.cause}" |
| 39 | super().__init__(f"{prefix}: {detail}" if detail else prefix) |
| 40 | |
| 41 | |
| 42 | class FfmpegNotAvailable(AudioExtractError): |
| 43 | """``imageio_ffmpeg.get_ffmpeg_exe()`` 找不到二进制、或 ``ffmpeg |
| 44 | -version`` 探测失败。""" |
| 45 | |
| 46 | cause = "ffmpeg_not_available" |
| 47 | |
| 48 | |
| 49 | class FfmpegTimeout(AudioExtractError): |
| 50 | """ffmpeg 抽音子进程在 :data:`_FFMPEG_TIMEOUT_SECONDS` 内未结束。""" |
| 51 | |
| 52 | cause = "audio_extract_timeout" |
| 53 | |
| 54 | |
| 55 | class FfmpegNonZeroExit(AudioExtractError): |
| 56 | """ffmpeg 抽音子进程以非零退出码结束。""" |
| 57 | |
| 58 | cause = "nonzero_exit_code" |
| 59 | |
| 60 | |
| 61 | class AudioExtractEmpty(AudioExtractError): |
| 62 | """ffmpeg 退出码为 0，但写出的目标文件不存在或大小为 0 字节。""" |
| 63 | |
| 64 | cause = "audio_extract_empty" |
| 65 | |
| 66 | |
| 67 | class PlatformUnsupported(AudioExtractError): |
| 68 | """``imageio-ffmpeg`` 在当前 OS / 架构上没有静态二进制（极少数 |
| 69 | 边角平台，例如某些 Linux ARM 子架构）。本质是 :class:`FfmpegNotAvailable` |
| 70 | 的特例，但用独立 ``cause`` 让上层日志能区分。""" |
| 71 | |
| 72 | cause = "platform_unsupported" |
| 73 | |
| 74 | |
| 75 | # --------------------------------------------------------------------------- |
| 76 | # FfmpegLocator |
| 77 | # --------------------------------------------------------------------------- |
| 78 | |
| 79 | |
| 80 | _AVAILABILITY_TTL_SECONDS = 60.0 |
| 81 | """可用性缓存的 TTL（requirements R2.7）。""" |
| 82 | |
| 83 | _VERSION_PROBE_TIMEOUT_SECONDS = 5.0 |
| 84 | """``ffmpeg -version`` 探测的硬超时（requirements R2.5）。""" |
| 85 | |
| 86 | |
| 87 | class FfmpegLocator: |
| 88 | """单例：缓存 ffmpeg 路径与可用性，避免每次抽音都重探。 |
| 89 | |
| 90 | 第一次调用 :meth:`locate` 触发 ``imageio_ffmpeg.get_ffmpeg_exe()`` |
| 91 | 取路径并跑一次 ``<ffmpeg> -version``。后续 60 秒内复用缓存。 |
| 92 | """ |
| 93 | |
| 94 | _instance: Optional["FfmpegLocator"] = None |
| 95 | |
| 96 | def __init__(self) -> None: |
| 97 | self._path: Optional[str] = None |
| 98 | self._version: Optional[str] = None |
| 99 | self._available: Optional[bool] = None |
| 100 | self._cached_at: float = 0.0 |
| 101 | self._last_error: Optional[str] = None |
| 102 | self._lock = asyncio.Lock() |
| 103 | |
| 104 | # -- public API --------------------------------------------------------- |
| 105 | |
| 106 | @classmethod |
| 107 | def instance(cls) -> "FfmpegLocator": |
| 108 | """模块级单例。在测试里通过 :meth:`reset_for_tests` 清掉。""" |
| 109 | if cls._instance is None: |
| 110 | cls._instance = cls() |
| 111 | return cls._instance |
| 112 | |
| 113 | @classmethod |
| 114 | def reset_for_tests(cls) -> None: |
| 115 | """仅供测试使用：清空单例与缓存。""" |
| 116 | cls._instance = None |
| 117 | |
| 118 | async def locate(self) -> str: |
| 119 | """返回可执行 ffmpeg 路径。 |
| 120 | |
| 121 | Raises: |
| 122 | FfmpegNotAvailable: 缓存中没有可用 ffmpeg 时（路径找不到、 |
| 123 | ``-version`` 探测失败、平台不支持等）。 |
| 124 | """ |
| 125 | async with self._lock: |
| 126 | await self._refresh_if_needed() |
| 127 | if not self._available: |
| 128 | raise FfmpegNotAvailable(self._last_error or "unknown") |
| 129 | assert self._path is not None # for mypy |
| 130 | return self._path |
| 131 | |
| 132 | async def diagnostic(self) -> dict: |
| 133 | """返回 ``GET /api/v1/transcript/diagnostic`` 的字段三元组。""" |
| 134 | async with self._lock: |
| 135 | await self._refresh_if_needed() |
| 136 | return { |
| 137 | "ffmpeg_available": bool(self._available), |
| 138 | "ffmpeg_path": self._path or "", |
| 139 | "ffmpeg_version": self._version, |
| 140 | } |
| 141 | |
| 142 | # -- internals ---------------------------------------------------------- |
| 143 | |
| 144 | async def _refresh_if_needed(self) -> None: |
| 145 | now = time.monotonic() |
| 146 | if ( |
| 147 | self._available is not None |
| 148 | and (now - self._cached_at) < _AVAILABILITY_TTL_SECONDS |
| 149 | ): |
| 150 | return |
| 151 | await self._probe() |
| 152 | self._cached_at = time.monotonic() |
| 153 | |
| 154 | async def _probe(self) -> None: |
| 155 | # Step 1: resolve binary path |
| 156 | try: |
| 157 | import imageio_ffmpeg # local import — keeps sidecar bootable |
| 158 | # if the optional dep is missing in a dev env |
| 159 | path = imageio_ffmpeg.get_ffmpeg_exe() |
| 160 | except Exception as exc: # imageio_ffmpeg raises RuntimeError on |
| 161 | # platforms without a bundled binary AND no system ffmpeg. |
| 162 | self._path = None |
| 163 | self._version = None |
| 164 | self._available = False |
| 165 | self._last_error = f"get_ffmpeg_exe failed: {exc!r}" |
| 166 | logger.warning( |
| 167 | "imageio_ffmpeg.get_ffmpeg_exe() failed: %r", exc |
| 168 | ) |
| 169 | return |
| 170 | |
| 171 | if not path or not os.path.exists(path): |
| 172 | self._path = path or None |
| 173 | self._version = None |
| 174 | self._available = False |
| 175 | self._last_error = f"ffmpeg path missing: {path!r}" |
| 176 | return |
| 177 | |
| 178 | # Step 2: probe `<ffmpeg> -version` |
| 179 | proc: Optional[asyncio.subprocess.Process] = None |
| 180 | try: |
| 181 | proc = await asyncio.create_subprocess_exec( |
| 182 | path, |
| 183 | "-version", |
| 184 | stdout=asyncio.subprocess.PIPE, |
| 185 | stderr=asyncio.subprocess.PIPE, |
| 186 | ) |
| 187 | stdout, _stderr = await asyncio.wait_for( |
| 188 | proc.communicate(), timeout=_VERSION_PROBE_TIMEOUT_SECONDS |
| 189 | ) |
| 190 | except asyncio.TimeoutError: |
| 191 | if proc is not None: |
| 192 | await _kill_and_reap(proc) |
| 193 | self._path = path |
| 194 | self._version = None |
| 195 | self._available = False |
| 196 | self._last_error = "ffmpeg -version timed out" |
| 197 | logger.warning("ffmpeg -version timed out at %s", path) |
| 198 | return |
| 199 | except Exception as exc: |
| 200 | if proc is not None: |
| 201 | await _kill_and_reap(proc) |
| 202 | self._path = path |
| 203 | self._version = None |
| 204 | self._available = False |
| 205 | self._last_error = f"subprocess error: {exc!r}" |
| 206 | logger.warning("ffmpeg -version subprocess error: %r", exc) |
| 207 | return |
| 208 | |
| 209 | if proc.returncode != 0 or b"ffmpeg version" not in stdout.lower(): |
| 210 | self._path = path |
| 211 | self._version = None |
| 212 | self._available = False |
| 213 | self._last_error = ( |
| 214 | f"ffmpeg -version exit={proc.returncode}, " |
| 215 | f"stdout_head={stdout[:64]!r}" |
| 216 | ) |
| 217 | return |
| 218 | |
| 219 | first_line = ( |
| 220 | stdout.split(b"\n", 1)[0] |
| 221 | .decode("utf-8", errors="replace") |
| 222 | .strip() |
| 223 | ) |
| 224 | self._path = path |
| 225 | self._version = first_line |
| 226 | self._available = True |
| 227 | self._last_error = None |
| 228 | |
| 229 | |
| 230 | # --------------------------------------------------------------------------- |
| 231 | # Public extraction API |
| 232 | # --------------------------------------------------------------------------- |
| 233 | |
| 234 | |
| 235 | _FFMPEG_TIMEOUT_SECONDS = 600.0 |
| 236 | """ffmpeg 抽音子进程硬超时（requirements R1.10 / R1.11）。""" |
| 237 | |
| 238 | _STDERR_RING_LIMIT_BYTES = 1 * 1024 * 1024 |
| 239 | """stderr 环形缓冲上限（requirements R1.12）。""" |
| 240 | |
| 241 | _STDERR_TAIL_BYTES = 4096 |
| 242 | """非零退出时返回给上层的 stderr 末尾字节数（requirements R1.13）。""" |
| 243 | |
| 244 | _FFMPEG_EXTRACT_ARGS = ( |
| 245 | "-vn", |
| 246 | "-ac", |
| 247 | "1", |
| 248 | "-ar", |
| 249 | "16000", |
| 250 | "-b:a", |
| 251 | "32k", |
| 252 | "-f", |
| 253 | "mp3", |
| 254 | ) |
| 255 | """固定参数列表（requirements R1.2）。tuple 而非 list 以防意外变更。""" |
| 256 | |
| 257 | |
| 258 | async def extract_audio( |
| 259 | video_path: Path, |
| 260 | output_dir: Path, |
| 261 | *, |
| 262 | locator: Optional[FfmpegLocator] = None, |
| 263 | ) -> Path: |
| 264 | """把 ``video_path`` 抽成 ``<stem>.mp3``，写到 ``output_dir`` 并返回路径。 |
| 265 | |
| 266 | 成功条件： |
| 267 | - ffmpeg 在 :data:`_FFMPEG_TIMEOUT_SECONDS` 内退出 |
| 268 | - 退出码 0 |
| 269 | - 输出文件大小严格大于 0 字节 |
| 270 | |
| 271 | Raises: |
| 272 | FfmpegNotAvailable: ffmpeg 二进制不可用（``locate()`` 抛出）。 |
| 273 | FfmpegTimeout: 子进程 600 秒未结束。 |
| 274 | FfmpegNonZeroExit: 子进程非零退出。 |
| 275 | AudioExtractEmpty: 子进程退出码 0，但输出文件不存在或为 0 字节。 |
| 276 | """ |
| 277 | locator = locator or FfmpegLocator.instance() |
| 278 | ffmpeg_path = await locator.locate() # may raise FfmpegNotAvailable |
| 279 | |
| 280 | output_dir.mkdir(parents=True, exist_ok=True) |
| 281 | output_path = output_dir / f"{video_path.stem}.mp3" |
| 282 | |
| 283 | args = ( |
| 284 | ffmpeg_path, |
| 285 | "-y", # overwrite output if it somehow already exists |
| 286 | "-i", |
| 287 | str(video_path), |
| 288 | *_FFMPEG_EXTRACT_ARGS, |
| 289 | str(output_path), |
| 290 | ) |
| 291 | |
| 292 | proc = await asyncio.create_subprocess_exec( |
| 293 | *args, |
| 294 | stdout=asyncio.subprocess.DEVNULL, |
| 295 | stderr=asyncio.subprocess.PIPE, |
| 296 | ) |
| 297 | |
| 298 | # Bounded stderr capture — bytes from the tail of the stream |
| 299 | # (requirements R1.12). ``maxlen`` on a deque of int (bytes) costs O(1) |
| 300 | # per append. |
| 301 | stderr_ring: collections.deque = collections.deque( |
| 302 | maxlen=_STDERR_RING_LIMIT_BYTES |
| 303 | ) |
| 304 | |
| 305 | async def _drain_stderr() -> None: |
| 306 | assert proc.stderr is not None # we set stderr=PIPE above |
| 307 | while True: |
| 308 | chunk = await proc.stderr.read(8192) |
| 309 | if not chunk: |
| 310 | break |
| 311 | stderr_ring.extend(chunk) |
| 312 | |
| 313 | try: |
| 314 | await asyncio.wait_for( |
| 315 | asyncio.gather(_drain_stderr(), proc.wait()), |
| 316 | timeout=_FFMPEG_TIMEOUT_SECONDS, |
| 317 | ) |
| 318 | except asyncio.TimeoutError: |
| 319 | await _kill_and_reap(proc) |
| 320 | _safe_unlink(output_path) |
| 321 | raise FfmpegTimeout(f"timeout after {int(_FFMPEG_TIMEOUT_SECONDS)}s") |
| 322 | except BaseException: |
| 323 | # Anything else (cancellation, drain_stderr crash, etc.) — leave |
| 324 | # the proc reaped so we don't orphan an ffmpeg process. Re-raise |
| 325 | # so the caller still sees the original failure. |
| 326 | await _kill_and_reap(proc) |
| 327 | _safe_unlink(output_path) |
| 328 | raise |
| 329 | |
| 330 | if proc.returncode != 0: |
| 331 | tail_bytes = bytes(stderr_ring)[-_STDERR_TAIL_BYTES:] |
| 332 | tail = tail_bytes.decode("utf-8", errors="replace") |
| 333 | _safe_unlink(output_path) |
| 334 | raise FfmpegNonZeroExit( |
| 335 | f"exit={proc.returncode}; stderr_tail={tail!r}" |
| 336 | ) |
| 337 | |
| 338 | try: |
| 339 | size = output_path.stat().st_size if output_path.exists() else 0 |
| 340 | except OSError: |
| 341 | size = 0 |
| 342 | if size <= 0: |
| 343 | _safe_unlink(output_path) |
| 344 | raise AudioExtractEmpty( |
| 345 | f"output missing or empty at {output_path}" |
| 346 | ) |
| 347 | |
| 348 | return output_path |
| 349 | |
| 350 | |
| 351 | async def _kill_and_reap(proc: asyncio.subprocess.Process) -> None: |
| 352 | """Best-effort: kill ``proc`` and wait up to 5 s for it to exit so the |
| 353 | caller doesn't leave an orphaned ffmpeg process behind. Used by both |
| 354 | the timeout path and the catch-all on ``extract_audio``'s gather. |
| 355 | """ |
| 356 | try: |
| 357 | proc.kill() |
| 358 | except ProcessLookupError: |
| 359 | # Already exited — nothing to do. |
| 360 | return |
| 361 | except Exception as exc: # pragma: no cover - defensive |
| 362 | logger.warning("failed to kill ffmpeg pid=%s: %r", proc.pid, exc) |
| 363 | try: |
| 364 | await asyncio.wait_for(proc.wait(), timeout=5.0) |
| 365 | except asyncio.TimeoutError: # pragma: no cover - 5s should be plenty |
| 366 | logger.warning( |
| 367 | "ffmpeg pid=%s did not exit within 5s after kill", proc.pid |
| 368 | ) |
| 369 | except Exception as exc: # pragma: no cover - defensive |
| 370 | logger.warning("ffmpeg pid=%s reap failed: %r", proc.pid, exc) |
| 371 | |
| 372 | |
| 373 | def _safe_unlink(path: Path) -> None: |
| 374 | """Delete ``path`` ignoring ``FileNotFoundError`` and other OS errors. |
| 375 | Used on failure paths where we want to leave no half-written file |
| 376 | behind but a cleanup failure must not eclipse the original error.""" |
| 377 | try: |
| 378 | path.unlink() |
| 379 | except FileNotFoundError: |
| 380 | pass |
| 381 | except OSError as exc: # pragma: no cover - defensive |
| 382 | logger.warning("failed to unlink %s: %r", path, exc) |
| 383 | |
| 384 | |
| 385 | __all__ = [ |
| 386 | "AudioExtractEmpty", |
| 387 | "AudioExtractError", |
| 388 | "FfmpegLocator", |
| 389 | "FfmpegNonZeroExit", |
| 390 | "FfmpegNotAvailable", |
| 391 | "FfmpegTimeout", |
| 392 | "PlatformUnsupported", |
| 393 | "extract_audio", |
| 394 | ] |
| 395 |