app/utils.py

"""
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)}"
        )