feat: add ffprobe media inspection

src/voice_transcriptor/services/__init__.py

@@ -0,0 +1,2 @@
+"""Application services independent from the GUI."""
+

src/voice_transcriptor/services/media_probe.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import json
+import shutil
+import subprocess
+from pathlib import Path
+from typing import Any
+
+from voice_transcriptor.models import MediaInfo, ToolStatus
+
+
+class MediaProbeError(RuntimeError):
+    """A media file could not be validated or inspected."""
+
+
+class InvalidMediaPathError(MediaProbeError):
+    """The selected path is not a readable local file."""
+
+
+class ProbeOutputError(MediaProbeError):
+    """FFprobe returned output that cannot be interpreted."""
+
+
+class ProbeExecutionError(MediaProbeError):
+    """FFprobe could not start or inspect the selected media."""
+
+
+class ProbeTimeoutError(MediaProbeError):
+    """FFprobe did not finish before its configured timeout."""
+
+
+def detect_tools() -> ToolStatus:
+    ffmpeg = shutil.which("ffmpeg")
+    ffprobe = shutil.which("ffprobe")
+    return ToolStatus(Path(ffmpeg) if ffmpeg else None, Path(ffprobe) if ffprobe else None)
+
+
+def validate_media_path(path: Path) -> Path:
+    path = path.expanduser()
+    if not path.exists():
+        raise InvalidMediaPathError("The selected file does not exist.")
+    if not path.is_file():
+        raise InvalidMediaPathError("The selected path is not a file.")
+    try:
+        path.stat()
+    except OSError as exc:
+        raise InvalidMediaPathError("The selected file cannot be read.") from exc
+    return path.resolve()
+
+
+def parse_probe_output(path: Path, size_bytes: int, payload: str) -> MediaInfo:
+    try:
+        data: dict[str, Any] = json.loads(payload)
+    except (json.JSONDecodeError, TypeError) as exc:
+        raise ProbeOutputError("FFprobe returned invalid data.") from exc
+    duration: float | None = None
+    raw_duration = data.get("format", {}).get("duration")
+    try:
+        if raw_duration is not None:
+            duration = float(raw_duration)
+    except (TypeError, ValueError):
+        duration = None
+    codec = next(
+        (stream.get("codec_name") for stream in data.get("streams", []) if stream.get("codec_type") == "audio"),
+        None,
+    )
+    return MediaInfo(path=path, size_bytes=size_bytes, duration_seconds=duration, audio_codec=codec)
+
+
+class MediaProbeService:
+    def __init__(self, ffprobe_path: Path, timeout_seconds: int = 30) -> None:
+        self.ffprobe_path = ffprobe_path
+        self.timeout_seconds = timeout_seconds
+
+    def probe(self, path: Path) -> MediaInfo:
+        media_path = validate_media_path(path)
+        command = [
+            str(self.ffprobe_path), "-v", "error", "-show_entries",
+            "format=duration:stream=codec_type,codec_name", "-of", "json", str(media_path),
+        ]
+        startup_flags = getattr(subprocess, "CREATE_NO_WINDOW", 0)
+        try:
+            result = subprocess.run(
+                command, capture_output=True, text=True, timeout=self.timeout_seconds,
+                check=False, creationflags=startup_flags,
+            )
+        except subprocess.TimeoutExpired as exc:
+            raise ProbeTimeoutError("FFprobe timed out while inspecting the file.") from exc
+        except OSError as exc:
+            raise ProbeExecutionError("FFprobe could not be started.") from exc
+        if result.returncode != 0:
+            raise ProbeExecutionError("FFprobe could not inspect this media file.")
+        return parse_probe_output(media_path, media_path.stat().st_size, result.stdout)

tests/test_media_probe.py

