fix(manifests): skip filtered dash periods after the main content and keep them over export

A manifest snapshot module now saves and restores DASH/ISM parser state for --remote and export/import, so an import keeps the service's period filter without fetching the manifest.

docs/dev/rest-api/endpoints.md

@@ -1091,7 +1091,7 @@ Requires an authenticated remote session. Otherwise it returns `404 SESSION_NOT_
 
 Get tracks and chapters for a specific title within the remote session. Body: `{ "title_id": "..." }` (required). Unlike `list-tracks`, track objects here **include** download URLs, along with session headers/cookies and manifest data the client needs to download directly. A slow answer arrives as a [slow response](#slow-responses).
 
-`manifests` holds each served DASH/ISM manifest once. A service can build an AdaptationSet of its own that the served manifest does not contain. `track_manifests` holds a one-AdaptationSet MPD for each such track, and the client re-parses that in place of the whole manifest.
+`manifests` holds each served DASH/ISM manifest once. For DASH, `filtered_period_ids` names the periods that the download skips, such as ad periods. A service can build an AdaptationSet of its own that the served manifest does not contain. `track_manifests` holds a one-track MPD for each such track, and the client re-parses that in place of the whole manifest. That MPD starts at the period of the track and keeps the content periods after it.
 
 === "Response `200`"
 
@@ -1103,7 +1103,7 @@ Get tracks and chapters for a specific title within the remote session. Body: `{
       "subtitles": [ { "id": "s-1", "url": "https://..." } ],
       "chapters": [ { "timestamp": "00:00:00.000", "name": "Intro" } ],
       "attachments": [ { "url": "https://...", "name": "font.ttf", "mime_type": "font/ttf", "description": "" } ],
-      "manifests": [ { "type": "dash", "url": "https://...", "data": "<base64 zlib xml>" } ],
+      "manifests": [ { "type": "dash", "url": "https://...", "data": "<base64 zlib xml>", "filtered_period_ids": [ "ad-1" ] } ],
       "track_manifests": [ { "track_id": "s-1", "type": "dash", "url": "https://...", "data": "<base64 zlib xml>" } ],
       "session_headers": { "User-Agent": "..." },
       "session_cookies": { "sid": "..." },

docs/dev/rest-api/remote-sessions.md

@@ -213,8 +213,9 @@ hood the client walks a remote session through its lifecycle.
     chosen title. The tracks come back with playback URLs, and the server ships
     DASH/ISM manifests as compressed XML so the client can re-parse them locally for
     downloading. A service can build an AdaptationSet of its own that the served
-    manifest does not contain. Each such track gets a one-AdaptationSet MPD in
-    `track_manifests`, and the client re-parses that instead. Any
+    manifest does not contain. Each such track gets a one-track MPD in
+    `track_manifests`, with the content periods after its own, and the client
+    re-parses that instead. The client skips the same DASH periods as the server. Any
     `session_headers` and cookies the server used are merged into the
     client's local HTTP session. Each cookie in `session_cookie_jar` keeps its
     domain, path and `secure` flag, so it goes only to its own hosts. `session_cookies` carries

docs/guide/cli-reference.md

@@ -269,7 +269,7 @@ unshackle logs in to the service before the search. A service that declares a se
 
 ## `import`
 
-Reconstruct a download (download → decrypt → mux) from an `--export` JSON file without re-contacting the service. It re-fetches the manifest, injects the stored keys, and then runs the normal `dl` pipeline. The service tag is read from the export file, not passed by you.
+Reconstruct a download (download → decrypt → mux) from an `--export` JSON file without re-contacting the service. It uses the DASH or ISM manifest that the export saved, or fetches the manifest again when the export has none. It injects the stored keys and then runs the normal `dl` pipeline. The service tag is read from the export file, not passed by you.
 
 ```
 unshackle import EXPORT_FILE [DL_ARGS...]

docs/guide/downloading.md

@@ -913,8 +913,10 @@ unshackle dl --skip-dl EXAMPLE 81234567
 
 `--export` writes a `mediaexport` JSON file into the configured exports directory. It holds
 the manifest, the DRM init data, the content keys, the KIDs of each track and the title
-metadata for each title. `unshackle import` reads it to reconstruct a download later, and gives
-each track only the content keys for its KIDs. The importer also reads the
+metadata for each title. For a DASH or ISM title, it also holds the manifest as the service
+left it, with the ad periods that the service skips. `unshackle import` reads it to reconstruct
+a download later, and gives each track only the content keys for its KIDs. The import uses the
+saved manifest, so it skips the same periods and does not fetch the manifest again. The importer also reads the
 older unshackle `version: 2` files and unidl's own export files.
 
 ```shell title="Export track info and keys"

tests/core/test_manifest_snapshot.py

@@ -0,0 +1,189 @@
+"""A manifest snapshot must keep the periods a service filtered out across a capture, a JSON trip and a restore."""
+
+from __future__ import annotations
+
+import json
+from copy import deepcopy
+from typing import Any, Optional
+
+import pytest
+
+from unshackle.core.manifests import DASH
+from unshackle.core.manifests.snapshot import capture, parse, restore
+from unshackle.core.tracks import Tracks
+from unshackle.core.tracks.track import Track
+
+URL = "https://cdn.example/title/manifest.mpd"
+
+
+def period(pid: Optional[str], kind: str) -> str:
+    attr = f' id="{pid}"' if pid else ""
+    name = pid or "noid"
+    return f"""
+  <Period{attr} duration="PT30S">
+    <SupplementalProperty schemeIdUri="urn:example:adtype" value="{kind}"/>
+    <AdaptationSet contentType="video" mimeType="video/mp4" lang="en">
+      <Representation id="v1" codecs="avc1.640028" bandwidth="5000000" width="1920" height="1080" frameRate="24">
+        <SegmentTemplate media="{name}/v1_$Number$.m4s" initialization="{name}/v1_init.mp4"
+                         duration="4" timescale="1" startNumber="1"/>
+      </Representation>
+    </AdaptationSet>
+    <AdaptationSet contentType="audio" mimeType="audio/mp4" lang="en">
+      <Representation id="a1" codecs="mp4a.40.2" bandwidth="128000" audioSamplingRate="48000">
+        <SegmentTemplate media="{name}/a1_$Number$.m4s" initialization="{name}/a1_init.mp4"
+                         duration="4" timescale="1" startNumber="1"/>
+      </Representation>
+    </AdaptationSet>
+  </Period>"""
+
+
+def mpd(*periods: str) -> str:
+    return f"""<?xml version="1.0"?>
+<MPD xmlns="urn:mpeg:dash:schema:mpd:2011" type="static" mediaPresentationDuration="PT3M" minBufferTime="PT2S"
+     profiles="urn:mpeg:dash:profile:isoff-live:2011">{"".join(periods)}
+</MPD>"""
+
+
+MPD = mpd(
+    period("pre-1", "PRE_ROLL"),
+    period("main-1", "CONTENT"),
+    period(None, "MID_ROLL"),
+    period("main-2", "CONTENT"),
+    period("mid-2", "MID_ROLL"),
+    period("main-3", "CONTENT"),
+)
+
+
+def is_ad(p: Any) -> bool:
+    return any(x.get("value", "").endswith("_ROLL") for x in p.findall("SupplementalProperty"))
+
+
+def service_tracks(text: str = MPD) -> Tracks:
+    return DASH.from_text(text, URL).to_tracks(language="en", period_filter=is_ad)
+
+
+def detach_video(tracks: Tracks, track_id: Optional[str] = None) -> Track:
+    """Give the video an AdaptationSet outside the tree, the way a service that builds its own does."""
+    video = tracks.videos[0]
+    data = video.data["dash"]
+    data["adaptation_set"] = deepcopy(data["adaptation_set"])
+    data["representation"] = data["adaptation_set"].find("Representation")
+    if track_id:
+        video.id = track_id
+    return video
+
+
+def export_and_import(tmp_path: Any, tracks: Tracks) -> Tracks:
+    from tests.core.test_export import StubService, import_ctx, make_dl
+    from unshackle.core.import_service import ImportService
+    from unshackle.core.titles import Movie
+
+    export = tmp_path / "export.json"
+    title = Movie(id_="movie-1", service=StubService, name="Example Movie", year=2024, language="en")
+    title.tracks = tracks
+    runner = make_dl()
+    for track in title.tracks:
+        runner.write_export(export, title, track)
+    service = ImportService(import_ctx(), "EXAMPLE", "movie-1", str(export))
+    return service.get_tracks(next(iter(service.get_titles())))
+
+
+def stitched(track: Track) -> list[str]:
+    """The media template of each period download_track would stitch for this track."""
+    data = track.data["dash"]
+    filtered = data.get("filtered_period_ids", [])
+    out = []
+    for p in data["manifest"].findall("Period"):
+        if not DASH.is_content_period(p, filtered):
+            continue
+        if p is data["period"]:
+            rep = data["representation"]
+        else:
+            found = DASH.resolve_representation(p, data["representation_id"], data["adaptation_set"])
+            if found is None:
+                continue
+            rep = found[1]
+        out.append(rep.find("SegmentTemplate").get("media").split("/")[0])
+    return out
+
+
+def round_trip(tracks: Tracks, level: int = -1) -> Tracks:
+    snapshot = json.loads(json.dumps(capture(tracks, level)))
+    assert {"manifests", "track_manifests"} <= set(snapshot)
+    client = Tracks()
+    for track in tracks:
+        client.add(Track.from_dict(track.to_dict()))
+    restore(client, snapshot)
+    return client
+
+
+def test_service_filter_drops_every_ad_period_after_the_first_content_period():
+    assert stitched(service_tracks().videos[0]) == ["main-1", "main-2", "main-3"]
+
+
+def test_whole_manifest_restore_keeps_the_service_filter():
+    video = round_trip(service_tracks()).videos[0]
+    assert video.data["dash"]["period"].get("id") == "main-1"
+    assert stitched(video) == ["main-1", "main-2", "main-3"]
+
+
+def test_one_track_fragment_keeps_every_period_and_the_service_filter():
+    tracks = service_tracks()
+    detach_video(tracks)
+    snapshot = json.loads(json.dumps(capture(tracks)))
+    assert len(snapshot["track_manifests"]) == 1
+    video = round_trip(tracks).videos[0]
+    assert stitched(video) == ["main-1", "main-2", "main-3"]
+
+
+def test_import_restores_the_service_filter_from_the_export_without_a_fetch(tmp_path):
+    video = export_and_import(tmp_path, service_tracks()).videos[0]
+    assert stitched(video) == ["main-1", "main-2", "main-3"]
+
+
+def test_capture_on_a_remote_client_saves_the_whole_manifest_not_a_fragment():
+    tracks = service_tracks()
+    detach_video(tracks)
+    client = round_trip(tracks)
+    saved = parse(capture(client), URL, "en")
+    assert saved is not None
+    assert (len(saved.videos), len(saved.audio)) == (1, 1)
+
+
+def test_import_rebuilds_a_service_built_track_from_its_fragment(tmp_path):
+    tracks = service_tracks()
+    detach_video(tracks, "service-built")
+    imported = export_and_import(tmp_path, tracks)
+    video = next(t for t in imported.videos if t.id == "service-built")
+    assert stitched(video) == ["main-1", "main-2", "main-3"]
+
+
+def test_a_filtered_period_that_reuses_a_kept_id_keeps_the_content():
+    tracks = service_tracks(mpd(period("main-1", "CONTENT"), period("main-1", "MID_ROLL"), period("main-2", "CONTENT")))
+    assert stitched(tracks.videos[0]) == ["main-1", "main-2"]
+
+
+@pytest.mark.parametrize("snapshot", ["garbage", {"manifests": "x"}, {"manifests": [1]}, {"track_manifests": {"a": 1}}])
+def test_a_malformed_snapshot_restores_nothing(snapshot):
+    client = Tracks()
+    client.add(Track.from_dict(service_tracks().videos[0].to_dict()))
+    restore(client, snapshot)
+    assert not client.videos[0].data.get("dash")
+    assert parse(snapshot, URL, "en") is None
+
+
+def test_a_saved_manifest_that_does_not_parse_gives_none():
+    assert parse({"manifests": [{"type": "dash", "url": URL, "data": "bm90IHpsaWI="}]}, URL, "en") is None
+
+
+def test_compression_level_zero_restores():
+    video = round_trip(service_tracks(), level=0).videos[0]
+    assert stitched(video) == ["main-1", "main-2", "main-3"]
+
+
+def test_an_export_from_a_remote_client_imports_every_track(tmp_path):
+    tracks = service_tracks()
+    detach_video(tracks)
+    imported = export_and_import(tmp_path, round_trip(tracks))
+    assert (len(imported.videos), len(imported.audio)) == (1, 1)
+    assert stitched(imported.videos[0]) == ["main-1", "main-2", "main-3"]

tests/remote/unit/test_remote_service_helpers.py

@@ -7,6 +7,7 @@ from enum import Enum
 import pytest
 
 from unshackle.core.api.handlers import serialize_audio_track, serialize_subtitle_track, serialize_video_track
+from unshackle.core.manifests.snapshot import match_track
 from unshackle.core.remote_service import (
     RemoteService,
     build_title,
@@ -15,7 +16,6 @@ from unshackle.core.remote_service import (
     deserialize_subtitle,
     deserialize_video,
     enum_get,
-    match_track,
     reconstruct_drm,
 )
 from unshackle.core.titles.episode import Episode
@@ -524,7 +524,7 @@ def test_resolve_manifest_data_matches_across_per_range_manifests() -> None:
     from tests.core.test_ism_init import VIDEO_HEVC_PQ_CPD, VIDEO_HEVC_SDR_CPD
     from tests.core.test_ism_range import manifest_xml
     from unshackle.core.manifests import ISM
-    from unshackle.core.remote_service import resolve_manifest_data
+    from unshackle.core.manifests.snapshot import restore
 
     urls = {"sdr": "https://x/sdr/manifest", "hdr": "https://x/hdr/manifest"}
     xml = {"sdr": manifest_xml(VIDEO_HEVC_SDR_CPD), "hdr": manifest_xml(VIDEO_HEVC_PQ_CPD)}
@@ -549,7 +549,7 @@ def test_resolve_manifest_data_matches_across_per_range_manifests() -> None:
         for k in ("sdr", "hdr")
     ]
 
-    resolve_manifest_data(tracks, manifests)
+    restore(tracks, {"manifests": manifests})
 
     assert remote_hdr.data["ism"]["quality_level"].get("CodecPrivateData") == VIDEO_HEVC_PQ_CPD
 
@@ -680,9 +680,8 @@ def test_no_dl_params_when_the_ctx_has_no_selection() -> None:
 def test_resolve_manifest_data_keeps_the_periods_the_server_filtered() -> None:
     """A service drops periods (a mid-roll ad) on the server only. The client must skip the
     same periods after it re-parses the manifest, or a remote download joins them back in."""
-    from unshackle.core.api.handlers import extract_manifests
     from unshackle.core.manifests import DASH
-    from unshackle.core.remote_service import resolve_manifest_data
+    from unshackle.core.manifests.snapshot import capture, restore
 
     periods = "".join(
         f"""<Period id="{pid}" duration="PT1S"><AdaptationSet contentType="video" mimeType="video/mp4" lang="en">
@@ -700,6 +699,6 @@ def test_resolve_manifest_data_keeps_the_periods_the_server_filtered() -> None:
     remote = deserialize_video(serialize_video_track(server.videos[0]))
     tracks = build_tracks({"video": [], "audio": [], "subtitles": []})
     tracks.videos.append(remote)
-    resolve_manifest_data(tracks, extract_manifests(server))
+    restore(tracks, capture(server))
 
     assert remote.data["dash"]["filtered_period_ids"] == ["ad1"]

tests/remote/unit/test_track_manifest_fragments.py

@@ -6,8 +6,7 @@ import pytest
 from langcodes import Language
 from lxml import etree
 
-from unshackle.core.api.handlers import extract_track_manifests
-from unshackle.core.remote_service import resolve_track_manifests
+from unshackle.core.manifests.snapshot import capture, restore
 from unshackle.core.tracks import Subtitle, Tracks
 
 pytestmark = pytest.mark.unit
@@ -53,8 +52,8 @@ def test_detached_adaptation_set_round_trips() -> None:
     )
     tracks = Tracks([track])
 
-    fragments = extract_track_manifests(tracks)
-    assert len(fragments) == 1
+    snapshot = capture(tracks)
+    assert len(snapshot["track_manifests"]) == 1
 
     client_track = Subtitle(
         id_="abc123",
@@ -63,7 +62,7 @@ def test_detached_adaptation_set_round_trips() -> None:
         language=Language.get("en"),
         descriptor=Subtitle.Descriptor.URL,
     )
-    resolve_track_manifests(Tracks([client_track]), fragments)
+    restore(Tracks([client_track]), snapshot)
 
     dash = client_track.data["dash"]
     assert dash["manifest"].find("BaseURL").text == "https://cdn.example/base/"
@@ -101,7 +100,7 @@ def test_single_track_manifest_round_trips_by_id() -> None:
         sdh=True,
         descriptor=Subtitle.Descriptor.URL,
     )
-    resolve_track_manifests(Tracks([client_track]), extract_track_manifests(Tracks([track])))
+    restore(Tracks([client_track]), capture(Tracks([track])))
 
     assert client_track.descriptor == Subtitle.Descriptor.DASH
     assert client_track.data["dash"]["representation_id"] == "sub-0"
@@ -127,7 +126,7 @@ def test_in_tree_adaptation_set_is_not_shipped() -> None:
             }
         },
     )
-    assert extract_track_manifests(Tracks([track])) == []
+    assert capture(Tracks([track]))["track_manifests"] == []
 
 
 def test_fragment_keeps_only_the_track_representation() -> None:
@@ -153,7 +152,7 @@ def test_fragment_keeps_only_the_track_representation() -> None:
         language=Language.get("en"),
         descriptor=Subtitle.Descriptor.URL,
     )
-    resolve_track_manifests(Tracks([client_track]), extract_track_manifests(Tracks([track])))
+    restore(Tracks([client_track]), capture(Tracks([track])))
 
     dash = client_track.data["dash"]
     assert dash["representation_id"] == "sub-1"

unshackle/commands/dl.py

@@ -76,6 +76,7 @@ from unshackle.core.drm import (
 )
 from unshackle.core.events import events
 from unshackle.core.export_name import ExportNamer, move_export, plural, season_episodes, title_label
+from unshackle.core.manifests.snapshot import capture
 from unshackle.core.providers.anilist import parse_anilist_ref
 from unshackle.core.providers.tvdb import SEASON_TYPES, parse_int
 from unshackle.core.proxies.resolve import (
@@ -1247,6 +1248,7 @@ class dl:
     LICENCE_LOCK = Lock()
     VAULT_WRITER = ThreadPoolExecutor(1, thread_name_prefix="vault-writer")
     EXPORT_LOCK = Lock()
+    export_snapshots: dict[str, Optional[dict[str, Any]]]
     LICENSE_KEY_CACHE: dict[UUID, str] = {}
     ALL_DRM_LOCK = Lock()
 
@@ -4219,9 +4221,19 @@ class dl:
         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.
+        dicts, title meta and manifest snapshot go under ``x-unshackle`` for its own importer.
         """
         namer: Optional[ExportNamer] = getattr(self, "export_namer", None)
+        snapshots = getattr(self, "export_snapshots", None)
+        if snapshots is None:
+            snapshots = self.export_snapshots = {}
+        if str(title.id) not in snapshots:
+            try:
+                fresh = capture(title.tracks)
+                snapshots[str(title.id)] = fresh if any(fresh.values()) else None
+            except Exception as e:
+                self.log.warning(f"Not saving the manifest of {title.id} in {export}: {e!r}")
+                snapshots[str(title.id)] = None
         with self.EXPORT_LOCK:
             if namer and namer.path:
                 export = namer.path
@@ -4248,6 +4260,8 @@ class dl:
                 track_dict["unwanted_segments"] = unwanted
             elif unwanted is not None:
                 track_dict.pop("unwanted_segments", None)
+            if saved := snapshots[str(title.id)]:
+                entry.ext("unshackle")["snapshot"] = saved
             # the row names the track's KIDs, so an import gives the track only their keys
             track_kids = [kid.hex for kid in own_kids(drm)] if drm is not None else []
             for row in entry.tracks:

unshackle/core/api/handlers.py

@@ -36,6 +36,7 @@ from unshackle.core.cdm.detect import cdm_type_stub
 from unshackle.core.config import config, load_yaml
 from unshackle.core.constants import AUDIO_CODEC_MAP, DYNAMIC_RANGE_MAP, VIDEO_CODEC_MAP, AnyTrack
 from unshackle.core.drm import BlowfishStripe, cbcs
+from unshackle.core.manifests.snapshot import capture
 from unshackle.core.providers.anilist import parse_anilist_ref
 from unshackle.core.proxies.resolve import initialize_proxy_providers, is_loopback, resolve_proxy
 from unshackle.core.services import Services
@@ -1575,121 +1576,6 @@ def stamp_service_flags(serialized: Dict[str, Any], service_instance: Any) -> Di
     return serialized
 
 
-def extract_manifests(tracks: Tracks) -> List[Dict[str, Any]]:
-    """Extract manifest data from tracks for client-side re-parsing.
-
-    Serializes DASH and ISM manifest XML as zlib-compressed base64 strings
-    so the client can reconstruct track.data locally. HLS tracks download
-    directly from their URL, so this function does not serialise their manifest.
-    """
-    import base64
-    import zlib
-
-    from lxml import etree
-
-    from unshackle.core.config import config as app_config
-
-    compression_level = app_config.serve.get("compression_level", 1)
-
-    seen: set[str] = set()
-    manifests: List[Dict[str, Any]] = []
-
-    for track in list(tracks.videos) + list(tracks.audio) + list(tracks.subtitles):
-        manifest_url = str(track.url) if track.url else None
-        if not manifest_url or manifest_url in seen:
-            continue
-
-        if track.data.get("dash") and track.data["dash"].get("manifest") is not None:
-            seen.add(manifest_url)
-            xml_bytes = etree.tostring(track.data["dash"]["manifest"], xml_declaration=True, encoding="UTF-8")
-            compressed = zlib.compress(xml_bytes, compression_level) if compression_level else xml_bytes
-            manifests.append(
-                {
-                    "type": "dash",
-                    "url": manifest_url,
-                    "data": base64.b64encode(compressed).decode("ascii"),
-                    "filtered_period_ids": list(track.data["dash"].get("filtered_period_ids") or []),
-                }
-            )
-        elif track.data.get("ism") and track.data["ism"].get("manifest") is not None:
-            seen.add(manifest_url)
-            xml_bytes = etree.tostring(track.data["ism"]["manifest"], xml_declaration=True, encoding="UTF-8")
-            compressed = zlib.compress(xml_bytes, compression_level) if compression_level else xml_bytes
-            manifests.append(
-                {
-                    "type": "ism",
-                    "url": manifest_url,
-                    "data": base64.b64encode(compressed).decode("ascii"),
-                }
-            )
-
-    return manifests
-
-
-def extract_track_manifests(tracks: Tracks) -> List[Dict[str, Any]]:
-    """Build a one-AdaptationSet MPD for each track the client cannot find again in the served manifest.
-
-    A service can build an AdaptationSet of its own, out of merged segment timelines or
-    copied nodes, that never joins the manifest tree the server serialises. A service can also
-    build a manifest that holds only its own track, which the client's attribute match cannot
-    tell apart from others. Each of those tracks gets its own MPD, matched by track id.
-    """
-    import base64
-    import zlib
-    from copy import deepcopy
-
-    from lxml import etree
-
-    from unshackle.core.config import config as app_config
-
-    compression_level = app_config.serve.get("compression_level", 1)
-    fragments: List[Dict[str, Any]] = []
-    representations: Dict[int, int] = {}
-
-    for track in list(tracks.videos) + list(tracks.audio) + list(tracks.subtitles):
-        dash = (getattr(track, "data", None) or {}).get("dash") or {}
-        manifest = dash.get("manifest")
-        adaptation_set = dash.get("adaptation_set")
-        if manifest is None or adaptation_set is None:
-            continue
-        if adaptation_set.getroottree().getroot() is manifest:
-            count = once(representations, id(manifest), lambda: len(manifest.findall(".//Representation")))
-            if count > 1:
-                continue
-
-        root = etree.Element("MPD", dict(manifest.attrib))
-        for child in manifest:
-            if child.tag != "Period":
-                root.append(deepcopy(child))
-
-        period = dash.get("period")
-        new_period = etree.SubElement(root, "Period", dict(period.attrib) if period is not None else {})
-        for child in period if period is not None else []:
-            if child.tag != "AdaptationSet":
-                new_period.append(deepcopy(child))
-        new_adaptation_set = deepcopy(adaptation_set)
-        new_period.append(new_adaptation_set)
-
-        representation = dash.get("representation")
-        if representation is not None:
-            for sibling in new_adaptation_set.findall("Representation"):
-                new_adaptation_set.remove(sibling)
-            new_adaptation_set.append(deepcopy(representation))
-
-        xml_bytes = etree.tostring(root, xml_declaration=True, encoding="UTF-8")
-        compressed = zlib.compress(xml_bytes, compression_level) if compression_level else xml_bytes
-        fragments.append(
-            {
-                "track_id": str(track.id),
-                "type": "dash",
-                "url": str(track.url) if track.url else None,
-                "data": base64.b64encode(compressed).decode("ascii"),
-            }
-        )
-
-    return fragments
-
-
 def serialize_drm(drm_list: Any) -> Optional[List[Dict[str, Any]]]:
     """Serialise DRM objects to JSON-serializable list."""
     if not drm_list:
@@ -4202,9 +4088,7 @@ async def session_tracks_handler(
             except (Exception, SystemExit):
                 session.chapters_by_title[str(title_id)] = []
 
-            manifests, track_manifests = await asyncio.to_thread(
-                lambda: (extract_manifests(tracks), extract_track_manifests(tracks))
-            )
+            snapshot = await asyncio.to_thread(capture, tracks, config.serve.get("compression_level", 1))
 
             session_headers, session_cookies = client_session_auth(session)
 
@@ -4254,8 +4138,7 @@ async def session_tracks_handler(
                     {"timestamp": ch.timestamp, "name": ch.name}
                     for ch in session.chapters_by_title.get(str(title_id), [])
                 ],
-                "manifests": manifests,
-                "track_manifests": track_manifests,
+                **snapshot,
                 "session_headers": session_headers,
                 "session_cookies": cookie_values(session_cookies),
                 "session_cookie_jar": session_cookies,

unshackle/core/import_service.py

@@ -17,11 +17,10 @@ from unshackle.core.config import config
 from unshackle.core.constants import AnyTrack
 from unshackle.core.credential import Credential
 from unshackle.core.drm import ClearKey, MonaLisa, drm_from_dict, real_kids
-from unshackle.core.manifests import DASH, HLS, ISM, MissingLanguageError
+from unshackle.core.manifests import DASH, HLS, ISM, MissingLanguageError, snapshot
 from unshackle.core.remote_service import (
     RemoteService,
     build_title,
-    match_track,
     resolve_proxy_arg,
     segment_uri_filter,
 )
@@ -96,12 +95,13 @@ def resolve_import_manifest_data(
     session: requests.Session,
     language: Any,
     title: Title_T,
+    saved: Optional[dict[str, Any]] = None,
 ) -> None:
     """Fill ``track.data`` for DASH/ISM tracks rebuilt from export dicts.
 
     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
+    ``title.tracks.manifest_url``) or carry one MPD per adaptation set. A track takes its
+    data from the export's manifest snapshot first; else its ``url`` is re-fetched and
     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.
     """
@@ -115,10 +115,12 @@ def resolve_import_manifest_data(
         for track in [*tracks.videos, *tracks.audio, *tracks.subtitles]
         if track.descriptor.name == manifest_type and not track.data.get(data_key)
     ]
+    fallback_lang = fallback_language(language)
+    snapshot.restore(Tracks(*pending), saved, language=fallback_lang)
+    pending = [track for track in pending if not track.data.get(data_key)]
     if not pending:
         return
 
-    fallback_lang = fallback_language(language)
     for url in {str(track.url) for track in pending if track.url}:
         try:
             manifest = parser.from_url(url=url, session=session)
@@ -138,7 +140,7 @@ def resolve_import_manifest_data(
         for track in pending:
             if track.data.get(data_key) or str(track.url) != url:
                 continue
-            matched = match_track(track, local_tracks)
+            matched = snapshot.match_track(track, local_tracks)
             if matched and matched.data.get(data_key):
                 track.data.update(matched.data)
 
@@ -268,6 +270,7 @@ class ImportService:
             "manifest_url": primary.url if primary else None,
             "manifest_type": manifest_type or None,
             "tracks": x.get("tracks") or {},
+            "snapshot": x.get("snapshot"),
             "chapters": entry.chapters,
             "attachments": x.get("attachments") or [],
         }
@@ -380,38 +383,48 @@ class ImportService:
         parser = PARSERS.get(manifest_type or "")
         # 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:
-                raise click.ClickException(
-                    f"Failed to fetch the {manifest_type} manifest for '{title}', the URL may have expired: {e}"
-                )
+            no_language = (
+                f"No language for '{title}': the {manifest_type} manifest has none and Title.language is unset"
+            )
             language = fallback_language(title.language)
-            if language is None and isinstance(manifest, HLS):
-                language = hls_audio_language(manifest)
-                if language is None:
-                    raise click.ClickException(f"No language for '{title}': the export and the HLS manifest have none")
             try:
-                parsed = manifest.to_tracks(language=language)
+                parsed = snapshot.parse(entry.get("snapshot"), str(manifest_url), language)
             except MissingLanguageError:
-                raise click.ClickException(
-                    f"No language for '{title}': the {manifest_type} manifest has none and Title.language is unset"
-                )
-            except Exception as e:
-                raise click.ClickException(f"Failed to parse the {manifest_type} manifest for '{title}': {e}")
+                raise click.ClickException(no_language)
+            if parsed is None:
+                try:
+                    manifest = parser.from_url(url=manifest_url, session=self.session)
+                except Exception as e:
+                    raise click.ClickException(
+                        f"Failed to fetch the {manifest_type} manifest for '{title}', the URL may have expired: {e}"
+                    )
+                if language is None and isinstance(manifest, HLS):
+                    language = hls_audio_language(manifest)
+                    if language is None:
+                        raise click.ClickException(
+                            f"No language for '{title}': the export and the HLS manifest have none"
+                        )
+                try:
+                    parsed = manifest.to_tracks(language=language)
+                except MissingLanguageError:
+                    raise click.ClickException(no_language)
+                except Exception as e:
+                    raise click.ClickException(f"Failed to parse the {manifest_type} manifest for '{title}': {e}")
             for parsed_track in parsed:
                 tracks.add(parsed_track)
             parsed_ids = {str(t.id) for t in tracks}
             url_dicts = [t for t in tracks_map.values() if t.get("descriptor", "URL") == "URL"]
-            # Merge back only tracks from another manifest, such as a separate HEVC one; the
-            # re-parse already covers manifest_url's own. Query stripped when comparing, since a
-            # service can sign the URL it fetches and register the bare one for the same manifest.
+            # Merge back only tracks from another manifest, such as a separate HEVC one, or that
+            # the snapshot keeps as a one-track fragment; the re-parse already covers manifest_url's
+            # own. Query stripped when comparing, since a service can sign the URL it fetches and
+            # register the bare one for the same manifest.
             manifest_base = str(manifest_url).split("?")[0]
+            fragments = snapshot.fragment_ids(entry.get("snapshot"))
             foreign_dicts = [
                 t
                 for t in tracks_map.values()
                 if t.get("descriptor") == manifest_type
-                and str(t.get("url")).split("?")[0] != manifest_base
+                and (str(t.get("url")).split("?")[0] != manifest_base or str(t.get("id")) in fragments)
                 and str(t.get("id")) not in parsed_ids
             ]
             track_dicts = [*url_dicts, *foreign_dicts]
@@ -452,6 +465,7 @@ class ImportService:
             session=self.session,
             language=title.language,
             title=title,
+            saved=entry.get("snapshot"),
         )
 
         for attachment in entry.get("attachments") or []:

unshackle/core/manifests/dash.py

@@ -11,7 +11,7 @@ from contextlib import ExitStack, closing
 from copy import deepcopy
 from functools import lru_cache, partial
 from pathlib import Path
-from typing import Any, BinaryIO, Callable, Optional, Union, cast
+from typing import Any, BinaryIO, Callable, Collection, Optional, Union, cast
 from urllib.parse import urljoin, urlparse
 from uuid import UUID
 from zlib import crc32
@@ -239,7 +239,9 @@ class DASH:
         Parameters:
             language: The Title's Original Recorded Language. unshackle also uses it as a
                 fallback track language value when the manifest gives no language information.
-            period_filter: Filter out period's within the manifest.
+            period_filter: Filter out period's within the manifest. A filtered period without an
+                id, or with the id of a kept period, gets its own, so filtered_period_ids names
+                every filtered period and no kept one.
 
         Only the first main-content period becomes tracks. Later periods do not add tracks of their
         own, but download_track stitches the segments of every content period into each returned
@@ -251,14 +253,19 @@ class DASH:
 
         filtered_period_ids: list[str] = []
 
-        for period in self.manifest.findall("Period"):
-            if callable(period_filter) and period_filter(period):
-                if period_id := period.get("id"):
-                    filtered_period_ids.append(period_id)
-                continue
-            if not DASH.is_content_period(period, []):
-                if period_id := period.get("id"):
-                    filtered_period_ids.append(period_id)
+        periods = self.manifest.findall("Period")
+        filtered = {
+            p for p in periods if (callable(period_filter) and period_filter(p)) or not DASH.is_content_period(p, [])
+        }
+        kept_ids = {p.get("id") for p in periods if p not in filtered}
+        for index, period in enumerate(periods):
+            if period in filtered:
+                if not period.get("id") or period.get("id") in kept_ids:
+                    period.set("id", f"unshackle-period-{index}")
+                filtered_period_ids.append(period.get("id"))
+
+        for period in periods:
+            if period in filtered:
                 continue
 
             for adaptation_set in period.findall("AdaptationSet"):
@@ -437,7 +444,7 @@ class DASH:
         adaptation_set: Element = track.data["dash"]["adaptation_set"]
         representation: Element = track.data["dash"]["representation"]
         rep_id: Optional[str] = track.data["dash"].get("representation_id") or representation.get("id")
-        filtered_period_ids: list[str] = track.data["dash"].get("filtered_period_ids", [])
+        filtered_period_ids = set(track.data["dash"].get("filtered_period_ids", []))
 
         # Preserve existing DRM if it was set by the service, especially when service set Widevine
         # but manifest only contains PlayReady protection (common scenario for some services)
@@ -926,7 +933,7 @@ class DASH:
         return bool(subsegments) and subsegments[0][1] == first_start and subsegments[-1][2] == last_end
 
     @staticmethod
-    def is_content_period(period: Element, filtered_period_ids: list[str]) -> bool:
+    def is_content_period(period: Element, filtered_period_ids: Collection[str]) -> bool:
         """Whether a period is a valid content period (not an ad, not filtered, not trick mode)."""
         period_id = period.get("id")
         if period_id and period_id in filtered_period_ids:

unshackle/core/manifests/snapshot.py

@@ -0,0 +1,298 @@
+"""The manifest snapshot: the DASH and ISM parser state of a title's tracks, saved for --remote and export/import.
+
+``capture`` saves each manifest tree as the service left it, with its filtered periods, plus a
+one-track fragment for each track that the full tree cannot identify again. ``restore`` gives
+that state back to tracks rebuilt somewhere else, where the service code does not run.
+"""
+
+from __future__ import annotations
+
+import base64
+import logging
+import zlib
+from copy import deepcopy
+from typing import Any, Optional
+
+from lxml import etree
+
+from unshackle.core.manifests.dash import DASH, MissingLanguageError
+from unshackle.core.manifests.ism import ISM
+from unshackle.core.tracks import Audio, Subtitle, Tracks, Video
+from unshackle.core.tracks.track import Track
+
+log = logging.getLogger("snapshot")
+
+Snapshot = dict[str, list[dict[str, Any]]]
+
+
+def capture(tracks: Tracks, level: int = zlib.Z_DEFAULT_COMPRESSION) -> Snapshot:
+    """The snapshot of these tracks, JSON-safe, at zlib ``level`` (0 gives an uncompressed zlib stream)."""
+    counts: dict[int, int] = {}
+    return {"manifests": whole_manifests(tracks, level, counts), "track_manifests": fragments(tracks, level, counts)}
+
+
+def restore(tracks: Tracks, snapshot: Any, language: Any = None) -> None:
+    """Fill ``track.data`` of each DASH or ISM track the snapshot holds; a track it does not hold is left as it is.
+
+    ``language`` is the parse fallback when no track has a language of its own.
+    """
+    restore_fragments(tracks, entries(snapshot, "track_manifests"))
+    restore_manifests(tracks, entries(snapshot, "manifests"), language)
+
+
+def parse(snapshot: Any, url: str, language: Any = None) -> Optional[Tracks]:
+    """Every track of the saved whole manifest for ``url``, with its filtered periods, else None.
+
+    The URL also matches without its query, since a service can sign the URL it fetches. A saved
+    manifest that does not parse gives None, so the caller can fetch the manifest instead.
+    """
+    saved = [m for m in entries(snapshot, "manifests") if isinstance(m.get("url"), str)]
+    entry = next((m for m in saved if m["url"] == url), None) or next(
+        (m for m in saved if m["url"].split("?")[0] == url.split("?")[0]), None
+    )
+    if entry is None:
+        return None
+    try:
+        tracks = to_tracks(entry, language)
+    except MissingLanguageError:
+        raise
+    except Exception as e:
+        log.warning("Failed to re-parse the saved %s manifest from %s: %s", entry.get("type"), url, e)
+        return None
+    return tracks if tracks is not None and list(tracks) else None
+
+
+def fragment_ids(snapshot: Any) -> set[str]:
+    """The ids of the tracks the snapshot holds a one-track fragment for."""
+    return {str(e["track_id"]) for e in entries(snapshot, "track_manifests") if e.get("track_id") is not None}
+
+
+def entries(snapshot: Any, key: str) -> list[dict[str, Any]]:
+    """The dict entries under ``key``; a malformed snapshot gives none."""
+    value = snapshot.get(key) if isinstance(snapshot, dict) else None
+    return [e for e in value if isinstance(e, dict)] if isinstance(value, list) else []
+
+
+def pack(element: Any, level: int) -> str:
+    xml_bytes = etree.tostring(element, xml_declaration=True, encoding="UTF-8")
+    return base64.b64encode(zlib.compress(xml_bytes, level)).decode("ascii")
+
+
+def unpack(data: str) -> Any:
+    from unshackle.core.api.compression import safe_inflate
+
+    return etree.fromstring(safe_inflate(base64.b64decode(data)))
+
+
+def counted(manifest: Any, tag: str, counts: dict[int, int]) -> int:
+    """The number of ``tag`` elements in ``manifest``, counted once per tree."""
+    if id(manifest) not in counts:
+        counts[id(manifest)] = len(manifest.findall(tag))
+    return counts[id(manifest)]
+
+
+def whole_manifests(tracks: Tracks, level: int, counts: dict[int, int]) -> list[dict[str, Any]]:
+    """One entry per manifest URL, and for DASH its filtered period ids.
+
+    When tracks of one URL point at different trees (a one-track fragment restored on a remote
+    client, or a tree a service built for one track), capture saves the tree with the most tracks.
+    """
+    best: dict[str, tuple[int, str, dict[str, Any]]] = {}
+    for track in tracks:
+        if not track.url:
+            continue
+        for kind, tag in (("dash", ".//Representation"), ("ism", ".//QualityLevel")):
+            data = track.data.get(kind) or {}
+            if data.get("manifest") is not None:
+                break
+        else:
+            continue
+        url = str(track.url)
+        size = counted(data["manifest"], tag, counts)
+        if url not in best or size > best[url][0]:
+            best[url] = (size, kind, data)
+
+    out: list[dict[str, Any]] = []
+    for url, (_, kind, data) in best.items():
+        entry: dict[str, Any] = {"type": kind, "url": url, "data": pack(data["manifest"], level)}
+        if kind == "dash":
+            entry["filtered_period_ids"] = list(data.get("filtered_period_ids") or [])
+        out.append(entry)
+    return out
+
+
+def one_track_period(period: Any, adaptation_set: Any, representation: Any) -> Any:
+    """A copy of ``period`` that holds only this AdaptationSet, cut to ``representation`` when there is one."""
+    new_period = etree.Element("Period", dict(period.attrib) if period is not None else {})
+    for child in period if period is not None else []:
+        if child.tag != "AdaptationSet":
+            new_period.append(deepcopy(child))
+    new_adaptation_set = deepcopy(adaptation_set)
+    if representation is not None:
+        for sibling in new_adaptation_set.findall("Representation"):
+            new_adaptation_set.remove(sibling)
+        new_adaptation_set.append(deepcopy(representation))
+    new_period.append(new_adaptation_set)
+    return new_period
+
+
+def fragments(tracks: Tracks, level: int, counts: dict[int, int]) -> list[dict[str, Any]]:
+    """A one-track MPD for each track the client cannot find again in its whole manifest.
+
+    A service can build an AdaptationSet of its own that never joins the manifest tree, or a
+    manifest that holds only its own track, which an attribute match cannot tell apart. The
+    fragment starts at the track's period and keeps the track's part of each content period
+    after it, so the download still stitches them; the filtered periods are left out.
+    """
+    out: list[dict[str, Any]] = []
+    for track in tracks:
+        dash = track.data.get("dash") or {}
+        manifest = dash.get("manifest")
+        adaptation_set = dash.get("adaptation_set")
+        if manifest is None or adaptation_set is None:
+            continue
+        if adaptation_set.getroottree().getroot() is manifest and counted(manifest, ".//Representation", counts) > 1:
+            continue
+
+        root = etree.Element("MPD", dict(manifest.attrib))
+        for child in manifest:
+            if child.tag != "Period":
+                root.append(deepcopy(child))
+        period = dash.get("period")
+        representation = dash.get("representation")
+        root.append(one_track_period(period, adaptation_set, representation))
+
+        periods = manifest.findall("Period")
+        if period is not None and period in periods:
+            filtered = set(dash.get("filtered_period_ids") or [])
+            rep_id = dash.get("representation_id") or (representation.get("id") if representation is not None else None)
+            for later in periods[periods.index(period) + 1 :]:
+                if not DASH.is_content_period(later, filtered):
+                    continue
+                found = DASH.resolve_representation(later, rep_id, adaptation_set)
+                if found is not None:
+                    root.append(one_track_period(later, *found))
+
+        out.append(
+            {
+                "track_id": str(track.id),
+                "type": "dash",
+                "url": str(track.url) if track.url else None,
+                "data": pack(root, level),
+            }
+        )
+    return out
+
+
+def restore_fragments(tracks: Tracks, saved: list[dict[str, Any]]) -> None:
+    """Fill ``track.data`` from the one-track fragments, by track id. This runs before the whole-manifest restore."""
+    by_id = {str(t.id): t for t in tracks}
+    for entry in saved:
+        track = by_id.get(str(entry.get("track_id")))
+        if track is None or entry.get("type") != "dash" or not isinstance(entry.get("data"), str):
+            continue
+        try:
+            manifest = unpack(entry["data"])
+            period = manifest.find("Period")
+            adaptation_set = period.find("AdaptationSet")
+            representation = adaptation_set.find("Representation")
+            if representation is None:
+                raise ValueError("fragment has no Representation")
+        except Exception as e:
+            log.warning("Failed to re-parse the track fragment for track %s: %s", track.id, e)
+            continue
+
+        track.data["dash"] = {
+            "manifest": manifest,
+            "period": period,
+            "adaptation_set": adaptation_set,
+            "representation": representation,
+            "representation_id": representation.get("id"),
+        }
+        track.descriptor = Track.Descriptor.DASH
+
+
+def to_tracks(entry: dict[str, Any], language: Any) -> Optional[Tracks]:
+    """The tracks of one whole-manifest entry, with the service's filtered periods dropped again."""
+    root = unpack(entry["data"])
+    if entry.get("type") == "dash":
+        skip = set(entry.get("filtered_period_ids") or [])
+        return DASH(root, entry["url"]).to_tracks(
+            language=language, period_filter=(lambda p: p.get("id") in skip) if skip else None
+        )
+    if entry.get("type") == "ism":
+        return ISM(root, entry["url"]).to_tracks(language=language)
+    return None
+
+
+def has_data(track: Any) -> bool:
+    return any(track.data.get(key) for key in ("dash", "ism"))
+
+
+def restore_manifests(tracks: Tracks, saved: list[dict[str, Any]], fallback: Any = None) -> None:
+    """Re-parse each whole manifest and copy ``data`` to the tracks that match, by id or by attributes."""
+    targets = [t for t in tracks if not has_data(t)]
+    if not targets:
+        return
+    language = next((t.language for t in tracks if t.language and str(t.language) != "und"), fallback)
+    parsed: list[Any] = []
+    for entry in saved:
+        if not isinstance(entry.get("data"), str) or not isinstance(entry.get("url"), str):
+            continue
+        try:
+            local = to_tracks(entry, language)
+        except Exception as e:
+            log.warning("Failed to re-parse %s manifest from %s: %s", entry.get("type"), entry.get("url"), e)
+            continue
+        if local is not None:
+            parsed += list(local)
+
+    for track in targets:
+        matched = match_track(track, parsed)
+        if matched and has_data(matched):
+            track.data.update(matched.data)
+            track.descriptor = matched.descriptor
+            if matched.drm and not track.drm:
+                track.drm = matched.drm
+
+
+def same_bitrate(local: Optional[int], remote: Optional[int]) -> bool:
+    """Equal to the nearest kb/s. A server that predates ``bitrate_bps`` rounds to kb/s in transit."""
+    if not local or not remote:
+        return not local and not remote
+    return round(local / 1000) == round(remote / 1000)
+
+
+def match_track(remote_track: Track, local_tracks: list) -> Optional[Track]:
+    """Find the locally-parsed track that matches a remote track by ID or by attributes.
+
+    HDR10 matches HDR10P, because the server's HDR10+ probe relabels a track that the local re-parse calls HDR10.
+    """
+    hdr10_ranges = {Video.Range.HDR10, Video.Range.HDR10P}
+    remote_id = str(remote_track.id)
+    for lt in local_tracks:
+        if str(lt.id) == remote_id:
+            return lt
+
+    if not isinstance(remote_track, (Video, Audio, Subtitle)):
+        return None
+    for lt in local_tracks:
+        if type(lt).__name__ != type(remote_track).__name__:
+            continue
+        if lt.codec != remote_track.codec or str(lt.language) != str(remote_track.language):
+            continue
+        if isinstance(remote_track, Video):
+            if (
+                lt.width == remote_track.width
+                and lt.height == remote_track.height
+                and (lt.range == remote_track.range or {lt.range, remote_track.range} == hdr10_ranges)
+                and same_bitrate(lt.bitrate, remote_track.bitrate)
+            ):
+                return lt
+        elif isinstance(remote_track, Audio):
+            if same_bitrate(lt.bitrate, remote_track.bitrate):
+                return lt
+        elif isinstance(remote_track, Subtitle):
+            if lt.forced == remote_track.forced and lt.sdh == remote_track.sdh:
+                return lt
+    return None

unshackle/core/remote_service.py

@@ -41,6 +41,7 @@ from unshackle.core.console import console, prompt_user, show_notice
 from unshackle.core.constants import AnyTrack
 from unshackle.core.credential import Credential
 from unshackle.core.drm import BlowfishStripe, cbcs, real_kids
+from unshackle.core.manifests.snapshot import match_track, restore
 from unshackle.core.proxies.resolve import is_loopback
 from unshackle.core.search_result import SearchResult
 from unshackle.core.session import MAX_BACKOFF
@@ -473,120 +474,6 @@ def apply_session_cookies(session: requests.Session, result: dict) -> None:
             )
 
 
-def resolve_track_manifests(tracks: Tracks, track_manifests: list) -> None:
-    """Fill track.data from the per-track MPD fragments the server sent.
-
-    A service can build an AdaptationSet the served manifest does not contain, so the
-    client's re-parse of that manifest cannot find it. The server sends a one-AdaptationSet
-    MPD for each of those tracks instead. This runs first, and the whole-manifest re-parse
-    then skips every track whose ``data`` already holds a ``dash`` entry.
-    """
-    import base64 as b64
-
-    from lxml import etree
-
-    from unshackle.core.api.compression import safe_inflate
-
-    if not track_manifests:
-        return
-
-    log_m = logging.getLogger("remote_service")
-    by_id = {str(t.id): t for t in list(tracks.videos) + list(tracks.audio) + list(tracks.subtitles)}
-
-    for entry in track_manifests:
-        track = by_id.get(str(entry.get("track_id")))
-        if track is None or entry.get("type") != "dash" or not entry.get("data"):
-            continue
-        try:
-            manifest = etree.fromstring(safe_inflate(b64.b64decode(entry["data"])))
-            period = manifest.find("Period")
-            adaptation_set = period.find("AdaptationSet")
-            representation = adaptation_set.find("Representation")
-            if representation is None:
-                raise ValueError("fragment has no Representation")
-        except Exception as e:
-            log_m.warning("Failed to re-parse the track fragment for track %s: %s", track.id, e)
-            continue
-
-        track.data["dash"] = {
-            "manifest": manifest,
-            "period": period,
-            "adaptation_set": adaptation_set,
-            "representation": representation,
-            "representation_id": representation.get("id"),
-        }
-        track.descriptor = Track.Descriptor.DASH
-
-
-def resolve_manifest_data(tracks: Tracks, manifests: list) -> None:
-    """Re-parse serialized manifests and fill track.data for downloading.
-
-    The server serializes DASH and ISM manifest XML as zlib-compressed base64.
-    We decode and decompress locally, re-parse with the appropriate manifest
-    parser, then match each remote track to the locally-parsed track by ID
-    to copy track.data. We skip HLS, because it re-fetches from track.url.
-    """
-    import base64 as b64
-
-    from unshackle.core.api.compression import safe_inflate
-
-    if not manifests:
-        return
-
-    log_m = logging.getLogger("remote_service")
-    all_tracks = list(tracks.videos) + list(tracks.audio) + list(tracks.subtitles)
-    local_all: list = []
-
-    for manifest_info in manifests:
-        m_type = manifest_info.get("type")
-        m_url = manifest_info.get("url")
-        m_data = manifest_info.get("data")
-        if not m_data or not m_url:
-            continue
-
-        try:
-            raw = safe_inflate(b64.b64decode(m_data))
-
-            fallback_lang = next(
-                (t.language for t in all_tracks if t.language and str(t.language) != "und"),
-                None,
-            )
-            if m_type == "dash":
-                from lxml import etree
-
-                from unshackle.core.manifests import DASH
-
-                skip = manifest_info.get("filtered_period_ids") or []
-                local_tracks = DASH(etree.fromstring(raw), m_url).to_tracks(
-                    language=fallback_lang, period_filter=(lambda p: p.get("id") in skip) if skip else None
-                )
-                if skip:
-                    for local_track in local_tracks:
-                        local_track.data["dash"]["filtered_period_ids"] = skip
-            elif m_type == "ism":
-                from lxml import etree
-
-                from unshackle.core.manifests import ISM
-
-                local_tracks = ISM(etree.fromstring(raw), m_url).to_tracks(language=fallback_lang)
-            else:
-                continue
-
-            local_all += list(local_tracks.videos) + list(local_tracks.audio) + list(local_tracks.subtitles)
-        except Exception as e:
-            log_m.warning("Failed to re-parse %s manifest from %s: %s", m_type, m_url, e)
-
-    for remote_track in all_tracks:
-        if any(remote_track.data.get(key) for key in ("dash", "ism")):
-            continue
-        matched = match_track(remote_track, local_all)
-        if matched and any(matched.data.get(key) for key in ("dash", "ism")):
-            remote_track.data.update(matched.data)
-            remote_track.descriptor = matched.descriptor
-            if matched.drm and not remote_track.drm:
-                remote_track.drm = matched.drm
-
-
 def segment_uri_filter(uris: Iterable[Any]) -> Callable[[Any], bool]:
     """Return an HLS ``OnSegmentFilter`` that drops the segments with these URIs.
 
@@ -603,48 +490,6 @@ def segment_uri_filter(uris: Iterable[Any]) -> Callable[[Any], bool]:
     return segment_filter
 
 
-def same_bitrate(local: Optional[int], remote: Optional[int]) -> bool:
-    """Equal to the nearest kb/s. A server that predates ``bitrate_bps`` rounds to kb/s in transit."""
-    if not local or not remote:
-        return not local and not remote
-    return round(local / 1000) == round(remote / 1000)
-
-
-def match_track(remote_track: Track, local_tracks: list) -> Optional[Track]:
-    """Find the locally-parsed track that matches a remote track by ID or by attributes.
-
-    HDR10 matches HDR10P, because the server's HDR10+ probe relabels a track that the local re-parse calls HDR10.
-    """
-    hdr10_ranges = {Video.Range.HDR10, Video.Range.HDR10P}
-    remote_id = str(remote_track.id)
-    for lt in local_tracks:
-        if str(lt.id) == remote_id:
-            return lt
-
-    if not isinstance(remote_track, (Video, Audio, Subtitle)):
-        return None
-    for lt in local_tracks:
-        if type(lt).__name__ != type(remote_track).__name__:
-            continue
-        if lt.codec != remote_track.codec or str(lt.language) != str(remote_track.language):
-            continue
-        if isinstance(remote_track, Video):
-            if (
-                lt.width == remote_track.width
-                and lt.height == remote_track.height
-                and (lt.range == remote_track.range or {lt.range, remote_track.range} == hdr10_ranges)
-                and same_bitrate(lt.bitrate, remote_track.bitrate)
-            ):
-                return lt
-        elif isinstance(remote_track, Audio):
-            if same_bitrate(lt.bitrate, remote_track.bitrate):
-                return lt
-        elif isinstance(remote_track, Subtitle):
-            if lt.forced == remote_track.forced and lt.sdh == remote_track.sdh:
-                return lt
-    return None
-
-
 EPISODE_PLACEHOLDER_NAME = re.compile(r"Episode ?#?\d+", re.IGNORECASE)
 
 
@@ -1847,8 +1692,7 @@ class RemoteService:
                 self._session.headers[k] = v
         apply_session_cookies(self._session, result)
 
-        resolve_track_manifests(tracks, result.get("track_manifests", []))
-        resolve_manifest_data(tracks, result.get("manifests", []))
+        restore(tracks, result)
         apply_service_track_data(tracks, result)
 
         server_cdm = bool(result.get("server_cdm", False)) or getattr(self, "_server_device", False)