alfred/alfred/infrastructure/probe/ffprobe_prober.py

"""FfprobeMediaProber — MediaProber adapter backed by the ffprobe CLI."""

from __future__ import annotations

import json
import logging
import subprocess
from pathlib import Path

from alfred.domain.shared.media import AudioTrack, MediaInfo, SubtitleTrack, VideoTrack
from alfred.domain.shared.ports import SubtitleStreamInfo

logger = logging.getLogger(__name__)

_FFPROBE_TIMEOUT_SECONDS = 30

_FFPROBE_FULL_CMD = [
    "ffprobe",
    "-v",
    "quiet",
    "-print_format",
    "json",
    "-show_streams",
    "-show_format",
]


class FfprobeMediaProber:
    """Inspect media files by shelling out to ``ffprobe``.

    Implements :class:`alfred.domain.shared.ports.MediaProber` structurally.
    Never raises — failures are logged and surfaced as empty results.
    """

    def list_subtitle_streams(self, video: Path) -> list[SubtitleStreamInfo]:
        if not video.exists():
            return []
        try:
            result = subprocess.run(
                [
                    "ffprobe",
                    "-v",
                    "quiet",
                    "-print_format",
                    "json",
                    "-show_streams",
                    "-select_streams",
                    "s",
                    str(video),
                ],
                capture_output=True,
                text=True,
                timeout=_FFPROBE_TIMEOUT_SECONDS,
                check=False,
            )
            data = json.loads(result.stdout)
        except (
            subprocess.TimeoutExpired,
            json.JSONDecodeError,
            FileNotFoundError,
        ) as e:
            logger.debug(f"FfprobeMediaProber: ffprobe failed for {video.name}: {e}")
            return []

        streams: list[SubtitleStreamInfo] = []
        for stream in data.get("streams", []):
            tags = stream.get("tags", {}) or {}
            disposition = stream.get("disposition", {}) or {}
            streams.append(
                SubtitleStreamInfo(
                    language=tags.get("language") or None,
                    is_hearing_impaired=bool(disposition.get("hearing_impaired")),
                    is_forced=bool(disposition.get("forced")),
                )
            )
        return streams

    def probe(self, video: Path) -> MediaInfo | None:
        """Run ffprobe on ``video`` and return a :class:`MediaInfo`.

        Returns ``None`` when ffprobe is not available, times out, or
        the file cannot be parsed. Never raises.
        """
        try:
            result = subprocess.run(
                [*_FFPROBE_FULL_CMD, str(video)],
                capture_output=True,
                text=True,
                timeout=_FFPROBE_TIMEOUT_SECONDS,
                check=False,
            )
        except (subprocess.TimeoutExpired, FileNotFoundError) as e:
            logger.warning("ffprobe failed on %s: %s", video, e)
            return None

        if result.returncode != 0:
            logger.warning("ffprobe failed on %s: %s", video, result.stderr.strip())
            return None

        try:
            data = json.loads(result.stdout)
        except json.JSONDecodeError:
            logger.warning("ffprobe returned invalid JSON for %s", video)
            return None

        return _parse_media_info(data)


def _parse_media_info(data: dict) -> MediaInfo:
    """Translate raw ffprobe JSON into a :class:`MediaInfo` snapshot."""
    streams = data.get("streams", [])
    fmt = data.get("format", {})

    duration_seconds: float | None = None
    bitrate_kbps: int | None = None
    if "duration" in fmt:
        try:
            duration_seconds = float(fmt["duration"])
        except ValueError:
            pass
    if "bit_rate" in fmt:
        try:
            bitrate_kbps = int(fmt["bit_rate"]) // 1000
        except ValueError:
            pass

    video_tracks: list[VideoTrack] = []
    audio_tracks: list[AudioTrack] = []
    subtitle_tracks: list[SubtitleTrack] = []

    for stream in streams:
        codec_type = stream.get("codec_type")

        if codec_type == "video":
            video_tracks.append(
                VideoTrack(
                    index=stream.get("index", len(video_tracks)),
                    codec=stream.get("codec_name"),
                    width=stream.get("width"),
                    height=stream.get("height"),
                    is_default=stream.get("disposition", {}).get("default", 0) == 1,
                )
            )

        elif codec_type == "audio":
            audio_tracks.append(
                AudioTrack(
                    index=stream.get("index", len(audio_tracks)),
                    codec=stream.get("codec_name"),
                    channels=stream.get("channels"),
                    channel_layout=stream.get("channel_layout"),
                    language=stream.get("tags", {}).get("language"),
                    is_default=stream.get("disposition", {}).get("default", 0) == 1,
                )
            )

        elif codec_type == "subtitle":
            subtitle_tracks.append(
                SubtitleTrack(
                    index=stream.get("index", len(subtitle_tracks)),
                    codec=stream.get("codec_name"),
                    language=stream.get("tags", {}).get("language"),
                    is_default=stream.get("disposition", {}).get("default", 0) == 1,
                    is_forced=stream.get("disposition", {}).get("forced", 0) == 1,
                    is_sdh=stream.get("disposition", {}).get("hearing_impaired", 0) == 1,
                )
            )

    return MediaInfo(
        video_tracks=tuple(video_tracks),
        audio_tracks=tuple(audio_tracks),
        subtitle_tracks=tuple(subtitle_tracks),
        duration_seconds=duration_seconds,
        bitrate_kbps=bitrate_kbps,
    )