"""Resumable media download with parallel ranges and atomic promotion.
One :class:`Downloader` per client. Fetches a *logical file* — either a
single bento URL or a sequence of DASH segment URLs — into ``dest``:
- bytes always land in ``dest + ".part"`` first and are promoted with
``os.replace`` only after size verification, so a file at the final
path is always complete (the skip-existing contract);
- a failed HEAD/size probe is never treated as "file changed": when
the remote size cannot be determined the local ``.part`` is trusted,
(legacy bug: one dead HEAD re-downloaded whole playlists);
- a 416 or a 206 short read that contradicts the probe means the
object changed mid-run (re-encode, stale CDN edge): the true size
(carried by the response) re-evaluates the local prefix — matching
promotes it complete, mismatching restarts the transfer;
- single-URL files download in parallel 1 MiB range parts (thread
pool); DASH segments download sequentially and are concatenated in
manifest order;
- interruption leaves the ``.part`` in place for the next run —
trusted only when it was written for the *same source*: the
source identity (track id, delivered quality/codec, stable media
URLs) is recorded in ``<dest>.part.source.json`` and a mismatch
restarts the transfer (a same-size different rendition would
otherwise slip past every size check; tidekeeper parity);
- progress is reported as plain callback events (thread-safe);
- cancellation raises :class:`~tidalpy.errors.CancelledError` and
keeps the ``.part``;
- a zero-byte result is never promoted (a remote that answers with
no data would otherwise poison the skip-existing contract);
- progress never double-counts: a resumed run reports the on-disk
prefix as its baseline and each byte as written at most once
(a re-fetched DASH segment overlapping the prefix re-claims no
progress).
"""
from __future__ import annotations
import errno
import hashlib
import io
import json
import logging
import os
import random
import threading
import time
from collections.abc import Mapping
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
import requests
from .config import Config
from .errors import (
CancelledError,
DiskFullError,
DownloadError,
)
from .transport import Transport, validate_media_url
logger = logging.getLogger("tidalpy.downloader")
_ENOSPC = (errno.ENOSPC, getattr(errno, "ENOSPC2", errno.ENOSPC))
_RETRY_BACKOFF_BASE = 0.5
_MAX_PART_RETRIES = 3
@dataclass
class _Part:
url: str
offset: int # offset in the logical file
length: int # bytes to fetch
def _disk_full(exc: OSError) -> bool:
return exc.errno in _ENOSPC
def _with_jitter(backoff: float) -> float:
"""Half-to-full jitter: spreads concurrent part retries off phase."""
return random.uniform(backoff / 2, backoff)
# Signed-query keys that expire in media URLs (the stable subset of
# the CDN authorization parameters; tidekeeper parity). A resumed
# transfer under a *renewed* token fetches the same immutable
# representation — these are the only query fields that change.
_AUTH_QUERY_KEYS = frozenset(
{
"authorization",
"expires",
"key-pair-id",
"policy",
"signature",
"sig",
"token",
"x-amz-algorithm",
"x-amz-content-sha256",
"x-amz-credential",
"x-amz-date",
"x-amz-expires",
"x-amz-security-token",
"x-amz-signature",
"x-amz-signedheaders",
"x-goog-algorithm",
"x-goog-credential",
"x-goog-date",
"x-goog-expires",
"x-goog-signature",
"x-goog-signedheaders",
}
)
SOURCE_MARKER_SUFFIX = ".part.source.json"
[docs]
def source_identity(urls: list[str], media: dict[str, str] | None) -> str:
"""The identity of a transfer's source: sha256 over the media
identity (track id, delivered quality, codec) and the stable form
of the URLs.
The resume contract's guard: a leftover ``.part`` is only trusted
when it was written for *this* source. A renewed CDN token keeps
the same identity (the representation is immutable); a different
rendition — quality, codec, or URL — does not, and a same-size
different rendition would otherwise slip past every size check
(a prefix that is fully inside a skipped segment is never
re-verified).
"""
source = json.dumps(
{"media": dict(media or {}), "urls": [stable_media_url(u) for u in urls]},
sort_keys=True,
)
return hashlib.sha256(source.encode("utf-8")).hexdigest()
class _ServerHttpResponse(Exception):
"""A 5xx from the media CDN — retried within the part budget."""
def __init__(self, status: int) -> None:
super().__init__(f"media server error (HTTP {status})")
self.status = status
class _StaleProbe(DownloadError):
"""The probed size no longer matches the object.
A 416 (range beyond the object) or a 206 short read that
advertises a smaller total means the object changed after the
probe (a re-encode, a stale CDN edge). Carries the true size
when the response advertised one (``actual_size``). Never
consumed by the part-retry budget — the caller re-probes.
"""
retryable = True
def __init__(self, message: str, actual_size: int | None) -> None:
super().__init__(message, status_code=416)
self.actual_size = actual_size
[docs]
class Downloader:
"""Fetch media URLs into a destination path (see module docstring)."""
[docs]
def __init__(
self,
transport: Transport,
config: Config,
stop_event: threading.Event,
on_progress: Callable[[int, int | None], None] | None = None,
resolver: Callable[[str], tuple[str, ...]] | None = None,
sleeper: Callable[[float], None] | None = None,
) -> None:
self._transport = transport
self._config = config
self._stop = stop_event
self._on_progress = on_progress
self._resolver = resolver
self._validate_urls = config.validate_media_urls
self._sleep = sleeper or self._interruptible_sleep
self._lock = threading.RLock()
self._written = 0
# -- public -------------------------------------------------------------
[docs]
def download(
self,
urls: list[str],
dest: Path,
encryption_key: str | None = None,
media_identity: dict[str, str] | None = None,
) -> Path:
"""Fetch the logical file described by ``urls`` into ``dest``.
``media_identity`` (the track id, delivered quality and codec
the URLs describe) enables the source guard: the leftover
``.part`` is only resumed when it was written for the same
source (see :func:`source_identity`). Omitted (a host using
the downloader directly for a bare URL), the guard is off and
the size checks are the only resume protection, as before.
"""
if not urls:
raise DownloadError("no media URLs to download")
if self._validate_urls:
for url in urls:
validate_media_url(url, resolver=self._resolver)
dest = Path(dest)
dest.parent.mkdir(parents=True, exist_ok=True)
part_path = dest.with_name(dest.name + ".part")
source_marker = dest.with_name(dest.name + SOURCE_MARKER_SUFFIX)
identity = source_identity(urls, media_identity) if media_identity else None
if identity is None:
# Without an identity the marker no longer describes the
# source: drop it (it is a resume optimisation, nothing
# more).
source_marker.unlink(missing_ok=True)
else:
recorded = self._read_source_marker(source_marker)
if recorded is not None and recorded != identity:
# The representation changed between runs: the prefix
# belongs to another byte stream and is not trusted.
logger.debug(
"%s: source identity changed; restarting the transfer",
dest.name,
)
part_path.unlink(missing_ok=True)
self._write_source_marker(source_marker, identity)
sizes = [self._probe(url) for url in urls]
known = all(size >= 0 for size in sizes)
total = sum(sizes) if known else -1
# Existing local state.
existing = 0
if part_path.exists():
try:
existing = part_path.stat().st_size
except OSError:
existing = 0
if known and existing > total:
# The remote shrank (re-encode) — start over.
part_path.unlink(missing_ok=True)
existing = 0
if known and existing == total:
logger.debug("%s: part file already complete (%s bytes)", dest.name, total)
else:
try:
self._fetch_all(urls, part_path, sizes, known, existing)
except _StaleProbe as stale:
# The object changed after the probe (a re-encode, a
# stale CDN edge): the 416/short-206 may carry the
# true size, otherwise re-probe. Then re-evaluate the
# local prefix against the true size (reference-fork
# parity: a 416 matching the prefix confirms it
# complete; a mismatching one restarts the transfer).
if len(urls) == 1 and stale.actual_size is not None:
sizes, known, total = [stale.actual_size], True, stale.actual_size
else:
sizes = [self._probe(url) for url in urls]
known = all(size >= 0 for size in sizes)
total = sum(sizes) if known else -1
if known:
try:
existing = part_path.stat().st_size
except OSError:
existing = 0
if known and existing == total:
logger.debug(
"%s: 416 confirmed the part is complete (%s bytes)",
dest.name,
total,
)
else:
part_path.unlink(missing_ok=True)
self._fetch_all(urls, part_path, sizes, known, 0)
# Size verification (only when it can be trusted).
if known:
final_size = part_path.stat().st_size
if final_size != total:
# The assembled bytes are proven inconsistent with the
# remote (e.g. a mid-run re-encode): a resume next run
# would trust the corrupt prefix, so drop the part
# (reference-client parity: the legacy tool removes the
# file on a mismatch instead of keeping it).
part_path.unlink(missing_ok=True)
raise DownloadError(
f"size mismatch after download: {final_size} != {total} "
f"for {dest.name}"
)
if part_path.stat().st_size == 0:
# A zero-byte "download" is never valid media: the remote
# answered with no data (a dead CDN edge, an empty 200).
# Promoting it would poison the skip-existing contract for
# every future run of this item (reference-fork pin: an
# empty transfer of unknown size is not promoted).
part_path.unlink(missing_ok=True)
raise DownloadError(
f"downloaded file is empty (0 bytes): {dest.name} got no "
"data from the media server; the object may be broken "
"or the CDN answered empty — re-download in a while"
)
if encryption_key:
from .decrypt import decrypt_in_place
decrypt_in_place(part_path, encryption_key)
# Atomic promotion (both branches): a file at ``dest`` is complete.
os.replace(part_path, dest)
source_marker.unlink(missing_ok=True)
return dest
[docs]
def remove_partial(self, dest: Path) -> None:
"""Drop a leftover ``.part`` (used after non-resumable failures)."""
dest = Path(dest)
dest.with_name(dest.name + ".part").unlink(missing_ok=True)
dest.with_name(dest.name + SOURCE_MARKER_SUFFIX).unlink(missing_ok=True)
# -- source marker ------------------------------------------------------
@staticmethod
def _read_source_marker(marker: Path) -> str | None:
try:
data = json.loads(marker.read_text(encoding="utf-8"))
except (OSError, ValueError):
return None
if not isinstance(data, dict):
return None
identity = data.get("identity")
return identity if isinstance(identity, str) and identity else None
@staticmethod
def _write_source_marker(marker: Path, identity: str) -> None:
# Best-effort by the resume contract: a lost marker costs one
# extra re-download next run, never a wrong file.
try:
tmp = marker.with_name(marker.name + ".tmp")
tmp.write_text(json.dumps({"identity": identity}), encoding="utf-8")
os.replace(tmp, marker)
except OSError:
try:
marker.unlink(missing_ok=True)
except OSError:
pass
# -- internals ------------------------------------------------------------
def _check_stop(self) -> None:
if self._stop.is_set():
raise CancelledError("download cancelled")
def _interruptible_sleep(self, seconds: float) -> None:
deadline = time.monotonic() + max(0.0, seconds)
while True:
self._check_stop()
remaining = deadline - time.monotonic()
if remaining <= 0:
return
time.sleep(min(0.5, remaining))
def _probe(self, url: str) -> int:
"""Remote size of one media URL (``-1`` when unknown)."""
try:
return self._transport.probe_size(url)
except DownloadError:
return -1
def _report(self) -> None:
if self._on_progress is not None:
with self._lock:
done = self._written
self._on_progress(done, self._expected_total)
def _note_written(self, count: int) -> None:
with self._lock:
self._written += count
self._report()
def _fetch_all(
self,
urls: list[str],
part_path: Path,
sizes: list[int],
known: bool,
existing: int,
) -> None:
self._expected_total = sum(sizes) if known else None
self._written = existing
if not part_path.exists():
part_path.touch()
if len(urls) == 1:
self._fetch_single_url(urls[0], part_path, sizes[0], known, existing)
else:
self._fetch_segments(urls, part_path, sizes, known, existing)
# -- single URL (parallel range parts) -------------------------------------
def _fetch_single_url(
self, url: str, part_path: Path, size: int, known: bool, existing: int
) -> None:
if not known:
# No size: range resume is impossible and a leftover .part
# cannot be verified — re-fetch the whole file from scratch.
logger.debug(
"remote size unknown; re-fetching %s from scratch", part_path.name
)
self._fetch_segments([url], part_path, [-1], False, 0)
return
if existing >= size:
return
self._fetch_parts(url, part_path, size, existing)
def _fetch_parts(
self, url: str, part_path: Path, size: int, start_offset: int
) -> None:
part_size = self._config.part_size
parts: list[_Part] = []
offset = start_offset
while offset < size:
# Parts stay part_size-aligned except for a leading tail when
# the leftover .part ended mid-part (an interrupted write).
boundary = (offset + part_size - 1) // part_size * part_size
end = min(boundary if boundary > offset else offset + part_size, size)
parts.append(_Part(url=url, offset=offset, length=end - offset))
offset = end
if not parts:
return
workers = max(1, min(self._config.download_workers, len(parts)))
if workers == 1 or len(parts) == 1:
with open(part_path, "r+b") as handle:
for part in parts:
self._check_stop()
self._fetch_part(part, handle)
return
with open(part_path, "r+b") as handle:
with ThreadPoolExecutor(
max_workers=workers, thread_name_prefix="tidal-dl"
) as pool:
futures = [
pool.submit(self._fetch_part, part, handle) for part in parts
]
errors: list[BaseException] = []
for future in futures:
try:
future.result()
except BaseException as exc: # noqa: BLE001 - re-raised below
errors.append(exc)
if errors:
raise errors[0]
def _fetch_part(self, part: _Part, handle: io.BufferedRandom) -> None:
last_error: Exception | None = None
for attempt in range(_MAX_PART_RETRIES + 1):
self._check_stop()
try:
response = self._transport.fetch(
part.url,
headers={
"Range": f"bytes={part.offset}-{part.offset + part.length - 1}"
},
)
try:
if response.status_code == 206:
try:
data = self._read_exact(response, part.length)
except DownloadError as exc:
actual = self._actual_size(response.headers)
if (
actual is not None
and actual < part.offset + part.length
):
# The object shrank mid-run: the short
# read is the truth, not a glitch.
raise _StaleProbe(
f"object shrank mid-run (HTTP 206 "
f"short read): {exc}",
actual_size=actual,
) from exc
raise
elif response.status_code == 416:
# The probed size is stale: the object is
# smaller than the probe said. The 416 may
# carry the true size (Content-Range: */N).
# Never retried — the caller re-probes.
raise _StaleProbe(
f"range not satisfiable (HTTP 416): the "
f"probed size is stale for {part.url}",
actual_size=self._actual_size(response.headers),
)
elif 500 <= response.status_code < 600:
# CDN 502/503 during deploys clears within the
# part budget; a fresh Range request restarts
# the part from scratch, so no partial write.
raise _ServerHttpResponse(response.status_code)
elif response.status_code == 200:
# Server ignored the Range header: the body is
# the whole file, not the part — unusable for
# parallel assembly.
raise DownloadError(
"media server does not support range requests"
)
else:
raise DownloadError(
f"range request failed (HTTP {response.status_code})"
)
finally:
response.close()
with self._lock:
handle.seek(part.offset)
handle.write(data)
self._note_written(len(data))
if self._config.pacing_bps:
self._sleep(max(0.05, (len(data) * 8) / self._config.pacing_bps))
return
except (CancelledError, DiskFullError, DownloadError):
raise
except (_ServerHttpResponse, requests.RequestException, OSError) as exc:
last_error = exc
if isinstance(exc, OSError) and _disk_full(exc):
raise DiskFullError(f"disk full while writing part: {exc}") from exc
if attempt == _MAX_PART_RETRIES:
break
backoff = _with_jitter(_RETRY_BACKOFF_BASE * (2**attempt))
logger.debug("part fetch error (%s); retrying in %.1fs", exc, backoff)
self._sleep(backoff)
raise DownloadError(
f"part download failed after retries: {last_error}"
) from last_error
# -- multiple URLs (sequential segments) -------------------------------------
def _fetch_segments(
self,
urls: list[str],
part_path: Path,
sizes: list[int],
known: bool,
existing: int,
) -> None:
self._check_stop()
with open(part_path, "r+b") as handle:
if not known:
# Full rewrite: nothing of the previous content is
# trusted (and it is not *reported* either).
handle.truncate(0)
existing = 0
position = 0 # next write position in the logical file
for index, url in enumerate(urls):
seg_size = sizes[index] if known else -1
seg_start = position
if known and seg_size >= 0 and existing >= position + seg_size:
# This segment is fully inside the existing prefix.
position += seg_size
continue
position = self._fetch_segment(url, handle, seg_start, existing)
if self._config.pacing_bps and seg_size >= 0:
self._sleep(max(0.05, (seg_size * 8) / self._config.pacing_bps))
def _fetch_segment(
self,
url: str,
handle: io.BufferedRandom,
seg_start: int,
existing: int,
) -> int:
"""Fetch one segment, retrying within the part budget.
A retry restarts the segment at ``seg_start`` (every write is
an absolute seek), so a mid-stream failure can never corrupt
the assembly. Returns the next logical write position.
``existing`` is the byte count already on disk before this
transfer: a re-fetched segment that overlaps it must not claim
those bytes as progress (a resumed run would double-count them
— reference-fork pin: a restarted transfer reports each byte
exactly once).
"""
new_from = max(seg_start, existing)
last_error: Exception | None = None
for attempt in range(_MAX_PART_RETRIES + 1):
self._check_stop()
write_pos = seg_start
try:
response = self._transport.fetch(url)
try:
self._check_stop()
if 500 <= response.status_code < 600:
raise _ServerHttpResponse(response.status_code)
if response.status_code != 200:
raise DownloadError(
f"segment download failed (HTTP {response.status_code})"
)
for chunk in response.iter_content(chunk_size=256 * 1024):
if not chunk:
continue
self._check_stop()
with self._lock:
handle.seek(write_pos)
handle.write(chunk)
fresh = max(0, len(chunk) - (new_from - write_pos))
write_pos += len(chunk)
if fresh:
self._note_written(fresh)
finally:
response.close()
return write_pos
except (CancelledError, DiskFullError, DownloadError):
raise
except (_ServerHttpResponse, requests.RequestException, OSError) as exc:
last_error = exc
if isinstance(exc, OSError) and _disk_full(exc):
raise DiskFullError(
f"disk full while writing segment: {exc}"
) from exc
if attempt == _MAX_PART_RETRIES:
break
backoff = _with_jitter(_RETRY_BACKOFF_BASE * (2**attempt))
logger.debug(
"segment fetch error (%s); retry %d in %.1fs",
exc,
attempt + 1,
backoff,
)
self._sleep(backoff)
raise DownloadError(
f"segment download failed after retries: {last_error}"
) from last_error
# -- shared helpers ------------------------------------------------------
@staticmethod
def _actual_size(headers: Mapping[str, Any]) -> int | None:
"""The true object size a 416/206 advertises (or ``None``).
``Content-Range: bytes 0-99/12345`` and the 416 form
``Content-Range: bytes */12345`` both carry the total.
"""
value = headers.get("Content-Range") or ""
if "/" not in value:
return None
total = value.rsplit("/", 1)[-1].strip()
return int(total) if total.isdigit() else None
def _read_exact(self, response: requests.Response, expected: int) -> bytes:
buffer = bytearray()
for chunk in response.iter_content(chunk_size=256 * 1024):
if not chunk:
continue
self._check_stop()
buffer.extend(chunk)
if len(buffer) > expected:
# Server over-delivered; trust the requested length.
buffer = buffer[:expected]
break
if len(buffer) != expected:
raise DownloadError(
f"short read from media server: {len(buffer)} != {expected}"
)
return bytes(buffer)