You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
103 lines
3.6 KiB
103 lines
3.6 KiB
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: Any = json.loads(payload) |
|
except (json.JSONDecodeError, TypeError) as exc: |
|
raise ProbeOutputError("FFprobe returned invalid data.") from exc |
|
if not isinstance(data, dict): |
|
raise ProbeOutputError("FFprobe returned invalid data.") |
|
format_data = data.get("format", {}) |
|
streams = data.get("streams", []) |
|
if ( |
|
not isinstance(format_data, dict) |
|
or not isinstance(streams, list) |
|
or not all(isinstance(stream, dict) for stream in streams) |
|
): |
|
raise ProbeOutputError("FFprobe returned invalid data.") |
|
duration: float | None = None |
|
raw_duration = format_data.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 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)
|
|
|