mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-14 13:05:09 +00:00
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.
This commit is contained in:
@@ -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. |
|
||||
|
||||
@@ -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. |
|
||||
|
||||
@@ -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:** `{}`
|
||||
|
||||
@@ -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` |
|
||||
|
||||
@@ -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
|
||||
@@ -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()
|
||||
@@ -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"""<?xml version="1.0"?>
|
||||
<MPD xmlns="urn:mpeg:dash:schema:mpd:2011" type="static" profiles="urn:mpeg:dash:profile:isoff-on-demand:2011">
|
||||
<BaseURL>{base(srv)}</BaseURL>
|
||||
<Period id="0">
|
||||
<AdaptationSet contentType="video" mimeType="video/mp4" lang="en">
|
||||
<Representation id="v0" codecs="avc1.640028" bandwidth="1000000" width="1920" height="1080">
|
||||
<SegmentList timescale="1">
|
||||
<Initialization sourceURL="init.bin"/>
|
||||
<SegmentURL media="part0.bin"/>
|
||||
<SegmentURL media="{part1}"/>
|
||||
</SegmentList>
|
||||
</Representation>
|
||||
</AdaptationSet>
|
||||
</Period>
|
||||
</MPD>"""
|
||||
|
||||
|
||||
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
|
||||
@@ -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()
|
||||
@@ -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))
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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 []
|
||||
|
||||
@@ -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")
|
||||
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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")
|
||||
|
||||
|
||||
|
||||
@@ -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)
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user