francwa
7cd24f3a31
AudioTrack, VideoTrack, SubtitleTrack and MediaInfo are snapshots of a single ffprobe run — model them as proper immutable value objects. - @dataclass(frozen=True) on all four - MediaInfo track collections become tuple[...] instead of list[...] - ffprobe adapter rewritten to build tuples up-front instead of appending/setattr'ing on a constructed instance
122 lines
3.6 KiB
Python
122 lines
3.6 KiB
Python
"""ffprobe — infrastructure adapter for extracting MediaInfo from a video file."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from alfred.domain.shared.media import AudioTrack, MediaInfo, SubtitleTrack, VideoTrack
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_FFPROBE_CMD = [
|
|
"ffprobe",
|
|
"-v",
|
|
"quiet",
|
|
"-print_format",
|
|
"json",
|
|
"-show_streams",
|
|
"-show_format",
|
|
]
|
|
|
|
|
|
def probe(path: Path) -> MediaInfo | None:
|
|
"""
|
|
Run ffprobe on path and return a MediaInfo.
|
|
|
|
Returns None if ffprobe is not available or the file cannot be probed.
|
|
"""
|
|
try:
|
|
result = subprocess.run(
|
|
[*_FFPROBE_CMD, str(path)],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
check=False,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
logger.warning("ffprobe timed out on %s", path)
|
|
return None
|
|
|
|
if result.returncode != 0:
|
|
logger.warning("ffprobe failed on %s: %s", path, result.stderr.strip())
|
|
return None
|
|
|
|
try:
|
|
data = json.loads(result.stdout)
|
|
except json.JSONDecodeError:
|
|
logger.warning("ffprobe returned invalid JSON for %s", path)
|
|
return None
|
|
|
|
return _parse(data)
|
|
|
|
|
|
def _parse(data: dict) -> MediaInfo:
|
|
streams = data.get("streams", [])
|
|
fmt = data.get("format", {})
|
|
|
|
# File-level duration/bitrate (ffprobe ``format`` block — independent of streams)
|
|
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,
|
|
)
|
|
)
|
|
|
|
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,
|
|
)
|