From 619c86cd105923a0c0a9d50302d71c95831d940d Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Sun, 16 Aug 2026 19:38:06 -0600 Subject: [PATCH] feat(downloader): resume failed downloads from completed segments across runs Fingerprint-gated reuse of atomically-published segment files (DASH/ISM/HLS) behind a continue_downloads config option and one-off dl --continue-downloads flag. --- docs/guide/cli-reference.md | 1 + docs/guide/downloading.md | 1 + docs/reference/configuration/download.md | 28 ++++ docs/reference/configuration/index.md | 2 +- tests/core/test_resume_fingerprint.py | 86 +++++++++++++ tests/core/test_resume_hls_integration.py | 139 ++++++++++++++++++++ tests/core/test_resume_integration.py | 149 ++++++++++++++++++++++ tests/core/test_resume_track_lifecycle.py | 109 ++++++++++++++++ unshackle/commands/dl.py | 12 ++ unshackle/core/api/download_manager.py | 1 + unshackle/core/api/handlers.py | 1 + unshackle/core/api/routes.py | 3 + unshackle/core/config.py | 1 + unshackle/core/manifests/dash.py | 13 +- unshackle/core/manifests/hls.py | 22 +++- unshackle/core/manifests/ism.py | 12 +- unshackle/core/tracks/resume.py | 84 ++++++++++++ unshackle/core/tracks/track.py | 22 +++- 18 files changed, 675 insertions(+), 11 deletions(-) create mode 100644 tests/core/test_resume_fingerprint.py create mode 100644 tests/core/test_resume_hls_integration.py create mode 100644 tests/core/test_resume_integration.py create mode 100644 tests/core/test_resume_track_lifecycle.py create mode 100644 unshackle/core/tracks/resume.py diff --git a/docs/guide/cli-reference.md b/docs/guide/cli-reference.md index 4560c244..58eed11b 100644 --- a/docs/guide/cli-reference.md +++ b/docs/guide/cli-reference.md @@ -202,6 +202,7 @@ order the service used, then renumbers the episodes into the order you asked for | `--workers` | downloader default | Per-track download threads. | | `--adaptive-workers` | off | Scale per-track segment workers dynamically (up to the `--workers` cap) based on measured CDN throughput and errors. | | `--download-processes` | `1` | Split large segment batches (24+) across this many download processes to exceed the single-process throughput cap on multi-gigabit connections. Ignored while a speed limit is set: the cap is one shared budget, which extra processes cannot share, so the download stays in a single process. | +| `--continue-downloads` | off | Keep completed segment files across runs and resume a previously failed download. One-off enable of the [`continue_downloads`](../reference/configuration/download.md#continue_downloads) config option. | | `--downloads` | `1` | Number of tracks downloaded concurrently. | | `--speed-limit` | unlimited | Cap total download speed across all threads and tracks, e.g. `500k`, `5M`, `1.5G` or plain bytes/sec. Values are bytes, not bits (`5M` = 5.0 MB/s). `off` disables a configured limit. | | `--no-cache` | off | Bypass the title cache. | diff --git a/docs/guide/downloading.md b/docs/guide/downloading.md index f6a41145..e062e412 100644 --- a/docs/guide/downloading.md +++ b/docs/guide/downloading.md @@ -613,6 +613,7 @@ Two related flags: | `--workers N` | Threads used per track for segment downloads. Default depends on the downloader. | | `--adaptive-workers` | Opt-in: start with a moderate per-track worker count and ramp it up or back off based on measured CDN throughput and errors, capped at `--workers`. Off by default (fixed worker count). | | `--download-processes N` | Opt-in: split a large segment batch across `N` download processes, each with its own worker pool. A single process tops out around 1.3 Gb/s (Python interpreter limit); `2` reaches 2.5GbE line rate on fast CDNs. Only engages for batches of 24+ segments. Ignored while `--speed-limit` (or serve's `global_speed_limit`) is set: the cap is one shared budget, which extra processes cannot share, so the download stays in a single process. Default `1`. | +| `--continue-downloads` | Opt-in: keep completed segment files when a download fails so the next run resumes instead of restarting from zero. One-off enable of the [`continue_downloads`](../reference/configuration/download.md#continue_downloads) config option, which documents what can and cannot resume. | | `--downloads N` | Number of tracks downloaded concurrently. Default `1`. | | `--slow [MIN-MAX]` | Add a delay between titles to look more like a real device. `--slow` alone means 60-120s; `--slow 20-40` sets a custom range. Minimum 20s. | | `--no-cache` | Bypass the title cache for this download. | diff --git a/docs/reference/configuration/download.md b/docs/reference/configuration/download.md index c66547c0..858a55f8 100644 --- a/docs/reference/configuration/download.md +++ b/docs/reference/configuration/download.md @@ -92,6 +92,34 @@ dl: to set server-wide download defaults for every request it handles. Keys the API does not know are ignored there (`speed_limit`, for example; `serve` has its own `global_speed_limit`). +## `continue_downloads` + +- **Type:** `bool`  ·  **Default:** `false` + +Keep a failed download's completed segment files so the next attempt resumes from where it +stopped instead of restarting from byte zero. + +Only whole, fully downloaded segments are reused; partially written segments are always +discarded, and no partial-byte (`Range`) resume across runs takes place. Reuse also +requires proof that the remote content is unchanged: the manifest URL, segment count, and +byte ranges must all match the previous run, otherwise the download silently restarts +clean. + +!!! note "What can and cannot resume" + - Segmented DASH, ISM, and HLS tracks resume. + - Single-file (non-segmented) downloads and DASH tracks collapsed to a single ranged + download never produce segment files, so this option does not apply to them (the + downloader has its own in-flight partial-file handling for those). + - HLS resumes Widevine/PlayReady and unencrypted content only; AES-128 content always + restarts. An HLS failure during the post-download merge step also restarts. + +For a one-off resume without changing config, pass `dl --continue-downloads` on the +command line (the flag can only enable resume, never disable it). + +```yaml title="Resume failed downloads" +continue_downloads: true +``` + ## `subtitle` - **Type:** `dict`  ·  **Default:** `{}` diff --git a/docs/reference/configuration/index.md b/docs/reference/configuration/index.md index eff698ef..35dc3bbc 100644 --- a/docs/reference/configuration/index.md +++ b/docs/reference/configuration/index.md @@ -31,7 +31,7 @@ Where the file lives, how unshackle discovers it, and how to edit it: |------|------| | [Directories](directories.md) | `directories`, `filenames` | | [Services & authentication](services.md) | `services`, `credentials`, `firefox_cookies`, `remote_services`, `serve` | -| [Download & processing](download.md) | `dl`, `subtitle`, `audio`, `muxing`, `language_tags`, `dual_multi_mode` | +| [Download & processing](download.md) | `dl`, `continue_downloads`, `subtitle`, `audio`, `muxing`, `language_tags`, `dual_multi_mode` | | [Output & naming](output.md) | `output_template`, `tag`, `tag_rules`, `tag_group_name`, `tag_imdb_tmdb`, `chapter_fallback_name`, `unicode_filenames` | | [DRM & CDM](drm.md) | `cdm`, `remote_cdm`, `decryption` | | [Network & proxy](network.md) | `network`, `headers`, `proxy_providers` | diff --git a/tests/core/test_resume_fingerprint.py b/tests/core/test_resume_fingerprint.py new file mode 100644 index 00000000..ec81540a --- /dev/null +++ b/tests/core/test_resume_fingerprint.py @@ -0,0 +1,86 @@ +"""Offline unit tests for the cross-run segment-resume sidecar: fingerprint stability +across rotating query tokens, sensitivity to real segmentation changes, sidecar +read/write robustness, and the sidecar-outside-save_dir invariant that protects the +merge glob from concatenating it into the output.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from unshackle.core.tracks import resume + +MPD = "https://cdn/manifest.mpd" +SEGMENTS = [ + ("https://cdn/media.mp4", "863-1999"), + ("https://cdn/media.mp4", "2000-3999"), + ("https://cdn/media.mp4", "4000-5999"), +] + + +def test_identical_inputs_identical_digest(): + assert resume.fingerprint(MPD, SEGMENTS) == resume.fingerprint(MPD, SEGMENTS) + + +def test_rotating_query_token_ignored(): + tokenised = [(f"{url}?token=abc123&expires=999", rng) for url, rng in SEGMENTS] + assert resume.fingerprint(f"{MPD}?auth=xyz", tokenised) == resume.fingerprint(MPD, SEGMENTS) + + +def test_changed_byte_range_changes_digest(): + changed = [SEGMENTS[0], ("https://cdn/media.mp4", "2000-4000"), SEGMENTS[2]] + assert resume.fingerprint(MPD, changed) != resume.fingerprint(MPD, SEGMENTS) + + +def test_changed_segment_count_changes_digest(): + assert resume.fingerprint(MPD, SEGMENTS[:2]) != resume.fingerprint(MPD, SEGMENTS) + + +def test_changed_url_path_changes_digest(): + moved = [("https://cdn/other.mp4", rng) for _, rng in SEGMENTS] + assert resume.fingerprint(MPD, moved) != resume.fingerprint(MPD, SEGMENTS) + + +def test_query_carried_identity_is_not_collapsed(): + # all segments share a path and differ only in the query (e.g. $Time$ as a query + # param): stripping must not collapse them into false matches; the digest goes + # verbatim and distinct segmentations stay distinct + a = [("https://cdn/seg?t=0", None), ("https://cdn/seg?t=100", None)] + b = [("https://cdn/seg?t=0", None), ("https://cdn/seg?t=200", None)] + assert resume.fingerprint(MPD, a) != resume.fingerprint(MPD, b) + + +def test_extra_inputs_change_digest(): + assert resume.fingerprint(MPD, SEGMENTS, extra=["100", "3"]) != resume.fingerprint(MPD, SEGMENTS) + + +def test_reusable_roundtrip(tmp_path: Path): + save_dir = tmp_path / "Video_x_segments" + digest = resume.fingerprint(MPD, SEGMENTS) + resume.write_sidecar(save_dir, digest) + assert resume.reusable(save_dir, digest) is True + assert resume.reusable(save_dir, "0" * 64) is False + + +def test_reusable_never_raises(tmp_path: Path): + save_dir = tmp_path / "Video_x_segments" + digest = resume.fingerprint(MPD, SEGMENTS) + assert resume.reusable(save_dir, digest) is False # no sidecar + resume.sidecar_path(save_dir).write_text("{not json", encoding="utf-8") + assert resume.reusable(save_dir, digest) is False # malformed + resume.sidecar_path(save_dir).write_text(json.dumps({"version": 1}), encoding="utf-8") + assert resume.reusable(save_dir, digest) is False # no digest key + + +def test_clear_sidecar(tmp_path: Path): + save_dir = tmp_path / "Video_x_segments" + digest = resume.fingerprint(MPD, SEGMENTS) + resume.write_sidecar(save_dir, digest) + resume.clear_sidecar(save_dir) + assert resume.reusable(save_dir, digest) is False + resume.clear_sidecar(save_dir) # idempotent on a missing file + + +def test_sidecar_is_never_inside_save_dir(tmp_path: Path): + save_dir = tmp_path / "Video_x_segments" + assert resume.sidecar_path(save_dir).parent == save_dir.parent diff --git a/tests/core/test_resume_hls_integration.py b/tests/core/test_resume_hls_integration.py new file mode 100644 index 00000000..0a2ae7b2 --- /dev/null +++ b/tests/core/test_resume_hls_integration.py @@ -0,0 +1,139 @@ +"""Integration coverage for cross-run segment resume on the HLS path. + +Same two-run shape as the DASH test, plus the HLS-specific rules: the sidecar (the +"download phase completed cleanly" proof) survives a mid-download crash and is gone once +the merge pass runs, and AES-128 content is excluded from resume entirely because its +segments are decrypted in place during the merge pass. +""" + +from __future__ import annotations + +import importlib +import threading +from functools import partial +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import pytest + +from unshackle.core.config import config +from unshackle.core.constants import DOWNLOAD_CANCELLED +from unshackle.core.manifests import HLS +from unshackle.core.tracks import resume +from unshackle.core.tracks.track import DownloadContext + +PART0 = bytes([0xA0]) * 300 +PART1 = bytes([0xB1]) * 200 +AES_KEY = bytes(range(16)) + + +class _Handler(BaseHTTPRequestHandler): + def do_GET(self): + server = self.server + with server.lock: + server.requests.append(self.path) + body = server.routes.get(self.path) + if body is None: + self.send_response(404) + self.send_header("Content-Length", "0") + self.end_headers() + return + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): + pass + + +@pytest.fixture +def server(): + srv = ThreadingHTTPServer(("127.0.0.1", 0), _Handler) + srv.lock = threading.Lock() + srv.requests = [] + srv.routes = {} + thread = threading.Thread(target=srv.serve_forever, daemon=True) + thread.start() + yield srv + srv.shutdown() + thread.join(timeout=5) + + +@pytest.fixture(autouse=True) +def resume_enabled(monkeypatch): + monkeypatch.setattr(config, "continue_downloads", True) + monkeypatch.setattr(importlib.import_module("unshackle.core.downloaders.requests"), "MAX_ATTEMPTS", 1) + DOWNLOAD_CANCELLED.clear() + yield + DOWNLOAD_CANCELLED.clear() + + +def base(srv) -> str: + return f"http://127.0.0.1:{srv.server_address[1]}/" + + +MASTER = """#EXTM3U +#EXT-X-STREAM-INF:BANDWIDTH=1000000,RESOLUTION=1920x1080,CODECS="avc1.640028" +media.m3u8 +""" + + +def media_playlist(key_line: str = "") -> bytes: + return f"""#EXTM3U +#EXT-X-VERSION:3 +#EXT-X-TARGETDURATION:4 +{key_line}#EXTINF:4.0, +part0.bin +#EXTINF:4.0, +part1.bin +#EXT-X-ENDLIST +""".encode() + + +def make_track(srv): + return HLS.from_text(MASTER, url=f"{base(srv)}master.m3u8").to_tracks(language="en").videos[0] + + +def make_ctx(tmp_path): + return DownloadContext( + save_path=tmp_path / "out.mp4", + save_dir=tmp_path / "v0_segments", + progress=partial(lambda **kw: None), + max_workers=1, + ) + + +def test_hls_second_run_reuses_completed_segments(server, tmp_path): + server.routes["/media.m3u8"] = media_playlist() + server.routes["/part0.bin"] = PART0 + # part1.bin missing -> run 1 completes part0 then dies during the download phase + + ctx = make_ctx(tmp_path) + with pytest.raises(Exception): + HLS.download_track(make_track(server), ctx) + + assert (ctx.save_dir / "segments" / "0.bin").read_bytes() == PART0 + assert resume.sidecar_path(ctx.save_dir).exists() + part0_fetches_run1 = server.requests.count("/part0.bin") + + DOWNLOAD_CANCELLED.clear() + server.routes["/part1.bin"] = PART1 + HLS.download_track(make_track(server), ctx) + + assert ctx.save_path.read_bytes() == PART0 + PART1 + assert server.requests.count("/part0.bin") == part0_fetches_run1 + assert not resume.sidecar_path(ctx.save_dir).exists() + + +def test_hls_aes128_content_never_writes_sidecar(server, tmp_path): + key_line = f'#EXT-X-KEY:METHOD=AES-128,URI="{base(server)}key.bin"\n' + server.routes["/media.m3u8"] = media_playlist(key_line) + server.routes["/key.bin"] = AES_KEY + server.routes["/part0.bin"] = PART0 + # part1.bin missing -> crash mid-download, the state resume would normally preserve + + ctx = make_ctx(tmp_path) + with pytest.raises(Exception): + HLS.download_track(make_track(server), ctx) + + assert not resume.sidecar_path(ctx.save_dir).exists() diff --git a/tests/core/test_resume_integration.py b/tests/core/test_resume_integration.py new file mode 100644 index 00000000..a3832cab --- /dev/null +++ b/tests/core/test_resume_integration.py @@ -0,0 +1,149 @@ +"""Integration coverage for cross-run segment resume on the segmented DASH path. + +Drives the real ``DASH.download_track`` twice against a localhost server with +``continue_downloads`` on: the first run dies on a missing segment (after completing the +other one), the second run finds the sidecar digest unchanged and must reuse the completed +segment file instead of re-fetching it. A third scenario changes the segmentation and +asserts the stale directory is wiped rather than mixed into the new download. +""" + +from __future__ import annotations + +import importlib +import threading +from functools import partial +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import pytest + +from unshackle.core.config import config +from unshackle.core.constants import DOWNLOAD_CANCELLED +from unshackle.core.manifests import DASH +from unshackle.core.tracks import resume +from unshackle.core.tracks.track import DownloadContext + +INIT = bytes(range(100)) +PART0 = bytes([0xA0]) * 300 +PART1 = bytes([0xB1]) * 200 + + +class _Handler(BaseHTTPRequestHandler): + def do_GET(self): + server = self.server + with server.lock: + server.requests.append(self.path) + body = server.routes.get(self.path) + if body is None: + self.send_response(404) + self.send_header("Content-Length", "0") + self.end_headers() + return + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): + pass + + +@pytest.fixture +def server(): + srv = ThreadingHTTPServer(("127.0.0.1", 0), _Handler) + srv.lock = threading.Lock() + srv.requests = [] + srv.routes = {} + thread = threading.Thread(target=srv.serve_forever, daemon=True) + thread.start() + yield srv + srv.shutdown() + thread.join(timeout=5) + + +@pytest.fixture(autouse=True) +def resume_enabled(monkeypatch): + monkeypatch.setattr(config, "continue_downloads", True) + # fail fast on the deliberately-missing segment instead of the full retry ladder + # (import via importlib: the downloaders package exports a same-named function) + monkeypatch.setattr(importlib.import_module("unshackle.core.downloaders.requests"), "MAX_ATTEMPTS", 1) + DOWNLOAD_CANCELLED.clear() + yield + DOWNLOAD_CANCELLED.clear() + + +def base(srv) -> str: + return f"http://127.0.0.1:{srv.server_address[1]}/" + + +def mpd(srv, part1: str = "part1.bin") -> str: + return f""" + + {base(srv)} + + + + + + + + + + + +""" + + +def make_track(text: str, srv): + return DASH.from_text(text, url=f"{base(srv)}manifest.mpd").to_tracks(language="en").videos[0] + + +def make_ctx(tmp_path): + return DownloadContext( + save_path=tmp_path / "out.mp4", + save_dir=tmp_path / "v0_segments", + progress=partial(lambda **kw: None), + # sequential: part0 must be complete before the deliberate part1 failure aborts run 1 + max_workers=1, + ) + + +def test_second_run_reuses_completed_segments(server, tmp_path): + server.routes["/init.bin"] = INIT + server.routes["/part0.bin"] = PART0 + # part1.bin is missing -> the first run completes part0 then dies on part1 + + ctx = make_ctx(tmp_path) + with pytest.raises(Exception): + DASH.download_track(make_track(mpd(server), server), ctx) + + assert (ctx.save_dir / "0.mp4").read_bytes() == PART0 + assert resume.sidecar_path(ctx.save_dir).exists() + part0_fetches_run1 = server.requests.count("/part0.bin") + + DOWNLOAD_CANCELLED.clear() + server.routes["/part1.bin"] = PART1 + DASH.download_track(make_track(mpd(server), server), ctx) + + assert ctx.save_path.read_bytes() == INIT + PART0 + PART1 + assert server.requests.count("/part0.bin") == part0_fetches_run1 + assert not resume.sidecar_path(ctx.save_dir).exists() + assert not ctx.save_dir.exists() + + +def test_changed_segmentation_wipes_stale_segments(server, tmp_path): + server.routes["/init.bin"] = INIT + server.routes["/part0.bin"] = PART0 + + ctx = make_ctx(tmp_path) + with pytest.raises(Exception): + DASH.download_track(make_track(mpd(server), server), ctx) + assert (ctx.save_dir / "0.mp4").exists() + + DOWNLOAD_CANCELLED.clear() + # different final segment path -> different digest -> stale part0 must be re-fetched + server.routes["/part2.bin"] = PART1 + part0_fetches_run1 = server.requests.count("/part0.bin") + DASH.download_track(make_track(mpd(server, part1="part2.bin"), server), ctx) + + assert ctx.save_path.read_bytes() == INIT + PART0 + PART1 + assert server.requests.count("/part0.bin") == part0_fetches_run1 + 1 diff --git a/tests/core/test_resume_track_lifecycle.py b/tests/core/test_resume_track_lifecycle.py new file mode 100644 index 00000000..ecfe2af0 --- /dev/null +++ b/tests/core/test_resume_track_lifecycle.py @@ -0,0 +1,109 @@ +"""Lifecycle tests for the conditional wipe in Track.download itself. + +The manifest-parser gates are covered by the DASH/HLS integration tests; these cover the +layer above them: with continue_downloads on, Track.download's entry sweep and its +exception handler must both preserve completed segments and the sidecar while purging +every partial, and with the flag off both must wipe the directory entirely. The manifest +parser is stubbed so only Track.download's own behaviour is under test. +""" + +from __future__ import annotations + +from functools import partial +from pathlib import Path + +import pytest + +from unshackle.core.config import config +from unshackle.core.constants import DOWNLOAD_CANCELLED, DOWNLOAD_LICENCE_ONLY +from unshackle.core.tracks import Video, resume + + +@pytest.fixture(autouse=True) +def clean_flags(tmp_path, monkeypatch): + monkeypatch.setattr(config.directories, "temp", tmp_path) + DOWNLOAD_CANCELLED.clear() + DOWNLOAD_LICENCE_ONLY.clear() + yield + DOWNLOAD_CANCELLED.clear() + + +def make_track() -> Video: + return Video( + id_="t1", + url="https://example.test/manifest.mpd", + language="en", + codec=Video.Codec.AVC, + range_=Video.Range.SDR, + width=1920, + height=1080, + bitrate=5_000_000, + descriptor=Video.Descriptor.DASH, + ) + + +def seed_prior_run(tmp_path: Path) -> Path: + save_dir = tmp_path / "Video_t1.mp4_segments" + (save_dir / "segments").mkdir(parents=True) + (save_dir / "0.mp4").write_bytes(b"complete segment") + (save_dir / "1.mp4.p.!dev").write_bytes(b"partial") + (save_dir / "segments" / "2.bin.h.!dev").write_bytes(b"hedge partial") + resume.write_sidecar(save_dir, "f" * 64) + return save_dir + + +def run_download(track: Video, monkeypatch, fail: bool): + import requests as requests_lib + + from unshackle.core.manifests import DASH + + def stub_download_track(track, ctx): + if fail: + raise RuntimeError("simulated mid-download crash") + ctx.save_path.write_bytes(b"output") + track.path = ctx.save_path + + monkeypatch.setattr(DASH, "download_track", staticmethod(stub_download_track)) + track.download( + session=requests_lib.Session(), + prepare_drm=partial(lambda *a, **kw: None), + progress=partial(lambda **kw: None), + ) + + +def test_flag_on_keeps_segments_and_sidecar_sweeps_partials(tmp_path, monkeypatch): + monkeypatch.setattr(config, "continue_downloads", True) + save_dir = seed_prior_run(tmp_path) + track = make_track() + + with pytest.raises(RuntimeError): + run_download(track, monkeypatch, fail=True) + + assert (save_dir / "0.mp4").read_bytes() == b"complete segment" + assert resume.sidecar_path(save_dir).exists() + assert not list(save_dir.rglob("*.!dev")) + + +def test_flag_off_wipes_segments_and_sidecar(tmp_path, monkeypatch): + monkeypatch.setattr(config, "continue_downloads", False) + save_dir = seed_prior_run(tmp_path) + track = make_track() + + with pytest.raises(RuntimeError): + run_download(track, monkeypatch, fail=True) + + assert not save_dir.exists() + assert not resume.sidecar_path(save_dir).exists() + + +def test_flag_off_clears_orphaned_sidecar_without_dir(tmp_path, monkeypatch): + monkeypatch.setattr(config, "continue_downloads", False) + save_dir = tmp_path / "Video_t1.mp4_segments" + # a kill between a parser's dir removal and its sidecar clear leaves exactly this + resume.write_sidecar(save_dir, "f" * 64) + track = make_track() + + with pytest.raises(RuntimeError): + run_download(track, monkeypatch, fail=True) + + assert not resume.sidecar_path(save_dir).exists() diff --git a/unshackle/commands/dl.py b/unshackle/commands/dl.py index f4466439..0960d693 100644 --- a/unshackle/commands/dl.py +++ b/unshackle/commands/dl.py @@ -819,6 +819,15 @@ class dl: "throughput cap. Only engages for large segment batches. Default 1 (single process)." ), ) + @click.option( + "--continue-downloads", + is_flag=True, + default=False, + help=( + "Keep completed segment files across runs and resume a previously failed download. " + "One-off enable of the continue_downloads config option." + ), + ) @click.option("--downloads", type=int, default=1, help="Amount of tracks to download concurrently.") @click.option( "--speed-limit", @@ -1408,6 +1417,7 @@ class dl: workers: Optional[int], adaptive_workers: bool, download_processes: int, + continue_downloads: bool, downloads: int, worst: bool, best_available: bool, @@ -1420,6 +1430,8 @@ class dl: *_: Any, **__: Any, ) -> None: + if continue_downloads: + config.continue_downloads = True self.tmdb_searched = False self.search_source = None self.service_anime = bool(getattr(service, "ANIME", False)) diff --git a/unshackle/core/api/download_manager.py b/unshackle/core/api/download_manager.py index 5a614ad2..d71cdb26 100644 --- a/unshackle/core/api/download_manager.py +++ b/unshackle/core/api/download_manager.py @@ -588,6 +588,7 @@ def perform_download( workers=params.get("workers"), adaptive_workers=params.get("adaptive_workers", False), download_processes=params.get("download_processes", 1), + continue_downloads=params.get("continue_downloads", False), downloads=params.get("downloads", 1), worst=params.get("worst", False), best_available=params.get("best_available", False), diff --git a/unshackle/core/api/handlers.py b/unshackle/core/api/handlers.py index 9e2db1c2..1eb88ea6 100644 --- a/unshackle/core/api/handlers.py +++ b/unshackle/core/api/handlers.py @@ -72,6 +72,7 @@ DEFAULT_DOWNLOAD_PARAMS = { "workers": None, "adaptive_workers": False, "download_processes": 1, + "continue_downloads": False, "downloads": 1, "worst": False, "best_available": False, diff --git a/unshackle/core/api/routes.py b/unshackle/core/api/routes.py index 9a2606a3..a918cf77 100644 --- a/unshackle/core/api/routes.py +++ b/unshackle/core/api/routes.py @@ -787,6 +787,9 @@ async def download(request: web.Request) -> web.Response: download_processes: type: integer description: Split a track's segments across this many processes; only engages for large batches (default - 1) + continue_downloads: + type: boolean + description: Keep completed segment files across runs and resume a previously failed download (default - false) downloads: type: integer description: Amount of tracks to download concurrently (default - 1) diff --git a/unshackle/core/config.py b/unshackle/core/config.py index 3a6cea21..ac8112af 100644 --- a/unshackle/core/config.py +++ b/unshackle/core/config.py @@ -130,6 +130,7 @@ class Config: self.update_check_interval: int = kwargs.get("update_check_interval", 24) # mask local base dirs (install root/venv/home) in logged paths; False shows full paths self.redact_paths: bool = kwargs.get("redact_paths", True) + self.continue_downloads: bool = kwargs.get("continue_downloads", False) self.language_tags: dict = kwargs.get("language_tags") or {} self.tag_rules: list = kwargs.get("tag_rules") or [] diff --git a/unshackle/core/manifests/dash.py b/unshackle/core/manifests/dash.py index 0f11a7ef..206155f2 100644 --- a/unshackle/core/manifests/dash.py +++ b/unshackle/core/manifests/dash.py @@ -23,11 +23,12 @@ from pywidevine.pssh import PSSH from requests import Session from unshackle.core.cdm.detect import is_playready_cdm +from unshackle.core.config import config from unshackle.core.constants import DOWNLOAD_CANCELLED, DOWNLOAD_LICENCE_ONLY, AnyTrack from unshackle.core.drm import DRM_T, ClearKeyCENC, PlayReady, Widevine from unshackle.core.events import events from unshackle.core.session import RnetSession -from unshackle.core.tracks import Audio, DownloadContext, Subtitle, Tracks, Video +from unshackle.core.tracks import Audio, DownloadContext, Subtitle, Tracks, Video, resume from unshackle.core.tracks.track import assert_fragments_decrypted from unshackle.core.utilities import is_close_match, log_event, try_ensure_utf8 from unshackle.core.utils.redact import safe_display_url @@ -456,6 +457,15 @@ class DASH: len(init_data) if init_data is not None else None, ) + if not collapse_single_url: + # reuse a prior run's segments only when the fingerprint proves the segmentation unchanged + digest = resume.fingerprint(track.url, segments) + if not (config.continue_downloads and resume.reusable(save_dir, digest)): + shutil.rmtree(save_dir, ignore_errors=True) + save_dir.mkdir(parents=True, exist_ok=True) + if config.continue_downloads: + resume.write_sidecar(save_dir, digest) + if collapse_single_url: # must carry no Range header, or the downloader would skip its ranged-parallel path downloader_args = dict( @@ -615,6 +625,7 @@ class DASH: except OSError: # Directory might not be empty, try removing recursively shutil.rmtree(save_dir, ignore_errors=True) + resume.clear_sidecar(save_dir) progress(downloaded="Downloaded") diff --git a/unshackle/core/manifests/hls.py b/unshackle/core/manifests/hls.py index fb973989..0117f40c 100644 --- a/unshackle/core/manifests/hls.py +++ b/unshackle/core/manifests/hls.py @@ -28,11 +28,12 @@ from requests import Session from unshackle.core import binaries from unshackle.core.cdm.detect import is_playready_cdm, is_widevine_cdm +from unshackle.core.config import config from unshackle.core.constants import DOWNLOAD_CANCELLED, DOWNLOAD_LICENCE_ONLY, AnyTrack from unshackle.core.drm import DRM_T, ClearKey, MonaLisa, PlayReady, Widevine from unshackle.core.events import events from unshackle.core.session import RnetResponse, RnetSession -from unshackle.core.tracks import Audio, DownloadContext, Subtitle, Tracks, Video +from unshackle.core.tracks import Audio, DownloadContext, Subtitle, Tracks, Video, resume from unshackle.core.tracks.track import assert_fragments_decrypted from unshackle.core.utilities import get_extension, is_close_match, log_event, try_ensure_utf8 from unshackle.core.utils.redact import safe_display_url @@ -667,6 +668,22 @@ class HLS: segment_save_dir = save_dir / "segments" + # no resume for AES-128/ClearKey: in-place decryption at merge time makes numbered files ambiguous + resumable_drm = (session_drm is None or isinstance(session_drm, (Widevine, PlayReady))) and not any( + key and key.method == "AES-128" for key in (master.keys or []) + ) + # media_sequence feeds AES IVs; total_segments pins the OnSegmentFilter verdict (shifts every index) + digest = resume.fingerprint( + track.url, + [(url["url"], url["headers"].get("Range")) for url in urls], + extra=[str(getattr(master, "media_sequence", None) or 0), str(total_segments)], + ) + if not (config.continue_downloads and resumable_drm and resume.reusable(save_dir, digest)): + shutil.rmtree(save_dir, ignore_errors=True) + segment_save_dir.mkdir(parents=True, exist_ok=True) + if config.continue_downloads and resumable_drm: + resume.write_sidecar(save_dir, digest) + downloader_args = dict( urls=urls, output_dir=segment_save_dir, @@ -707,6 +724,9 @@ class HLS: for control_file in segment_save_dir.glob("*.!dev"): control_file.unlink(missing_ok=True) + # merge mutates segment files in place from here on; withdraw the reuse proof + resume.clear_sidecar(save_dir) + progress(total=total_segments, completed=0, downloaded="Merging") name_len = len(str(total_segments)) diff --git a/unshackle/core/manifests/ism.py b/unshackle/core/manifests/ism.py index 0423df16..03e85ef0 100644 --- a/unshackle/core/manifests/ism.py +++ b/unshackle/core/manifests/ism.py @@ -17,6 +17,7 @@ from pyplayready.system.pssh import PSSH as PR_PSSH from pywidevine.pssh import PSSH from requests import Session +from unshackle.core.config import config from unshackle.core.constants import DOWNLOAD_CANCELLED, DOWNLOAD_LICENCE_ONLY, AnyTrack from unshackle.core.drm import DRM_T, PlayReady, Widevine from unshackle.core.events import events @@ -28,7 +29,7 @@ from unshackle.core.manifests.ism_init import ( read_track_id, ) from unshackle.core.session import RnetSession -from unshackle.core.tracks import Audio, DownloadContext, Subtitle, Track, Tracks, Video +from unshackle.core.tracks import Audio, DownloadContext, Subtitle, Track, Tracks, Video, resume from unshackle.core.tracks.track import assert_fragments_decrypted from unshackle.core.utilities import log_event, try_ensure_utf8 from unshackle.core.utils.redact import safe_display_url @@ -430,6 +431,14 @@ class ISM: progress(total=len(segments)) downloader = track.downloader + + digest = resume.fingerprint(track.url, [(url, None) for url in segments]) + if not (config.continue_downloads and resume.reusable(save_dir, digest)): + shutil.rmtree(save_dir, ignore_errors=True) + save_dir.mkdir(parents=True, exist_ok=True) + if config.continue_downloads: + resume.write_sidecar(save_dir, digest) + downloader_args = dict( urls=[{"url": url} for url in segments], output_dir=save_dir, @@ -573,6 +582,7 @@ class ISM: except OSError: # a superseded hedge download may still drop a .!dev file here shutil.rmtree(save_dir, ignore_errors=True) + resume.clear_sidecar(save_dir) progress(downloaded="Downloaded") diff --git a/unshackle/core/tracks/resume.py b/unshackle/core/tracks/resume.py new file mode 100644 index 00000000..c2778a47 --- /dev/null +++ b/unshackle/core/tracks/resume.py @@ -0,0 +1,84 @@ +"""Cross-run segment-resume sidecars. + +A track's completed segment files may be reused by a later run, but only once +the manifest proves the segmentation is unchanged. The proof is a fingerprint +of the segmentation stored in a JSON sidecar next to the segment directory. + +The sidecar is deliberately a sibling of the segment directory, never a file +inside it: the manifest parsers merge by globbing the directory with no +extension filter, so any file left in there would be concatenated into the +final output. +""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +from typing import Optional, Sequence +from urllib.parse import urlsplit, urlunsplit + +_VERSION = 1 + + +def sidecar_path(save_dir: Path) -> Path: + """Path of the resume sidecar for save_dir (always a sibling, see module docstring).""" + return save_dir.with_name(save_dir.name + ".resume.json") + + +def _strip_url(url: Optional[str]) -> str: + """Drop query and fragment: auth tokens rotate between runs and would make every digest unique.""" + if not url: + return "" + scheme, netloc, path, _, _ = urlsplit(url) + return urlunsplit((scheme, netloc, path, "", "")) + + +def fingerprint( + manifest_url: Optional[str], + segments: Sequence[tuple[str, Optional[str]]], + extra: Sequence[str] = (), +) -> str: + """Stable digest of a track's segmentation. + + Hashes the manifest URL (sans query), the segment count, and each + segment's URL (sans query) with its byte-range string verbatim; byte + ranges are the strongest identity signal available and never carry + tokens. extra carries protocol-specific identity inputs (HLS: the + media_sequence start and post-filter segment count). + """ + stripped = [[_strip_url(url), byte_range or ""] for url, byte_range in segments] + if len({tuple(pair) for pair in stripped}) != len(stripped): + stripped = [[url, byte_range or ""] for url, byte_range in segments] + payload = { + "version": _VERSION, + "manifest": _strip_url(manifest_url), + "count": len(segments), + "segments": stripped, + "extra": list(extra), + } + canonical = json.dumps(payload, separators=(",", ":"), sort_keys=True) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +def reusable(save_dir: Path, digest: str) -> bool: + """True when save_dir's sidecar records this exact digest. + + Never raises: a missing, unreadable, or malformed sidecar means "cannot + prove reuse is safe", which means restart clean. + """ + try: + data = json.loads(sidecar_path(save_dir).read_text(encoding="utf-8")) + return bool(data.get("digest")) and data["digest"] == digest + except (OSError, ValueError): + return False + + +def write_sidecar(save_dir: Path, digest: str) -> None: + """Record digest so a later run can prove save_dir's segments are reusable.""" + sidecar_path(save_dir).write_text(json.dumps({"version": _VERSION, "digest": digest}), encoding="utf-8") + + +def clear_sidecar(save_dir: Path) -> None: + """Withdraw the reuse proof (merge is about to mutate files, or the track completed).""" + sidecar_path(save_dir).unlink(missing_ok=True) diff --git a/unshackle/core/tracks/track.py b/unshackle/core/tracks/track.py index 1493e63a..88559a3b 100644 --- a/unshackle/core/tracks/track.py +++ b/unshackle/core/tracks/track.py @@ -26,6 +26,7 @@ from unshackle.core.downloaders import requests from unshackle.core.drm import DRM_T, ClearKeyCENC, PlayReady, Widevine from unshackle.core.events import events from unshackle.core.session import RnetSession +from unshackle.core.tracks import resume from unshackle.core.utilities import get_boxes, log_event, try_ensure_utf8 from unshackle.core.utils.subprocess import ffprobe @@ -437,10 +438,19 @@ class Track: else: save_dir = save_path.parent - def cleanup(): + keep_segments = config.continue_downloads and self.descriptor != self.Descriptor.URL + + def cleanup() -> None: save_path.unlink(missing_ok=True) - if save_dir.exists() and save_dir.name.endswith("_segments"): - shutil.rmtree(save_dir) + if save_dir.name.endswith("_segments"): + if keep_segments: + if save_dir.exists(): + for partial in save_dir.rglob("*.!dev"): + partial.unlink(missing_ok=True) + else: + if save_dir.exists(): + shutil.rmtree(save_dir) + resume.clear_sidecar(save_dir) if not DOWNLOAD_LICENCE_ONLY.is_set(): if config.directories.temp.is_file(): @@ -448,10 +458,8 @@ class Track: config.directories.temp.mkdir(parents=True, exist_ok=True) - # Delete any pre-existing temp files matching this track. - # We can't re-use or continue downloading these tracks as they do not use a - # lock file. Or at least the majority don't. Even if they did I've encountered - # corruptions caused by sudden interruptions to the lock file. + # completed segments are reusable once the parser proves the segmentation + # unchanged (resume sidecar); partial .!dev files never survive a run boundary cleanup() try: