video_analyzer.py

import subprocess
import base64
import os
import tempfile
import logging

log = logging.getLogger(__name__)

MAX_WIDTH = 768
JPEG_QUALITY = 4  # ffmpeg -q:v scale, lower = better quality
MIN_FRAMES = 6
MAX_FRAMES = 10


def _num_frames_for_duration(duration: float) -> int:
    """Scale sample count with duration so longer videos get more coverage, capped at MAX_FRAMES."""
    if duration <= 60:
        return MIN_FRAMES
    if duration <= 300:
        return 8
    return MAX_FRAMES


def _sample_fractions(num_frames: int) -> list[float]:
    # evenly spaced, skipping the very start/end so we don't sample black intro/outro frames
    return [0.05 + (0.90 * i / (num_frames - 1)) for i in range(num_frames)]


def _get_duration(path: str) -> float | None:
    result = subprocess.run(
        ["ffprobe", "-v", "error", "-show_entries", "format=duration",
         "-of", "default=noprint_wrappers=1:nokey=1", path],
        capture_output=True, text=True,
    )
    if result.returncode != 0 or not result.stdout.strip():
        log.warning("ffprobe failed for %s: %s", path, result.stderr)
        return None
    try:
        return float(result.stdout.strip())
    except ValueError:
        return None


def extract_frames(path: str, num_frames: int | None = None) -> list[tuple[str, float]] | None:
    """Extract evenly-spaced JPEG frames from a video, base64-encoded.

    Sample count scales with duration (up to MAX_FRAMES) unless `num_frames` is given explicitly.
    Returns a list of (base64_jpeg, timestamp_seconds), or None on failure.
    """
    if not os.path.exists(path):
        log.warning("video file not found: %s", path)
        return None

    duration = _get_duration(path)
    if duration is None or duration <= 0:
        return None

    if num_frames is None:
        num_frames = _num_frames_for_duration(duration)

    timestamps = [round(duration * f, 2) for f in _sample_fractions(num_frames)]

    frames: list[tuple[str, float]] = []
    with tempfile.TemporaryDirectory() as tmpdir:
        for i, ts in enumerate(timestamps):
            out_path = os.path.join(tmpdir, f"frame_{i}.jpg")
            result = subprocess.run(
                [
                    "ffmpeg", "-y", "-ss", str(ts), "-i", path,
                    "-frames:v", "1", "-vf", f"scale={MAX_WIDTH}:-1",
                    "-q:v", str(JPEG_QUALITY), out_path,
                ],
                capture_output=True, text=True,
            )
            if result.returncode != 0 or not os.path.exists(out_path):
                log.warning("frame extraction failed at %.2fs: %s", ts, result.stderr)
                continue
            with open(out_path, "rb") as f:
                data = f.read()
            frames.append((base64.b64encode(data).decode(), ts))

    if not frames:
        log.warning("no frames extracted from %s", path)
        return None

    log.info("extracted %d frames from %s (duration %.1fs)", len(frames), path, duration)
    return frames