@@ -0,0 +1,114 @@
+import json
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from voice_transcriptor.services.media_probe import (
+    InvalidMediaPathError,
+    MediaProbeError,
+    MediaProbeService,
+    ProbeExecutionError,
+    ProbeOutputError,
+    ProbeTimeoutError,
+    detect_tools,
+    parse_probe_output,
+    validate_media_path,
+)
+
+
+def test_detects_ffmpeg_and_ffprobe_executables(monkeypatch: pytest.MonkeyPatch) -> None:
+    discovered = {"ffmpeg": r"C:\\tools\\ffmpeg.exe", "ffprobe": r"C:\\tools\\ffprobe.exe"}
+    monkeypatch.setattr("voice_transcriptor.services.media_probe.shutil.which", discovered.get)
+
+    status = detect_tools()
+
+    assert status.ffmpeg_path == Path(r"C:\tools\ffmpeg.exe")
+    assert status.ffprobe_path == Path(r"C:\tools\ffprobe.exe")
+
+
+def test_rejects_directory_path(tmp_path: Path) -> None:
+    with pytest.raises(InvalidMediaPathError, match="not a file"):
+        validate_media_path(tmp_path)
+
+
+def test_parses_audio_only_media(tmp_path: Path) -> None:
+    media = tmp_path / "memo.m4a"
+    payload = json.dumps({
+        "format": {"duration": "4.5"},
+        "streams": [{"codec_type": "audio", "codec_name": "aac"}],
+    })
+
+    info = parse_probe_output(media, 3, payload)
+
+    assert info.duration_seconds == 4.5
+    assert info.audio_codec == "aac"
+
+
+def test_parses_video_with_audio(tmp_path: Path) -> None:
+    media = tmp_path / "meeting.mov"
+    media.write_bytes(b"x" * 12)
+    payload = json.dumps({
+        "format": {"duration": "3661.25"},
+        "streams": [{"codec_type": "video", "codec_name": "h264"}, {"codec_type": "audio", "codec_name": "aac"}],
+    })
+    info = parse_probe_output(media, 12, payload)
+    assert info.duration_seconds == 3661.25
+    assert info.audio_codec == "aac"
+
+
+def test_missing_metadata_is_unknown(tmp_path: Path) -> None:
+    media = tmp_path / "memo.m4a"
+    info = parse_probe_output(media, 0, '{"format": {}, "streams": []}')
+    assert info.duration_seconds is None
+    assert info.audio_codec is None
+
+
+def test_rejects_malformed_probe_json(tmp_path: Path) -> None:
+    with pytest.raises(MediaProbeError, match="invalid data"):
+        parse_probe_output(tmp_path / "bad.mov", 0, "not json")
+
+
+def test_rejects_missing_file(tmp_path: Path) -> None:
+    with pytest.raises(MediaProbeError, match="does not exist"):
+        validate_media_path(tmp_path / "missing.mov")
+
+
+def test_maps_nonzero_ffprobe_result(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    media = tmp_path / "bad.mov"
+    media.write_bytes(b"bad")
+    monkeypatch.setattr(subprocess, "run", lambda *a, **k: subprocess.CompletedProcess(a, 1, "", "invalid input"))
+    with pytest.raises(MediaProbeError, match="could not inspect"):
+        MediaProbeService(Path("ffprobe")).probe(media)
+
+
+def test_maps_ffprobe_timeout(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    media = tmp_path / "slow.mov"
+    media.write_bytes(b"slow")
+    def timeout(*args: object, **kwargs: object) -> None:
+        raise subprocess.TimeoutExpired("ffprobe", 30)
+    monkeypatch.setattr(subprocess, "run", timeout)
+    with pytest.raises(MediaProbeError, match="timed out"):
+        MediaProbeService(Path("ffprobe")).probe(media)
+
+
+def test_maps_probe_failures_to_typed_errors(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    missing = tmp_path / "missing.mov"
+    with pytest.raises(InvalidMediaPathError):
+        validate_media_path(missing)
+
+    with pytest.raises(ProbeOutputError):
+        parse_probe_output(tmp_path / "bad.mov", 0, "not json")
+
+    media = tmp_path / "bad.mov"
+    media.write_bytes(b"bad")
+    monkeypatch.setattr(subprocess, "run", lambda *a, **k: subprocess.CompletedProcess(a, 1, "", "invalid input"))
+    with pytest.raises(ProbeExecutionError):
+        MediaProbeService(Path("ffprobe")).probe(media)
+
+    def timeout(*args: object, **kwargs: object) -> None:
+        raise subprocess.TimeoutExpired("ffprobe", 30)
+
+    monkeypatch.setattr(subprocess, "run", timeout)
+    with pytest.raises(ProbeTimeoutError):
+        MediaProbeService(Path("ffprobe")).probe(media)