utils/video_utils.py

"""Video utility functions for codec conversion and temp handling."""

from __future__ import annotations

import os
import shutil
import subprocess
from pathlib import Path
from typing import Optional


def _probe_codec(video_path: str) -> Optional[str]:
    """Return the codec name for the first video stream using ffprobe."""
    try:
        result = subprocess.run(
            [
                "ffprobe",
                "-v",
                "quiet",
                "-select_streams",
                "v:0",
                "-show_entries",
                "stream=codec_name",
                "-of",
                "csv=p=0",
                video_path,
            ],
            capture_output=True,
            text=True,
            check=False,
        )
        if result.returncode == 0:
            return result.stdout.strip().lower()
    except FileNotFoundError:
        pass
    except Exception:
        pass
    return None


def convert_video_to_h264(input_path: str, output_path: Optional[str] = None) -> str:
    """Convert *input_path* to H.264 if needed and return the playable path.

    If the file already uses an H.264 video stream, the original path is
    returned. When conversion is required, a new MP4 file is written (leaving
    the original untouched). The new path is returned, or the original path if
    conversion fails.
    """

    if not input_path:
        return input_path

    input_path = str(Path(input_path))
    if not os.path.isfile(input_path):
        return input_path

    codec = _probe_codec(input_path)
    if codec and "h264" in codec:
        # Already compatible
        return input_path

    base = Path(input_path)
    if output_path is None:
        output_path = str(base.with_name(f"{base.stem}_h264.mp4"))

    try:
        cmd = [
            "ffmpeg",
            "-y",
            "-i",
            input_path,
            "-vcodec",
            "libx264",
            "-acodec",
            "aac",
            "-pix_fmt",
            "yuv420p",
            output_path,
        ]
        result = subprocess.run(cmd, capture_output=True, text=True, check=False)
        if result.returncode == 0 and os.path.isfile(output_path):
            return output_path
        # If conversion failed, fall back to original path.
        return input_path
    except FileNotFoundError:
        # ffmpeg missing; nothing we can do besides returning original
        return input_path
    except Exception:
        return input_path


def copy_video_to_temp(video_path: str, temp_dir: Optional[str] = None) -> str:
    """Copy *video_path* to *temp_dir* (or a default temp) and return new path."""

    if not video_path:
        return video_path

    if temp_dir is None:
        temp_dir = Path("/tmp") / f"video_temp_{os.getpid()}"
    temp_dir = Path(temp_dir)
    temp_dir.mkdir(parents=True, exist_ok=True)

    src = Path(video_path)
    if not src.exists():
        return video_path

    dest = temp_dir / src.name
    try:
        shutil.copy2(src, dest)
        return str(dest)
    except Exception:
        return video_path