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

@@ -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,

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

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,
             }

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