Spaces:
Sleeping
Sleeping
File size: 3,633 Bytes
75ba57a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | """
LectureLens — FFmpeg / FFprobe Utilities
All subprocess helpers and temp-file management live here.
"""
from __future__ import annotations
import asyncio
import json
import os
import subprocess
import tempfile
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any
# ── FFmpeg runner ─────────────────────────────────────────────────────────────
def run_ffmpeg(args: list[str], *, capture_stderr: bool = True) -> str:
"""
Run FFmpeg synchronously and return stderr (where FFmpeg writes its output).
Raises RuntimeError if the process exits with a non-zero code.
"""
cmd = ["ffmpeg", "-hide_banner", "-y"] + args
result = subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
if result.returncode != 0 and result.returncode != 1:
# FFmpeg uses exit code 1 for warnings in some filter chains — tolerate it
raise RuntimeError(
f"FFmpeg failed (exit {result.returncode}):\n{result.stderr}"
)
return result.stderr if capture_stderr else result.stdout
def run_ffprobe(args: list[str]) -> dict[str, Any]:
"""
Run FFprobe with JSON output and return the parsed dict.
"""
cmd = ["ffprobe", "-v", "quiet", "-print_format", "json"] + args
result = subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
if result.returncode != 0:
raise RuntimeError(
f"FFprobe failed (exit {result.returncode}):\n{result.stderr}"
)
return json.loads(result.stdout)
async def run_ffmpeg_async(args: list[str]) -> str:
"""Async wrapper around run_ffmpeg for use inside async route handlers."""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, run_ffmpeg, args)
# ── Temporary file helpers ────────────────────────────────────────────────────
@asynccontextmanager
async def save_upload_to_tempfile(data: bytes, suffix: str):
"""
Async context manager: write upload bytes to a temp file and yield its path.
Guarantees cleanup even if an exception occurs.
"""
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=suffix)
try:
tmp.write(data)
tmp.flush()
tmp.close()
yield Path(tmp.name)
finally:
try:
os.unlink(tmp.name)
except FileNotFoundError:
pass
# ── Allowed media extensions ──────────────────────────────────────────────────
AUDIO_EXTENSIONS = {".m4a", ".wav", ".mp3", ".aac", ".ogg", ".flac"}
VIDEO_EXTENSIONS = {".mp4", ".mov", ".mkv", ".webm", ".avi"}
def detect_extension(filename: str) -> str:
"""Return lowercased file extension including the dot, e.g. '.mp4'."""
return Path(filename).suffix.lower()
def validate_extension(filename: str, media_type: str) -> None:
"""Raise ValueError if the extension does not match the declared media_type."""
ext = detect_extension(filename)
allowed = AUDIO_EXTENSIONS if media_type == "audio" else VIDEO_EXTENSIONS
if ext not in allowed:
raise ValueError(
f"File extension '{ext}' is not supported for media_type='{media_type}'. "
f"Allowed: {sorted(allowed)}"
)
|