diff --git a/src/voice_transcriptor/services/__init__.py b/src/voice_transcriptor/services/__init__.py new file mode 100644 index 0000000..c960e15 --- /dev/null +++ b/src/voice_transcriptor/services/__init__.py @@ -0,0 +1,2 @@ +"""Application services independent from the GUI.""" + diff --git a/src/voice_transcriptor/services/media_probe.py b/src/voice_transcriptor/services/media_probe.py new file mode 100644 index 0000000..b952062 --- /dev/null +++ b/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) diff --git a/tests/test_media_probe.py b/tests/test_media_probe.py new file mode 100644 index 0000000..a4144f2 --- /dev/null +++ b/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)