From 746385101b511e9cf722f138dd17364231686e64 Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Sun, 20 Sep 2026 00:07:04 -0600 Subject: [PATCH 1/2] feat(export): write mediaexport files and read legacy exports through the package --export writes the mediaexport format with DRM, content keys and manifests at title level, and unshackle's own track dicts under x-unshackle. unshackle import reads mediaexport, unidl v1 and unshackle v2 through the shared reader, applies the export's request headers to the import session, and rebuilds DRM from the title-level pool. mediaexport is a git dependency on unshackle-dl/mediaexport. --- docs/guide/downloading.md | 18 +++- pyproject.toml | 2 + tests/core/test_export.py | 94 ++++++++++++++----- tests/core/test_import_hls_ism.py | 5 +- unshackle/commands/dl.py | 149 ++++++++++++++++++++---------- unshackle/core/api/routes.py | 2 +- unshackle/core/import_service.py | 130 +++++++++++++++++++------- uv.lock | 7 ++ 8 files changed, 297 insertions(+), 110 deletions(-) diff --git a/docs/guide/downloading.md b/docs/guide/downloading.md index 744eb9f0..71a719da 100644 --- a/docs/guide/downloading.md +++ b/docs/guide/downloading.md @@ -771,14 +771,26 @@ a content key when the vault misses. You can force one side or the other: unshackle dl --skip-dl EXAMPLE 81234567 ``` -`--export` writes a JSON file, into the configured exports directory, containing track -info and the acquired content keys for each title. This is the format consumed by -`unshackle import` to reconstruct a download later. +`--export` writes a `mediaexport` JSON file into the configured exports directory. It holds +the manifest, the DRM init data, the content keys and the title metadata for each title. +`unshackle import` reads it to reconstruct a download later. The importer also reads the +older unshackle `version: 2` files and unidl's own export files. ```shell title="Export track info and keys" 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. + +!!! warning "The export file is a secret" + It holds the content keys and usually a signed manifest URL. Hand it over on purpose, + not by accident. + +The [mediaexport](https://github.com/unshackle-dl/mediaexport) package specifies the +`mediaexport` format and is the reader and writer unshackle uses. + !!! note "Region is recorded only with a proxy" When you use `--proxy`, the export records the region so an import can reproduce the correct geofence. Without a proxy, the export stores no region. diff --git a/pyproject.toml b/pyproject.toml index d144c355..000c5844 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -70,6 +70,7 @@ dependencies = [ "mutagen>=1.47.0,<2", "defusedxml>=0.7.1", "rich-click>=1.9.8", + "mediaexport", ] [project.urls] @@ -172,3 +173,4 @@ skips = [ [tool.uv.sources] unshackle = { workspace = true } +mediaexport = { git = "https://github.com/unshackle-dl/mediaexport", branch = "main" } diff --git a/tests/core/test_export.py b/tests/core/test_export.py index 4522e627..ef8939bb 100644 --- a/tests/core/test_export.py +++ b/tests/core/test_export.py @@ -1,4 +1,4 @@ -"""Tests for ``dl.write_export``, the ``--export`` JSON sidecar. +"""Tests for ``dl.write_export``, the ``--export`` mediaexport sidecar. Regression: DRM-free tracks never pass through ``prepare_drm``, so ``write_export`` must accept ``drm=None`` (and DRM systems without ``to_dict``/``content_keys`` such @@ -83,6 +83,10 @@ def read_export(path: Path) -> dict: return json.loads(path.read_text(encoding="utf8")) +def entry(path: Path) -> dict: + return read_export(path)["titles"][0] + + def test_drm_free_track_exports(tmp_path: Path) -> None: """The reported bug: DRM-free downloads produced no usable export.""" export = tmp_path / "export.json" @@ -92,13 +96,14 @@ def test_drm_free_track_exports(tmp_path: Path) -> None: make_dl().write_export(export, title, video) doc = read_export(export) - assert doc["version"] == 2 - assert doc["service"] == "EXAMPLE" - tinfo = doc["titles"]["movie-1"] - assert tinfo["meta"]["name"] == "Example Movie" - assert set(tinfo["tracks"]) == {"v1", "a1", "s1"} - assert "keys" not in tinfo["tracks"]["v1"] - assert "drm" not in tinfo["tracks"]["v1"] + assert doc["kind"] == "mediaexport" and doc["version"] == 1 + assert doc["service"]["tag"] == "EXAMPLE" + tinfo = doc["titles"][0] + assert tinfo["title"] == "Example Movie" and tinfo["kind"] == "movie" + assert {row["id"] for row in tinfo["tracks"]} == {"v1", "a1", "s1"} + assert tinfo["x-unshackle"]["meta"]["name"] == "Example Movie" + assert set(tinfo["x-unshackle"]["tracks"]) == {"v1"} + assert "keys" not in tinfo and "drm" not in tinfo def test_clearkey_drm_exports_track_without_keys(tmp_path: Path) -> None: @@ -109,10 +114,9 @@ def test_clearkey_drm_exports_track_without_keys(tmp_path: Path) -> None: make_dl().write_export(export, title, video, ClearKey(key="bb" * 16)) - doc = read_export(export) - track = doc["titles"]["movie-1"]["tracks"]["v1"] - assert "keys" not in track - assert "drm" not in track + tinfo = entry(export) + assert "keys" not in tinfo and "drm" not in tinfo + assert "v1" in tinfo["x-unshackle"]["tracks"] def test_post_download_write_keeps_licensed_keys(tmp_path: Path) -> None: @@ -125,9 +129,10 @@ def test_post_download_write_keeps_licensed_keys(tmp_path: Path) -> None: runner.write_export(export, title, video, StubDRM()) # prepare_drm runner.write_export(export, title, video) # post-download hook - track = read_export(export)["titles"]["movie-1"]["tracks"]["v1"] - assert track["drm"] == [{"system": "Widevine", "pssh_b64": "AAAA"}] - assert track["keys"] == {KID.hex: "aa" * 16} + tinfo = entry(export) + assert tinfo["drm"] == [{"system": "widevine", "pssh": "AAAA"}] + assert tinfo["keys"] == {KID.hex: "aa" * 16} + assert tinfo["x-unshackle"]["drm"] == [{"system": "Widevine", "pssh_b64": "AAAA"}] def test_drm_free_export_roundtrips_through_import_service(tmp_path: Path) -> None: @@ -174,11 +179,13 @@ def test_clearkey_cenc_exports_drm_and_keys(tmp_path: Path) -> None: make_dl().write_export(export, title, video, drm) - track = read_export(export)["titles"]["movie-1"]["tracks"]["v1"] - assert track["drm"] == [{"system": "ClearKeyCENC", "kids": [KID.hex], "laurl": "https://license.example.test/ck"}] - assert track["keys"] == {KID.hex: "cc" * 16} + tinfo = entry(export) + assert tinfo["drm"] == [{"system": "clearkey"}] + assert tinfo["keys"] == {KID.hex: "cc" * 16} + own = tinfo["x-unshackle"]["drm"] + assert own == [{"system": "ClearKeyCENC", "kids": [KID.hex], "laurl": "https://license.example.test/ck"}] - rebuilt = drm_from_dict({**track["drm"][0], "content_keys": track["keys"]}) + rebuilt = drm_from_dict({**own[0], "content_keys": tinfo["keys"]}) assert isinstance(rebuilt, ClearKeyCENC) assert rebuilt.content_keys == {KID: "cc" * 16} @@ -193,6 +200,49 @@ def test_keyless_content_keys_writes_no_keys_entry(tmp_path: Path) -> None: make_dl().write_export(export, title, video, drm) - track = read_export(export)["titles"]["movie-1"]["tracks"]["v1"] - assert track["drm"] == [{"system": "Widevine", "pssh_b64": "AAAA"}] - assert "keys" not in track + tinfo = entry(export) + assert tinfo["drm"] == [{"system": "widevine", "pssh": "AAAA"}] + assert "keys" not in tinfo + + +def test_unidl_export_imports_without_track_dicts(tmp_path: Path) -> None: + """A file from another tool has no x-unshackle block: titles, keys and chapters still rebuild.""" + export = tmp_path / "unidl.json" + export.write_text( + json.dumps( + { + "kind": "unidl-export", + "version": 1, + "service": "example", + "titles": [ + { + "save_name": "Show.S01E02", + "title": { + "id": "ep-2", + "kind": "episode", + "name": "Show", + "episode_name": "Two", + "season": 1, + "episode": 2, + "year": "2024", + }, + "keys": [f"{KID.hex}:{'dd' * 16}"], + "manifest_url": "https://example.test/m.mpd", + "headers": {"User-Agent": "unidl-ua"}, + "drm": {"system": "widevine", "pssh": "AAAA"}, + "chapters": [{"start_ms": 5000, "title": "Recap"}], + } + ], + } + ), + encoding="utf8", + ) + svc = ImportService(SimpleNamespace(parent=None, params={}), "EXAMPLE", "ep-2", str(export)) + + ep = next(iter(svc.get_titles())) + assert (ep.title, ep.season, ep.number) == ("Show", 1, 2) + assert svc.session.headers["User-Agent"] == "unidl-ua" + assert svc.key_pool() == {KID: "dd" * 16} + assert svc.exported_drm_system() == "widevine" + assert [c.name for c in svc.get_chapters(ep)] == [None, "Recap"] + assert svc.titles_data["ep-2"]["manifest_type"] == "DASH" diff --git a/tests/core/test_import_hls_ism.py b/tests/core/test_import_hls_ism.py index f99bbf18..752ec527 100644 --- a/tests/core/test_import_hls_ism.py +++ b/tests/core/test_import_hls_ism.py @@ -121,11 +121,10 @@ def test_encrypted_ism_rung_gets_the_exported_keys(tmp_path: Path, monkeypatch: monkeypatch.setattr(ISM, "from_url", lambda url, session=None, **kw: ISM.from_text(ISM_ENCRYPTED, url)) parsed = ISM.from_text(ISM_ENCRYPTED, ISM_URL).to_tracks(language="en") - # export the licensed rung only, as a finished download would - licensed = max(parsed.videos, key=lambda t: t.bitrate or 0) export = export_tracks(tmp_path, parsed, ISM_URL) doc = json.loads(export.read_text(encoding="utf8")) - doc["titles"]["movie-1"]["tracks"][licensed.id]["keys"] = {"0" * 32: "ab" * 16} + doc["titles"][0]["keys"] = {"0" * 32: "ab" * 16} + doc["titles"][0]["drm"] = [{"system": "playready"}] export.write_text(json.dumps(doc), encoding="utf8") svc, tracks = import_tracks(export) diff --git a/unshackle/commands/dl.py b/unshackle/commands/dl.py index ef5fd6c9..d5e4b6b7 100644 --- a/unshackle/commands/dl.py +++ b/unshackle/commands/dl.py @@ -1,7 +1,6 @@ from __future__ import annotations import html -import json import logging import math import os @@ -27,6 +26,7 @@ from typing import Any, Callable, Collection, Optional, Sequence, TypedDict, Uni from uuid import UUID import click +import mediaexport import yaml from click.core import ParameterSource from langcodes import Language, tag_is_valid @@ -40,7 +40,7 @@ from rich.table import Table from rich.text import Text from rich.tree import Tree -from unshackle.core import binaries, providers +from unshackle.core import __version__, binaries, providers from unshackle.core.cdm import DecryptLabsRemoteCDM from unshackle.core.cdm.detect import cdm_type_stub, is_playready_cdm, is_widevine_cdm from unshackle.core.config import config, resolve_cdm_name, resolve_decryption @@ -3914,63 +3914,118 @@ class dl: return meta def write_export(self, export: Path, title: Title_T, track: AnyTrack, drm: Any = None) -> None: - """Write a shareable v2 export usable by ``unshackle import``. + """Write a shareable mediaexport file usable by ``unshackle import`` and unidl. - Carries no HTTP session, cookies, or dl-flags. The export records the region (country - code) only when the export used ``--proxy``, as an import geofence. Each track records - only the licensed DRM system. Content keys live once under the track's ``keys``. ``drm`` may be None - (DRM-free track) or a DRM system without ``to_dict``/``content_keys`` (e.g. ClearKey) - - the export still records the track, manifest, chapter and attachment info. + Carries no HTTP session or cookies. The export records the region (country code) + only when the export used ``--proxy``, as an import geofence. DRM init data and + content keys live once per title. ``drm`` may be None (DRM-free track) or a DRM + system without ``to_dict``/``content_keys`` (e.g. ClearKey) - the export still + records the track, manifest, chapter and attachment info. unshackle's full track + dicts and title meta go under ``x-unshackle`` for its own importer. """ with self.EXPORT_LOCK: - doc: dict[str, Any] = {} - if export.is_file(): - doc = json.loads(export.read_text(encoding="utf8")) or {} + doc = mediaexport.read(export) if export.is_file() else mediaexport.Document(service_tag=self.service) + doc.generator.setdefault("app", "unshackle") + doc.generator.setdefault("version", __version__) + if not doc.region and getattr(self, "proxy_requested", False): + doc.region = getattr(getattr(self, "export_service", None), "current_region", None) or "" - doc.setdefault("version", 2) - doc.setdefault("service", self.service) - if "region" not in doc and getattr(self, "proxy_requested", False): - region = getattr(getattr(self, "export_service", None), "current_region", None) - if region: - doc["region"] = region + entry = doc.get(str(title.id)) + if entry is None: + entry = self.export_entry(title) + doc.add(entry) - titles = doc.setdefault("titles", {}) - tinfo = titles.setdefault(str(title.id), {}) - tinfo.setdefault("meta", self.title_to_meta(title)) + tracks_map = entry.ext("unshackle").setdefault("tracks", {}) + tracks_map.setdefault(str(track.id), track.to_dict()) + for row in entry.tracks: + if row.get("id") == str(track.id): + row["selected"] = True - if title.tracks.manifest_url: - tinfo.setdefault("manifest_url", title.tracks.manifest_url) - - all_tracks = [*title.tracks.videos, *title.tracks.audio, *title.tracks.subtitles] - if "manifest_type" not in tinfo: - tinfo["manifest_type"] = next( - (t.descriptor.name for t in all_tracks if t.descriptor != Video.Descriptor.URL), None - ) - - tracks_map = tinfo.setdefault("tracks", {}) - if not tracks_map: - for t in all_tracks: - tracks_map[str(t.id)] = t.to_dict() - - track_data = tracks_map.setdefault(str(track.id), track.to_dict()) if drm is not None: if hasattr(drm, "to_dict"): - track_data["drm"] = [drm.to_dict()] + d = drm.to_dict() + own = entry.ext("unshackle").setdefault("drm", []) + if d not in own: + own.append(d) + system = str(d.get("system", "")).lower().replace("cenc", "") + pssh = str(d.get("pssh_b64") or "") + if system and all(x.pssh != pssh or x.system != system for x in entry.drm): + entry.drm.append(mediaexport.Drm(system, pssh)) content_keys = getattr(drm, "content_keys", None) or {} - if content_keys: - keys = track_data.setdefault("keys", {}) - for kid, key in content_keys.items(): - keys[kid.hex] = key + for kid, key in content_keys.items(): + entry.keys[kid.hex] = key - if "chapters" not in tinfo: - tinfo["chapters"] = [ - {"timestamp": chapter.timestamp, "name": chapter.name} for chapter in (title.tracks.chapters or []) - ] + mediaexport.write(export, doc) - if "attachments" not in tinfo: - tinfo["attachments"] = [a.to_dict() for a in (title.tracks.attachments or []) if a.url] + def export_entry(self, title: Title_T) -> mediaexport.Entry: + """A mediaexport entry for ``title``: manifests, chapters, a small track list, no keys yet.""" + meta = self.title_to_meta(title) + all_tracks = [*title.tracks.videos, *title.tracks.audio, *title.tracks.subtitles] + primary_url = title.tracks.manifest_url or next( + (str(t.url) for t in all_tracks if t.descriptor != Video.Descriptor.URL), "" + ) + manifest_type = next( + (t.descriptor.name.lower() for t in all_tracks if t.descriptor != Video.Descriptor.URL), "" + ) + session = getattr(getattr(self, "export_service", None), "session", None) + headers = { + k: v + for k, v in (getattr(session, "headers", None) or {}).items() + if k.lower() not in ("cookie", "authorization") + } + manifests = [mediaexport.Manifest(primary_url, manifest_type, headers, role="primary")] if primary_url else [] + seen = {primary_url.split("?")[0]} + for t in all_tracks: + url = str(t.url) + if t.descriptor in (Video.Descriptor.DASH, Video.Descriptor.ISM) and url.split("?")[0] not in seen: + seen.add(url.split("?")[0]) + manifests.append(mediaexport.Manifest(url, t.descriptor.name.lower(), role="extra")) - export.write_text(json.dumps(doc, indent=4, ensure_ascii=False), encoding="utf8") + rows = [] + for t in all_tracks: + row: dict[str, Any] = { + "id": str(t.id), + "type": t.__class__.__name__.lower(), + "codec": codec.name.lower() if (codec := getattr(t, "codec", None)) else "", + "language": str(t.language) if t.language else "", + "bitrate": int(bitrate) if (bitrate := getattr(t, "bitrate", None)) else None, + } + if isinstance(t, Video): + row.update(width=t.width, height=t.height, range=t.range.name.lower() if t.range else "") + elif isinstance(t, Audio): + row.update(channels=str(t.channels) if t.channels else "", atmos=bool(t.joc), descriptive=t.descriptive) + elif isinstance(t, Subtitle): + row.update(sdh=t.sdh, forced=t.forced, cc=t.cc) + if t.descriptor == Video.Descriptor.URL: + row["url"] = str(t.url) + rows.append({k: v for k, v in row.items() if v not in (None, "")}) + + return mediaexport.Entry( + id=str(title.id), + kind=meta.get("type", "movie"), + title=meta.get("name") or "", + series=meta.get("series_title") or "", + season=meta.get("season"), + episode=meta.get("number"), + year=meta.get("year"), + language=meta.get("language") or "", + artist=meta.get("artist") or "", + album=meta.get("album") or "", + track_number=meta.get("track"), + manifests=manifests, + chapters=[ + {"start_ms": mediaexport.ts_ms(c.timestamp), "title": c.name or ""} + for c in (title.tracks.chapters or []) + ], + tracks=rows, + extensions={ + "x-unshackle": { + "meta": meta, + "tracks": {}, + "attachments": [a.to_dict() for a in (title.tracks.attachments or []) if a.url], + } + }, + ) def decrypt_verified( self, diff --git a/unshackle/core/api/routes.py b/unshackle/core/api/routes.py index e77d8ce4..b2c812d2 100644 --- a/unshackle/core/api/routes.py +++ b/unshackle/core/api/routes.py @@ -794,7 +794,7 @@ async def download(request: web.Request) -> web.Response: description: Skip downloading, only retrieve decryption keys (default - false) export: type: boolean - description: Export manifest, track URLs, keys, and subtitles to JSON in the exports directory (default - false) + description: Export manifest, DRM init data, keys, and track info to a JSON file in the exports directory (default - false) cdm_only: type: boolean description: Only use CDM for content key retrieval (true) or only vaults (false) (default - None) diff --git a/unshackle/core/import_service.py b/unshackle/core/import_service.py index 9494fb62..dcce3129 100644 --- a/unshackle/core/import_service.py +++ b/unshackle/core/import_service.py @@ -1,6 +1,5 @@ from __future__ import annotations -import json import logging from http.cookiejar import CookieJar from pathlib import Path @@ -8,6 +7,7 @@ from typing import Any, Optional, Union from uuid import UUID import click +import mediaexport import requests from langcodes import tag_is_valid @@ -124,33 +124,72 @@ class ImportService: if not export_path.is_file(): raise click.ClickException(f"Export file not found: {export_path}") - self.data: dict[str, Any] = json.loads(export_path.read_text(encoding="utf8")) - version = self.data.get("version") - if version != 2: - raise click.ClickException( - f"Unsupported export version {version!r}. Re-create the export with a current build." - ) + try: + self.doc = mediaexport.read(export_path) + except mediaexport.ExportError as e: + raise click.ClickException(f"Cannot import {export_path.name}: {e}") - self.titles_data: dict[str, Any] = self.data.get("titles", {}) - self.region: Optional[str] = self.data.get("region") + self.titles_data: dict[str, Any] = {e.id: self.legacy_entry(e) for e in self.doc.titles} + self.region: Optional[str] = self.doc.region or None self.titles: Optional[Titles_T] = None self.tracks_by_title: dict[str, Tracks] = {} self._server_cdm = True self._server_cdm_type = "widevine" - self.session = self.build_session(ctx, self.region) + first = self.doc.titles[0].primary if self.doc.titles else None + headers = first.headers if first else {} + self.session = self.build_session(ctx, self.region, headers) @staticmethod - def build_session(ctx: click.Context, region: Optional[str] = None) -> requests.Session: + def legacy_entry(entry: mediaexport.Entry) -> dict[str, Any]: + """The per-title dict the rest of this class reads, from a mediaexport entry. + + unshackle's own exports carry the full title meta and track dicts under + ``x-unshackle``; a file from another tool has only the shared fields, so the meta is + rebuilt from those and the track map stays empty (the manifest is parsed instead). + """ + x = entry.extensions.get("x-unshackle") or {} + meta = x.get("meta") or { + "type": entry.kind, + "id": entry.id, + "name": entry.title, + "series_title": entry.series, + "season": entry.season, + "number": entry.episode, + "year": entry.year, + "language": entry.language or None, + "artist": entry.artist, + "album": entry.album, + "track": entry.track_number, + } + primary = entry.primary + manifest_type = x.get("manifest_type") or ( + (primary.type or mediaexport.guess_type(primary.url)).upper() if primary else "" + ) + return { + "meta": meta, + "manifest_url": primary.url if primary else None, + "manifest_type": manifest_type or None, + "tracks": x.get("tracks") or {}, + "chapters": entry.chapters, + "attachments": x.get("attachments") or [], + } + + @staticmethod + def build_session( + ctx: click.Context, region: Optional[str] = None, headers: Optional[dict[str, str]] = None + ) -> requests.Session: """HTTP session for re-fetching the manifest. - Honours the importer's ``--proxy``. Without one, it falls back to the export region - as a geofence. An explicit proxy that unshackle cannot find raises an error. A region - fallback gives a warning. + Exported request headers go over the config defaults, since a CDN can gate the + manifest on the User-Agent the exporter used. Honours the importer's ``--proxy``. + Without one, it falls back to the export region as a geofence. An explicit proxy + that unshackle cannot find raises an error. A region fallback gives a warning. """ session = requests.Session() session.headers.update(config.headers) + session.headers.update(headers or {}) params = ctx.parent.params if ctx.parent else {} if params.get("no_proxy"): @@ -224,7 +263,8 @@ class ImportService: tracks.manifest_url = manifest_url parser = PARSERS.get(manifest_type or "") - if manifest_url and parser is not None and manifest_type in ("DASH", "ISM"): + # An export from another tool has no track dicts, so HLS is parsed from the manifest too + if manifest_url and parser is not None and (manifest_type in ("DASH", "ISM") or not tracks_map): try: manifest = parser.from_url(url=manifest_url, session=self.session) except Exception as e: @@ -272,7 +312,7 @@ class ImportService: except Exception as e: self.log.warning(f"Skipping exported track {track_dict.get('id')!r}: {e}") continue - drm = self.rebuild_drm(track_dict) + drm = self.rebuild_drm(track_dict, title_id) if drm: track.drm = drm tracks.add(track, warn_only=True) @@ -307,18 +347,37 @@ class ImportService: def key_pool(self, title_id: Optional[str] = None) -> dict[UUID, str]: """Exported KID:KEY pairs as {UUID: key_hex}, for one title or across every title.""" - pool: dict[UUID, str] = {} - entries = [self.titles_data.get(title_id, {})] if title_id else self.titles_data.values() - for entry in entries: - for track_dict in (entry.get("tracks") or {}).values(): - for kid_hex, key in (track_dict.get("keys") or {}).items(): - pool[UUID(hex=kid_hex)] = key - return pool + entry = self.doc.get(title_id) if title_id else None + keys = entry.keys if entry else self.doc.key_pool() + return {UUID(hex=kid): key for kid, key in keys.items()} - def rebuild_drm(self, track_dict: dict[str, Any]) -> Optional[list[Any]]: - """Rebuild a DRM object (from stored PSSH, falling back to a stub) with the exported keys.""" - keys = track_dict.get("keys") or {} - drm_dicts = track_dict.get("drm") or [] + def title_drm_dicts(self, title_id: str) -> list[dict[str, Any]]: + """The title's DRM entries in ``drm_from_dict`` form. + + unshackle's own export keeps the exact ``to_dict`` output under ``x-unshackle``, so a + ClearKeyCENC licence URL or a PlayReady PSSH comes back byte for byte. + """ + entry = self.doc.get(title_id) + if entry is None: + return [] + own = (entry.extensions.get("x-unshackle") or {}).get("drm") + if own: + return list(own) + out = [] + for d in entry.drm: + system = {"widevine": "Widevine", "playready": "PlayReady", "clearkey": "ClearKeyCENC"}.get(d.system) + if system: + out.append({"system": system, "pssh_b64": d.pssh, "kids": list(entry.keys)}) + return out + + def rebuild_drm(self, track_dict: dict[str, Any], title_id: str) -> Optional[list[Any]]: + """Rebuild a DRM object (from stored PSSH, falling back to a stub) with the exported keys. + + Keys and DRM are title-level in the shared format; a legacy v2 track dict that still + carries its own is honoured first. + """ + keys = track_dict.get("keys") or {kid.hex: key for kid, key in self.key_pool(title_id).items()} + drm_dicts = track_dict.get("drm") or self.title_drm_dicts(title_id) if not drm_dicts and not keys: return None @@ -395,17 +454,20 @@ class ImportService: def exported_drm_system(self) -> str: """The DRM system the exporter licensed (e.g. 'playready'), defaulting to widevine.""" - for entry in self.titles_data.values(): - for track_dict in (entry.get("tracks") or {}).values(): - for drm_dict in track_dict.get("drm") or []: - if drm_dict.get("system"): - return drm_dict["system"].lower() + for entry in self.doc.titles: + for d in entry.drm: + if d.system: + return d.system return "widevine" def get_chapters(self, title: Title_T) -> Chapters: - entry = self.titles_data.get(str(title.id), {}) + entry = self.doc.get(str(title.id)) return Chapters( - [Chapter(ch["timestamp"], ch.get("name")) for ch in (entry.get("chapters") or []) if ch.get("timestamp")] + [ + Chapter(int(ch["start_ms"]), ch.get("title") or None) + for ch in (entry.chapters if entry else []) + if ch.get("start_ms") is not None + ] ) def get_widevine_service_certificate(self, **_: Any) -> Optional[str]: diff --git a/uv.lock b/uv.lock index 619998b1..c04a4afb 100644 --- a/uv.lock +++ b/uv.lock @@ -1074,6 +1074,11 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" }, ] +[[package]] +name = "mediaexport" +version = "0.1.0" +source = { git = "https://github.com/unshackle-dl/mediaexport?branch=main#07893237199fc865bd5c63b6848235e4f6ef81c8" } + [[package]] name = "more-itertools" version = "10.8.0" @@ -2155,6 +2160,7 @@ dependencies = [ { name = "langcodes" }, { name = "language-data" }, { name = "lxml" }, + { name = "mediaexport" }, { name = "mutagen" }, { name = "protobuf" }, { name = "pycaption" }, @@ -2223,6 +2229,7 @@ requires-dist = [ { name = "langcodes", specifier = ">=3.4.0,<4" }, { name = "language-data", specifier = ">=1.4.0" }, { name = "lxml", specifier = ">=5.2.1,<7" }, + { name = "mediaexport", git = "https://github.com/unshackle-dl/mediaexport?branch=main" }, { name = "mutagen", specifier = ">=1.47.0,<2" }, { name = "protobuf", specifier = ">=4.25.3,<7" }, { name = "pycaption", specifier = ">=2.2.6,<2.2.27" }, From a25483012115bf1077072dee970ed32689a69085 Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Sun, 20 Sep 2026 10:39:13 -0600 Subject: [PATCH 2/2] fix(export): import side-loaded subtitles and keep conflicting keys apart The importer builds subtitle tracks from track rows that carry a url, which the format promised and no reader did. The writer keeps a conflicting KID instead of overwriting it, identifies a manifest by its whole URL, writes subtitle rows as srt or vtt rather than enum names, and survives an export file it cannot read. --- docs/guide/downloading.md | 3 ++- tests/core/test_import_tracks.py | 41 ++++++++++++++++++++++++++++++++ unshackle/commands/dl.py | 35 ++++++++++++++++++--------- unshackle/core/import_service.py | 41 ++++++++++++++++++++++++++++++-- 4 files changed, 106 insertions(+), 14 deletions(-) diff --git a/docs/guide/downloading.md b/docs/guide/downloading.md index 71a719da..32524745 100644 --- a/docs/guide/downloading.md +++ b/docs/guide/downloading.md @@ -782,7 +782,8 @@ 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. +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. !!! warning "The export file is a secret" It holds the content keys and usually a signed manifest URL. Hand it over on purpose, diff --git a/tests/core/test_import_tracks.py b/tests/core/test_import_tracks.py index f503302b..c815bf4e 100644 --- a/tests/core/test_import_tracks.py +++ b/tests/core/test_import_tracks.py @@ -136,3 +136,44 @@ def test_fetch_failure_still_reports_a_possible_expiry(tmp_path: Path, failure: with pytest.raises(click.ClickException) as error: service.get_tracks(title) assert "may have expired" in error.value.message + + +def test_a_subtitle_side_load_from_another_tool_is_added(tmp_path: Path) -> None: + """A foreign export has no track dicts, so a ``tracks[]`` row with a url is the side-load.""" + export = { + "kind": "mediaexport", + "version": 1, + "service": {"tag": "EXAMPLE"}, + "titles": [ + { + "id": "movie-1", + "kind": "clip", + "title": "Example Movie", + "language": "ja", + "manifests": [{"url": "https://example.invalid/m.mpd", "type": "dash"}], + "tracks": [ + {"type": "video", "id": "v1", "codec": "avc"}, + { + "type": "subtitle", + "id": "sub-en", + "url": "https://example.invalid/en.srt", + "codec": "srt", + "language": "en", + "sdh": True, + }, + {"type": "subtitle", "id": "sub-x", "url": "https://example.invalid/x.srt"}, + ], + } + ], + } + path = tmp_path / "export.json" + path.write_text(json.dumps(export), encoding="utf8") + service = ImportService(click.Context(click.Command("dl")), "EXAMPLE", "movie-1", str(path)) + tracks = get_tracks(service, MPD_WITH_LANG) + assert [(t.descriptor.name, str(t.language)) for t in tracks.subtitles] == [ + ("DASH", "ja"), + ("URL", "en"), + ("URL", "ja"), + ] + 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 diff --git a/unshackle/commands/dl.py b/unshackle/commands/dl.py index d5e4b6b7..3a52255f 100644 --- a/unshackle/commands/dl.py +++ b/unshackle/commands/dl.py @@ -3924,7 +3924,12 @@ class dl: dicts and title meta go under ``x-unshackle`` for its own importer. """ with self.EXPORT_LOCK: - doc = mediaexport.read(export) if export.is_file() else mediaexport.Document(service_tag=self.service) + try: + doc = mediaexport.read(export) if export.is_file() else mediaexport.Document(service_tag=self.service) + except mediaexport.ExportError as e: + # a download mid-licensing must not die over the file it exports to + self.log.warning(f"Not exporting {title.id}: {export} is not a usable export ({e})") + return doc.generator.setdefault("app", "unshackle") doc.generator.setdefault("version", __version__) if not doc.region and getattr(self, "proxy_requested", False): @@ -3952,8 +3957,16 @@ class dl: if system and all(x.pssh != pssh or x.system != system for x in entry.drm): entry.drm.append(mediaexport.Drm(system, pssh)) content_keys = getattr(drm, "content_keys", None) or {} + pool = doc.key_pool() for kid, key in content_keys.items(): - entry.keys[kid.hex] = key + # a KID that already holds a different key must not be overwritten: + # the reader rejects a file that disagrees with itself + if pool.setdefault(kid.hex, key) != key: + self.log.warning( + f"KID {kid.hex} already has a different key in {export}, keeping the existing one" + ) + else: + entry.keys[kid.hex] = key mediaexport.write(export, doc) @@ -3968,25 +3981,25 @@ class dl: (t.descriptor.name.lower() for t in all_tracks if t.descriptor != Video.Descriptor.URL), "" ) session = getattr(getattr(self, "export_service", None), "session", None) - headers = { - k: v - for k, v in (getattr(session, "headers", None) or {}).items() - if k.lower() not in ("cookie", "authorization") - } + # mediaexport drops Cookie and Authorization itself, on write and on read + headers = dict(getattr(session, "headers", None) or {}) manifests = [mediaexport.Manifest(primary_url, manifest_type, headers, role="primary")] if primary_url else [] - seen = {primary_url.split("?")[0]} + # the whole URL identifies a manifest: two profiles on one endpoint are two manifests + seen = {primary_url} for t in all_tracks: url = str(t.url) - if t.descriptor in (Video.Descriptor.DASH, Video.Descriptor.ISM) and url.split("?")[0] not in seen: - seen.add(url.split("?")[0]) + if t.descriptor in (Video.Descriptor.DASH, Video.Descriptor.ISM) and url not in seen: + seen.add(url) manifests.append(mediaexport.Manifest(url, t.descriptor.name.lower(), role="extra")) rows = [] for t in all_tracks: + codec = getattr(t, "codec", None) row: dict[str, Any] = { "id": str(t.id), "type": t.__class__.__name__.lower(), - "codec": codec.name.lower() if (codec := getattr(t, "codec", None)) else "", + # a subtitle side-load names its file format so another tool can fetch it + "codec": (codec.value if isinstance(t, Subtitle) else codec.name).lower() if codec else "", "language": str(t.language) if t.language else "", "bitrate": int(bitrate) if (bitrate := getattr(t, "bitrate", None)) else None, } diff --git a/unshackle/core/import_service.py b/unshackle/core/import_service.py index dcce3129..b2bba50c 100644 --- a/unshackle/core/import_service.py +++ b/unshackle/core/import_service.py @@ -18,7 +18,7 @@ from unshackle.core.drm import drm_from_dict from unshackle.core.manifests import DASH, HLS, ISM from unshackle.core.remote_service import RemoteService, build_title, match_track, resolve_proxy_arg from unshackle.core.titles import Episode, Movies, Series, Title_T, Titles_T, remap_titles -from unshackle.core.tracks import Audio, Chapter, Chapters, Tracks, Video +from unshackle.core.tracks import Audio, Chapter, Chapters, Subtitle, Tracks, Video from unshackle.core.tracks.attachment import Attachment from unshackle.core.tracks.track import Track @@ -151,7 +151,7 @@ class ImportService: """ x = entry.extensions.get("x-unshackle") or {} meta = x.get("meta") or { - "type": entry.kind, + "type": entry.kind if entry.kind in ("movie", "episode", "song") else "movie", "id": entry.id, "name": entry.title, "series_title": entry.series, @@ -306,6 +306,10 @@ class ImportService: else: 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 track_dict in track_dicts: try: track = Track.from_dict(track_dict) @@ -345,6 +349,39 @@ 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``. + + 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. + """ + entry = self.doc.get(title_id) + out = [] + 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: + 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")), + ) + ) + return out + def key_pool(self, title_id: Optional[str] = None) -> dict[UUID, str]: """Exported KID:KEY pairs as {UUID: key_hex}, for one title or across every title.""" entry = self.doc.get(title_id) if title_id else None