From 80088fe4f614bc634e246c94a92620eff7421e55 Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Wed, 23 Sep 2026 15:04:22 -0600 Subject: [PATCH] feat(import): build tracks from direct-file export rows Imports unidl json_manifest exports made of whole files. --- docs/guide/downloading.md | 7 +- tests/core/test_import_tracks.py | 90 +++++++++++++++++++++ unshackle/core/import_service.py | 131 ++++++++++++++++++++++++------- 3 files changed, 196 insertions(+), 32 deletions(-) diff --git a/docs/guide/downloading.md b/docs/guide/downloading.md index ea014cf4..c203db91 100644 --- a/docs/guide/downloading.md +++ b/docs/guide/downloading.md @@ -790,8 +790,11 @@ unshackle dl --skip-dl --export EXAMPLE 81234567 The export holds a content key for each track your flags selected and no others. To make a file another person can use, select generously: for example `-a AAC,EC3` and every language -you want them to have. An import can only pick tracks whose key is in the file. A subtitle -that another tool exported as a direct URL is added next to the manifest's own. +you want them to have. An import can only pick tracks whose key is in the file. A track +that another tool exported as a direct file URL is added next to the manifest's own tracks. +An export with no manifest, for example a unidl export of whole media files, gets all of its +tracks from those URLs. If such an export does not give the title a language, `orig` has no +value: select the languages with `-l` and `-vl`. An import uses only the content keys in the file and in your vaults. It never gets a licence from a CDM. When a track needs a content key that neither has, the import stops and names the diff --git a/tests/core/test_import_tracks.py b/tests/core/test_import_tracks.py index c815bf4e..2fbc0b93 100644 --- a/tests/core/test_import_tracks.py +++ b/tests/core/test_import_tracks.py @@ -13,6 +13,7 @@ import json from pathlib import Path from typing import Any, Optional from unittest.mock import patch +from uuid import UUID import click import pytest @@ -20,6 +21,7 @@ import requests from unshackle.core.import_service import ImportService from unshackle.core.manifests import DASH +from unshackle.core.tracks import Audio, Subtitle, Video MPD_TEMPLATE = """ @@ -177,3 +179,91 @@ def test_a_subtitle_side_load_from_another_tool_is_added(tmp_path: Path) -> None ] assert [t.id for t in tracks.subtitles[1:]] == ["sub-en", "sub-x"] assert tracks.subtitles[1].sdh and tracks.subtitles[1].codec is not None + + +VIDEO_KID, AUDIO_KID = "0a" * 16, "0b" * 16 + + +def file_export(tmp_path: Path, rows: list[dict[str, Any]]) -> ImportService: + """An export with no manifest: its ``tracks[]`` rows with a url are all of its media.""" + export = { + "kind": "mediaexport", + "version": 1, + "service": {"tag": "EXAMPLE"}, + "titles": [ + { + "id": "movie-1", + "kind": "movie", + "title": "Example Movie", + "language": "fr", + "drm": [{"system": "playready", "wrm_header": ""}], + "keys": {VIDEO_KID: "a1" * 16, AUDIO_KID: "b2" * 16}, + "tracks": rows, + } + ], + } + 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)) + + +FILE_ROWS: list[dict[str, Any]] = [ + { + "id": "v", + "type": "video", + "url": "https://example.invalid/v.mp4", + "codec": "dvh1.05.07", + "bitrate": 15000000, + "width": 3840, + "height": 2160, + "fps": 25.0, + "range": "dv", + "headers": {"User-Agent": "exporter-ua"}, + "kids": [VIDEO_KID], + }, + { + "id": "a", + "type": "audio", + "url": "https://example.invalid/a.mp4", + "codec": "ec-3", + "language": "en", + "bitrate": 578523, + "channels": "6", + "atmos": True, + "kids": [AUDIO_KID], + }, + {"id": "s", "type": "subtitle", "url": "https://example.invalid/s.mp4", "codec": "stpp", "language": "fr"}, +] + + +def test_a_title_with_no_manifest_gets_a_track_for_each_file_row(tmp_path: Path) -> None: + service = file_export(tmp_path, FILE_ROWS) + tracks = service.get_tracks(next(iter(service.get_titles()))) + video, audio, subtitle = tracks.videos[0], tracks.audio[0], tracks.subtitles[0] + assert [t.descriptor.name for t in (video, audio, subtitle)] == ["URL", "URL", "URL"] + assert (video.codec, video.range, video.width, video.height, video.bitrate, str(video.language)) == ( + Video.Codec.HEVC, + Video.Range.DV, + 3840, + 2160, + 15000000, + "fr", + ) + assert (audio.codec, audio.atmos, str(audio.language), audio.bitrate) == (Audio.Codec.EC3, True, "en", 578523) + assert subtitle.codec == Subtitle.Codec.fTTML + assert service.session.headers["User-Agent"] == "exporter-ua" + + +def test_each_file_track_gets_only_the_keys_for_its_kids(tmp_path: Path) -> None: + service = file_export(tmp_path, FILE_ROWS) + tracks = service.get_tracks(next(iter(service.get_titles()))) + assert tracks.videos[0].drm[0].content_keys == {UUID(hex=VIDEO_KID): "a1" * 16} + assert tracks.audio[0].drm[0].content_keys == {UUID(hex=AUDIO_KID): "b2" * 16} + assert not tracks.subtitles[0].drm + + +def test_a_file_row_can_name_its_codec_and_range_by_name(tmp_path: Path) -> None: + row = {"id": "v", "type": "video", "url": "https://example.invalid/v.mp4", "codec": "hevc", "range": "hdr10p"} + service = file_export(tmp_path, [row]) + video = service.get_tracks(next(iter(service.get_titles()))).videos[0] + assert (video.codec, video.range) == (Video.Codec.HEVC, Video.Range.HDR10P) diff --git a/unshackle/core/import_service.py b/unshackle/core/import_service.py index 2a4bebe8..1b1aefc4 100644 --- a/unshackle/core/import_service.py +++ b/unshackle/core/import_service.py @@ -1,9 +1,10 @@ from __future__ import annotations import logging +from enum import Enum from http.cookiejar import CookieJar from pathlib import Path -from typing import Any, Optional, Union +from typing import Any, Callable, Optional, TypeVar, Union from uuid import UUID import click @@ -23,6 +24,7 @@ from unshackle.core.tracks.attachment import Attachment from unshackle.core.tracks.track import Track log = logging.getLogger("import") +E = TypeVar("E", bound=Enum) PARSERS = {"DASH": DASH, "HLS": HLS, "ISM": ISM} MANIFEST_DATA_KEYS = {"DASH": "dash", "ISM": "ism"} @@ -95,6 +97,37 @@ def resolve_import_manifest_data( track.data.update(matched.data) +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.""" + if not value: + return None + if str(value).upper() in codec.__members__: + return codec[str(value).upper()] + try: + return parse(str(value)) + except ValueError: + return None + + +def video_range(value: Any) -> Optional[Video.Range]: + """The range a row names, by enum name (``hdr10p``) or value (``HDR10+``), else None for SDR.""" + name = str(value or "").upper() + if name in Video.Range.__members__: + return Video.Range[name] + try: + return Video.Range(name) + except ValueError: + return None + + +def int_or_none(value: Any) -> Optional[int]: + """The integer a row field holds, else None.""" + try: + return int(value) + except (TypeError, ValueError, OverflowError): + return None + + class ImportService: """Reconstructs a download from an export JSON. @@ -142,8 +175,13 @@ class ImportService: self._server_cdm = True self._server_cdm_type = "widevine" - first = self.doc.titles[0].primary if self.doc.titles else None - headers = first.headers if first else {} + first = self.doc.titles[0] + # a title with no manifest keeps its request headers on its file rows + headers = ( + first.primary.headers + if first.primary + else next((r["headers"] for r in first.tracks if r.get("url") and r.get("headers")), {}) + ) self.session = self.build_session(ctx, self.region, headers) @staticmethod @@ -254,6 +292,7 @@ class ImportService: set in favour of the exported one. HLS/URL: rebuild from the stored per-track dicts, since the variant is re-fetched from track.url at download time and a master playlist can hand out a new token on every fetch. + An export from another tool with no manifest gets all its tracks from its file rows. """ title_id = str(title.id) if title_id in self.tracks_by_title: @@ -312,8 +351,8 @@ class ImportService: track_dicts = list(tracks_map.values()) # a side-load from another tool is only a tracks[] row with a url, not a track dict - for sub in self.sideloaded_subtitles(title_id, skip=set(tracks_map)): - tracks.add(sub, warn_only=True) + for sideloaded in self.sideloaded_tracks(title_id, skip=set(tracks_map)): + tracks.add(sideloaded, warn_only=True) for track_dict in track_dicts: try: @@ -354,37 +393,69 @@ class ImportService: self.tracks_by_title[title_id] = tracks return tracks - def sideloaded_subtitles(self, title_id: str, skip: set[str]) -> list[Subtitle]: - """Subtitle tracks for the ``tracks[]`` rows with a ``url``, except the ids in ``skip``. + def sideloaded_tracks(self, title_id: str, skip: set[str]) -> list[AnyTrack]: + """Tracks for the ``tracks[]`` rows with a ``url``, except the ids in ``skip``. - The row names the file format as its codec, or nothing, in which case the download - finds it with MediaInfo. A row with no language cannot be muxed, so it is skipped. + Each row is one complete file. A row names its codec by name or as an RFC 6381 codecs + string, or not at all. Then the download finds the codec with MediaInfo. A subtitle row + with no language cannot be muxed, so it is skipped. A video or audio row with no + language takes the title language, else ``und``. A video or audio track gets the keys + for the KIDs of its row. """ entry = self.doc.get(title_id) - out = [] + out: list[AnyTrack] = [] for row in entry.tracks if entry else []: - if row.get("type") != "subtitle" or not row.get("url") or str(row.get("id")) in skip: + kind = row.get("type") + if kind not in ("video", "audio", "subtitle") or not row.get("url") or str(row.get("id")) in skip: continue - language = row.get("language") or (entry.language if entry else "") - if not language: - self.log.warning(f"Skipping exported subtitle {row.get('id')!r}: it has no language") - continue - codec = None - try: - codec = Subtitle.Codec.from_mime(str(row.get("codec") or "")) - except ValueError: - pass - out.append( - Subtitle( - id_=str(row.get("id") or ""), - url=str(row["url"]), - language=str(language), - codec=codec, - sdh=bool(row.get("sdh")), - forced=bool(row.get("forced")), - cc=bool(row.get("cc")), + language = str(row.get("language") or (entry.language if entry else "") or "") + track_id, url = str(row.get("id") or ""), str(row["url"]) + if kind == "subtitle": + if not language: + self.log.warning(f"Skipping exported subtitle {row.get('id')!r}: it has no language") + continue + out.append( + Subtitle( + id_=track_id, + url=url, + language=language, + codec=parse_codec(Subtitle.Codec, Subtitle.Codec.from_mime, row.get("codec")), + sdh=bool(row.get("sdh")), + forced=bool(row.get("forced")), + cc=bool(row.get("cc")), + ) ) - ) + continue + track: Union[Video, Audio] + try: + if kind == "video": + track = Video( + id_=track_id, + url=url, + language=language or "und", + codec=parse_codec(Video.Codec, Video.Codec.from_codecs, row.get("codec")), + range_=video_range(row.get("range")), + bitrate=int_or_none(row.get("bitrate")), + width=int_or_none(row.get("width")), + height=int_or_none(row.get("height")), + fps=row.get("fps") or None, + ) + else: + track = Audio( + id_=track_id, + url=url, + language=language or "und", + codec=parse_codec(Audio.Codec, Audio.Codec.from_codecs, row.get("codec")), + bitrate=int_or_none(row.get("bitrate")), + channels=row.get("channels") or None, + descriptive=bool(row.get("descriptive")), + extra={"atmos": True} if row.get("atmos") else None, + ) + except (TypeError, ValueError, NotImplementedError) as e: + self.log.warning(f"Skipping exported track {row.get('id')!r}: {e}") + continue + track.drm = self.rebuild_drm({"id": track_id}, title_id) + out.append(track) return out def key_pool(self, title_id: Optional[str] = None) -> dict[UUID, str]: