fix(import): take a missing title language from the exported tracks

A title with no language now takes the language of an original-language track, else a
video track, so an unlabelled MPD re-parses and -l orig works. A track that no longer
matches its manifest is dropped with a warning instead of failing at download.
This commit is contained in:
imSp4rky
2026-10-21 10:36:03 -06:00
parent 1b20b2bbf9
commit f5bd24b5e9
3 changed files with 75 additions and 14 deletions
+6
View File
@@ -87,6 +87,12 @@ def test_export_without_manifest_url_still_rebuilds_dash_manifest_data(tmp_path:
assert set(tracks.videos[0].data.get("dash", {})) >= {"manifest", "adaptation_set", "representation"}
def test_exported_track_its_manifest_no_longer_lists_is_dropped(tmp_path: Path) -> None:
"""A track that matches no representation would get no manifest data and fail only at download."""
service = make_service(tmp_path, None, {"v-hevc": exported_video("v-hevc", AVC_URL, "HEVC", 3840, 2160)})
assert service.get_tracks(next(iter(service.get_titles()))).videos == []
def test_exported_hevc_on_its_own_mpd_survives_the_reparse(tmp_path: Path) -> None:
"""A ladder split across per-codec MPDs loses every codec but manifest_url's if only URL tracks merge back."""
service = make_service(
+42 -12
View File
@@ -54,19 +54,20 @@ URL_SUBTITLE = {
}
def make_service(tmp_path: Path, language: Optional[str], tracks: Optional[dict[str, Any]] = None) -> ImportService:
export = {
"version": 2,
"service": "EXAMPLE",
"titles": {
"movie-1": {
"meta": {"type": "movie", "id": "movie-1", "name": "Example Movie", "year": 2024, "language": language},
"manifest_url": "https://example.invalid/m.mpd",
"manifest_type": "DASH",
"tracks": tracks or {},
}
},
def make_service(
tmp_path: Path,
language: Optional[str],
tracks: Optional[dict[str, Any]] = None,
manifest_url: Optional[str] = "https://example.invalid/m.mpd",
) -> ImportService:
entry: dict[str, Any] = {
"meta": {"type": "movie", "id": "movie-1", "name": "Example Movie", "year": 2024, "language": language},
"manifest_type": "DASH",
"tracks": tracks or {},
}
if manifest_url:
entry["manifest_url"] = manifest_url
export = {"version": 2, "service": "EXAMPLE", "titles": {"movie-1": entry}}
path = tmp_path / "export.json"
path.write_text(json.dumps(export), encoding="utf8")
return ImportService(click.Context(click.Command("dl")), "EXAMPLE", "movie-1", str(path))
@@ -95,6 +96,35 @@ def test_unlabelled_manifest_without_usable_language_blames_the_service(
assert "may have expired" not in error.value.message
def exported_video(language: str = "ja") -> dict[str, Any]:
return {
"type": "Video",
"id": "v1",
"url": "https://example.invalid/m.mpd",
"language": language,
"descriptor": "DASH",
"codec": "AVC",
"bitrate": 5000000,
"width": 1920,
"height": 1080,
}
@pytest.mark.parametrize("manifest_url", ["https://example.invalid/m.mpd", None], ids=["manifest-url", "per-track-url"])
def test_unlabelled_manifest_takes_the_exported_track_language(tmp_path: Path, manifest_url: Optional[str]) -> None:
"""A title with no language takes it from its exported tracks, for either import branch."""
service = make_service(tmp_path, language=None, tracks={"v1": exported_video()}, manifest_url=manifest_url)
tracks = get_tracks(service, MPD_NO_LANG)
assert [(str(t.language), t.data.get("dash") is not None) for t in tracks.videos] == [("ja", True)]
def test_original_language_track_gives_the_title_language(tmp_path: Path) -> None:
"""The exporter marks the original-language tracks, which beat a video of another language."""
audio = {"type": "Audio", "id": "a1", "language": "de", "is_original_lang": True}
service = make_service(tmp_path, language=None, tracks={"v1": exported_video("en"), "a1": audio})
assert str(next(iter(service.get_titles())).language) == "de"
def test_exported_url_subtitles_replace_the_manifests(tmp_path: Path) -> None:
"""Services drop the manifest's subtitles and side-load their own; importing both duplicates them."""
tracks = get_tracks(make_service(tmp_path, "ja", {"sideloaded-en": URL_SUBTITLE}), MPD_WITH_LANG)
+27 -2
View File
@@ -102,7 +102,8 @@ def resolve_import_manifest_data(
Exports can be without ``manifest_url`` (for example, when the service never set
``title.tracks.manifest_url``) or carry one MPD per adaptation set. ``Track.from_dict``
does not serialise manifest XML, so each exported track's ``url`` is re-fetched and
matched to a locally parsed representation before download.
matched to a locally parsed representation before download. A track that matches none is
dropped with a warning, since without its manifest data it fails only when it downloads.
"""
if manifest_type not in MANIFEST_DATA_KEYS:
return
@@ -143,6 +144,13 @@ def resolve_import_manifest_data(
if matched and matched.data.get(data_key):
track.data.update(matched.data)
unmatched = {str(track.id) for track in pending if not track.data.get(data_key)}
for track_id in unmatched:
log.warning(f"Skipping exported track {track_id}: its {manifest_type} manifest no longer lists it")
tracks.select_video(lambda t: str(t.id) not in unmatched)
tracks.select_audio(lambda t: str(t.id) not in unmatched)
tracks.select_subtitles(lambda t: str(t.id) not in unmatched)
def parse_codec(codec: type[E], parse: Callable[[str], E], value: Any) -> Optional[E]:
"""The codec a row names by enum name (``hevc``) or as ``parse`` reads it, else None."""
@@ -316,7 +324,7 @@ class ImportService:
if self.titles is not None:
return self.titles
titles_list = [
build_title(entry.get("meta", {}), self.service_tag, fallback_id=title_id)
build_title(self.title_meta(entry), self.service_tag, fallback_id=title_id)
for title_id, entry in self.titles_data.items()
]
self.titles = (
@@ -324,6 +332,23 @@ class ImportService:
)
return self.titles
@staticmethod
def title_meta(entry: dict[str, Any]) -> dict[str, Any]:
"""The title meta, with the language of an exported track when the title has none.
A service can set no title language yet label its tracks. The manifest re-parse and
``-l orig`` need a title language, so an original-language track gives it, else a video.
"""
meta = entry.get("meta") or {}
if fallback_language(meta.get("language")):
return meta
source = min(
(d for d in (entry.get("tracks") or {}).values() if fallback_language(d.get("language"))),
key=lambda d: (not d.get("is_original_lang"), d.get("type") != "Video"),
default=None,
)
return {**meta, "language": source["language"]} if source else meta
def get_titles_cached(self, title_id: Optional[str] = None) -> Titles_T:
"""Apply the service's title_map to titles reconstructed from the export sidecar."""
title_map = (config.services.get(self.service_tag) or {}).get("title_map") or {}