francwa
b7979c0f8b
ParsedRelease is now @dataclass(frozen=True). The enrichment passes that used to patch fields in place now produce new instances: - enrich_from_probe(parsed, info, kb) returns a new ParsedRelease via dataclasses.replace (no allocation when no field changed). - inspect_release rebinds 'parsed' after detect_media_type (wrapped in MediaTypeToken — the strict isinstance check now also runs on replace) and after enrich_from_probe. languages becomes a tuple[str, ...] so the VO is properly immutable. Parser pipeline packs languages as a tuple in the assemble dict. Callers updated: inspect_release, testing/recognize_folders_in_downloads.py. Tests updated: 22 enrich_from_probe call sites rebound, language assertions switched to tuple literals, test_release_fixtures normalizes result['languages'] back to list for YAML-fixture comparison. Suite: 1077 passed.
75 lines
2.9 KiB
Python
75 lines
2.9 KiB
Python
"""enrich_from_probe — fill missing ParsedRelease fields from MediaInfo."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
|
|
from alfred.domain.release.ports import ReleaseKnowledge
|
|
from alfred.domain.release.value_objects import ParsedRelease
|
|
from alfred.domain.shared.media import MediaInfo
|
|
|
|
|
|
def enrich_from_probe(
|
|
parsed: ParsedRelease, info: MediaInfo, kb: ReleaseKnowledge
|
|
) -> ParsedRelease:
|
|
"""
|
|
Return a new ParsedRelease with None fields filled from ffprobe MediaInfo.
|
|
|
|
Only overwrites fields that are currently None — token-level values
|
|
from the release name always take priority. ``ParsedRelease`` is
|
|
frozen; this returns a new instance via :func:`dataclasses.replace`.
|
|
|
|
Translation tables (ffprobe codec name → scene token, channel count
|
|
→ layout) live in ``kb.probe_mappings`` (loaded from
|
|
``alfred/knowledge/release/probe_mappings.yaml``). When ffprobe
|
|
reports a value with no mapping entry, the fallback is the uppercase
|
|
raw value so unknown codecs still surface in a predictable form.
|
|
"""
|
|
mappings = kb.probe_mappings
|
|
video_codec_map: dict[str, str] = mappings.get("video_codec", {})
|
|
audio_codec_map: dict[str, str] = mappings.get("audio_codec", {})
|
|
channel_map: dict[int, str] = mappings.get("audio_channels", {})
|
|
|
|
updates: dict[str, object] = {}
|
|
|
|
if parsed.quality is None and info.resolution:
|
|
updates["quality"] = info.resolution
|
|
|
|
if parsed.codec is None and info.video_codec:
|
|
updates["codec"] = video_codec_map.get(
|
|
info.video_codec.lower(), info.video_codec.upper()
|
|
)
|
|
|
|
# bit_depth: ffprobe exposes it via pix_fmt — not in MediaInfo yet, skip.
|
|
|
|
# Audio — use the default track, fallback to first
|
|
default_track = next((t for t in info.audio_tracks if t.is_default), None)
|
|
track = default_track or (info.audio_tracks[0] if info.audio_tracks else None)
|
|
|
|
if track:
|
|
if parsed.audio_codec is None and track.codec:
|
|
updates["audio_codec"] = audio_codec_map.get(
|
|
track.codec.lower(), track.codec.upper()
|
|
)
|
|
|
|
if parsed.audio_channels is None and track.channels:
|
|
updates["audio_channels"] = channel_map.get(
|
|
track.channels, f"{track.channels}ch"
|
|
)
|
|
|
|
# Languages — merge ffprobe languages with token-level ones
|
|
# "und" = undetermined, not useful
|
|
if info.audio_languages:
|
|
existing_upper = {lang.upper() for lang in parsed.languages}
|
|
new_languages = list(parsed.languages)
|
|
for lang in info.audio_languages:
|
|
if lang.lower() != "und" and lang.upper() not in existing_upper:
|
|
new_languages.append(lang)
|
|
existing_upper.add(lang.upper())
|
|
if len(new_languages) != len(parsed.languages):
|
|
updates["languages"] = tuple(new_languages)
|
|
|
|
if not updates:
|
|
return parsed
|
|
return replace(parsed, **updates)
|