"""Download pipeline: resolve -> fetch -> post-process -> report.
The pipeline is the only place that turns a resolved
:class:`~tidalpy.models.Resource` into a
:class:`~tidalpy.models.DownloadReport`. It implements the two
contracts that make a collection download safe in a shared
environment:
- **per-item isolation**: every track is resolved, downloaded and
post-processed inside its own try/except; one bad item cannot kill
the run. ``skip_broken=False`` stops after the first unrecoverable
item instead of continuing;
- **resume & skip-existing**: streams land in ``<path>.part`` and are
promoted atomically, so a non-empty file at the final path is
complete and skipped (the downloader owns the byte-level details).
Post-processing per item, in order: FLAC-in-MP4 remux (when configured
and ffmpeg is present), ffprobe verification, tag writing (best-effort
by contract — never fails the item).
"""
from __future__ import annotations
import logging
import os
import threading
from pathlib import Path
from typing import Any, Callable
from . import media as media_mod
from . import naming
from . import receipts as receipts_mod
from . import tags as tags_mod
from .api import TidalApi, artwork_url
from .config import Config
from .downloader import Downloader
from .errors import CancelledError, TidalError
from .models import (
Album,
DownloadReport,
ItemReport,
Playlist,
Resource,
StreamInfo,
Track,
)
from .quality import LOSSLESS_API_QUALITIES, Quality, playback_attempts
from .transport import Transport
logger = logging.getLogger("tidalpy.pipeline")
ProgressCallback = Callable[[dict[str, Any]], None]
# Cover art is fetched from Tidal's static image CDN at a fixed size;
# the URL itself comes from :func:`tidalpy.api.artwork_url` (the
# five-segment id with dashes turned into slashes — the exact pattern
# the reference tool uses, tagHelper).
COVER_WIDTH = 1280
COVER_HEIGHT = 1280
def _base_event(
kind: str, index: int, total: int, track: Track | None, **extra: Any
) -> dict:
event = {
"stage": "item",
"kind": kind,
"index": index,
"total": total,
"id": track.id if track is not None else None,
"title": track.title if track is not None else None,
"status": None,
"reason": "",
}
event.update(extra)
return event
[docs]
class Pipeline:
"""Orchestrates a whole download run (see module docstring)."""
[docs]
def __init__(
self,
api: TidalApi,
config: Config,
transport: Transport,
media_resolver: Callable[[str], tuple[str, ...]] | None = None,
) -> None:
self._api = api
self._config = config
self._transport = transport
self._media_resolver = media_resolver
# -- public -----------------------------------------------------------------
[docs]
def download(
self,
resource: Resource,
*,
dest: str | Path | None = None,
quality: Quality | None = None,
skip_broken: bool | None = None,
on_progress: ProgressCallback | None = None,
stop_event: threading.Event,
) -> DownloadReport:
"""Download one resolved resource; always returns a report."""
quality = quality or self._config.quality
skip_broken = self._config.skip_broken if skip_broken is None else skip_broken
report = DownloadReport(
kind=resource.kind,
title=resource.title,
requested_quality=quality.value,
)
try:
tracks = list(self._api.get_items(resource))
except CancelledError:
# Cancelled before the run even started (the stop event is
# also bound to the transport): still return a report.
report.cancelled = True
self._emit_complete(report, on_progress)
return report
dest_dir = self._dest_dir(resource, dest)
self._emit(
on_progress,
{
"stage": "start",
"kind": resource.kind,
"title": resource.title,
"total": len(tracks),
"requested_quality": quality.value,
"dest": str(dest_dir),
},
)
cancelled = False
try:
self._download_items(
report,
tracks,
resource,
dest_dir,
quality,
skip_broken,
stop_event,
on_progress,
)
except CancelledError:
# Cancellation between items (or before the first) ends the run;
# the report is still returned so the host sees the partial
# result. Cancellation *during* an item is recorded on that
# item (status=failed, code=cancelled) and then also lands here.
cancelled = True
report.cancelled = cancelled
self._emit_complete(report, on_progress)
return report
# -- events ------------------------------------------------------------------
def _emit_complete(
self, report: DownloadReport, on_progress: ProgressCallback | None
) -> None:
self._emit(
on_progress,
{
"stage": "complete",
"kind": report.kind,
"title": report.title,
"ok": report.ok,
"total": len(report.items),
"failed": len(report.failures),
"skipped": report.skipped_existing,
"total_bytes": report.total_bytes,
"cancelled": report.cancelled,
},
)
# -- per item loop -----------------------------------------------------------
def _download_items(
self,
report: DownloadReport,
tracks: list[Track | None],
resource: Resource,
dest_dir: Path,
quality: Quality,
skip_broken: bool,
stop_event: threading.Event,
on_progress: ProgressCallback | None,
) -> None:
"""Download every track (see :meth:`download` for the contracts)."""
# Canonical-name registry: when two *different* tracks in this
# run would produce the identical file name, the later one gets a
# deterministic `` (index)`` suffix (see :meth:`_claim_dest`)
# instead of being skipped over the earlier one's file.
claimed: dict[Path, object] = {}
for index, track in enumerate(tracks, start=1):
self._check_stop(stop_event)
if track is None:
# A non-audio slot (music video): counted, never
# downloaded — the catalog's item count must match the
# report's.
self._record(
report,
self._skipped(
index,
len(tracks),
None,
"video item (audio-only library)",
kind="video",
),
)
continue
if track.id is None:
self._record(
report,
self._skipped(index, len(tracks), None, "not a track"),
)
continue
try:
item = self._download_item(
resource,
track,
index,
len(tracks),
dest_dir,
quality,
stop_event,
on_progress,
claimed,
)
except CancelledError as exc:
item = self._failed(
index,
len(tracks),
track,
exc,
code="cancelled",
on_progress=on_progress,
)
item.retryable = False
self._record(report, item)
# The run is over: let the exception end the loop.
raise
except TidalError as exc:
item = self._failed(
index, len(tracks), track, exc, on_progress=on_progress
)
except Exception as exc: # noqa: BLE001 - isolation contract
logger.warning(
"unexpected error on item %s: %s", track.id, exc, exc_info=True
)
item = self._failed(
index,
len(tracks),
track,
exc,
code="internal",
on_progress=on_progress,
)
self._record(report, item)
if item.status == "failed" and not skip_broken:
# Abort the run: remaining items are not attempted.
for later_index, later in enumerate(tracks[index:], start=index + 1):
self._record(
report,
self._skipped(
later_index,
len(tracks),
later,
"aborted: previous item failed",
),
)
break
# -- destination layout ----------------------------------------------------
def _dest_dir(self, resource: Resource, dest: str | Path | None) -> Path:
base = Path(dest).expanduser() if dest else self._config.output_dir
data = resource.data
if resource.kind == "album" and isinstance(data, Album):
return base / naming.album_dir(data, self._config.album_format)
if isinstance(data, Playlist):
return base / naming.playlist_dir(data, self._config.playlist_format)
return base
# -- per item ----------------------------------------------------------------
def _download_item(
self,
resource: Resource,
track: Track,
index: int,
total: int,
dest_dir: Path,
quality: Quality,
stop_event: threading.Event,
on_progress: ProgressCallback | None,
claimed: dict[Path, object],
) -> ItemReport:
self._emit(
on_progress,
_base_event(
resource.kind,
index,
total,
track,
status="start",
),
)
if track.id is None:
raise TidalError(f"item {index} has no track id")
# Stream-free skip check (reference-fork parity): resuming a
# large collection must not spend a rate-limited playback
# request on every track that is already on disk. By default a
# non-empty file at a catalog-computed candidate is complete
# (atomic promotion), so it skips; with receipts on
# (save_receipts) the file must additionally carry a receipt
# proving it is this track, intact, post-processed, and at
# least the requested quality — a lower-quality file
# re-downloads (the upgrade path). This check is deliberately
# weaker than the stream-based one below — no remote-size
# probe, which would need the very request being saved. A
# candidate already claimed by a different track in this run
# is not this track's file; the stream path dedups it as usual.
for candidate in self._planned_dests(resource, track, dest_dir, index, quality):
if claimed.get(candidate) not in (None, track.id):
continue
if self._config.save_receipts:
if not receipts_mod.satisfied(
candidate, receipts_mod.identity("track", track.id), quality
):
continue
else:
try:
if candidate.stat().st_size <= 0:
continue
except OSError:
continue
dest = self._claim_dest(candidate, track.id, index, claimed)
logger.info("skip existing: %s", dest)
self._emit(
on_progress,
_base_event(
resource.kind,
index,
total,
track,
status="skipped",
reason="file exists",
path=str(dest),
),
)
return ItemReport(
index=index,
kind="track",
id=track.id,
title=track.title,
status="skipped",
reason="file exists",
path=str(dest),
)
stream = self._api.resolve_stream(track.id, quality, track)
dest = self._item_dest(resource, track, stream, dest_dir, index)
dest = self._claim_dest(dest, track.id, index, claimed)
if self._skip_existing(dest, stream, track.id, quality):
logger.info("skip existing: %s", dest)
self._emit(
on_progress,
_base_event(
resource.kind,
index,
total,
track,
status="skipped",
reason="file exists",
path=str(dest),
delivered_quality=stream.delivered_quality,
codec=stream.codec,
),
)
return ItemReport(
index=index,
kind="track",
id=track.id,
title=track.title,
status="skipped",
reason="file exists",
path=str(dest),
delivered_quality=stream.delivered_quality,
)
downloader = Downloader(
self._transport,
self._config,
stop_event,
on_progress=lambda done, expected: (
self._emit(
on_progress,
_base_event(
resource.kind,
index,
total,
track,
stage="bytes",
bytes=done,
total_bytes=expected,
),
)
if expected
else None
),
resolver=self._media_resolver,
)
# The source identity guards the .part resume: a renewed CDN
# token keeps the same identity, a different rendition
# (quality/codec) invalidates the prefix (tidekeeper parity).
path = downloader.download(
stream.urls,
dest,
encryption_key=stream.encryption_key,
media_identity={
"type": "track",
"id": str(stream.track_id),
"quality": stream.delivered_quality,
"codec": stream.codec,
},
)
# FLAC-in-MP4: remux to a true .flac when configured. Decided
# from the bytes on disk (the fLaC magic), not the manifest
# prediction: a FLAC codec that is *not* real FLAC on disk is
# a container and must be remuxed, while genuine FLAC skips the
# ffmpeg pass (reference-client parity: the legacy tool sniffs
# the same header before converting).
if (
self._config.save_as_flac
and "flac" in (stream.codec or "").lower()
and not media_mod.is_flac_file(path)
):
path = media_mod.export_flac_from_container(path, stop_event)
size = path.stat().st_size
if self._config.verify_media:
media_mod.verify_media(path, stream, stop_event)
lyrics_data: dict[str, str] | None = None
if self._config.write_lyrics or self._config.lyric_file:
lyrics_data = self._api.get_lyrics(track.id, track)
if self._config.write_tags:
cover = self._cover_bytes(track, resource)
composer = self._composer_names(track.id)
# The tag carries the plain text, falling back to the
# timed version when that is all Tidal has (reference-
# client parity).
tag_lyrics = None
if self._config.write_lyrics and lyrics_data:
tag_lyrics = lyrics_data["lyrics"] or lyrics_data["subtitles"] or None
# The stream's normalization facts (v2, when the
# resolution path provides them) win; the v1 catalog's
# own track loudness facts fill the gap — without them
# the primary bento path would write no ReplayGain tags
# at all (the v1 track payload carries them on every
# item, but the bento stream carries none).
replaygain = {**(track.replaygain or {}), **(stream.replaygain or {})}
tags_mod.write_tags(
path,
track,
track.album,
cover_bytes=cover,
replaygain=replaygain or None,
composer=composer,
lyrics=tag_lyrics,
)
if self._config.lyric_file and lyrics_data:
self._write_lyrics_sidecar(path, lyrics_data)
if self._config.save_receipts:
# Everything this run writes is now on disk: the receipt
# claims the file complete, at the delivered quality.
# Best effort — a lost receipt costs one re-download next
# run, never this one.
try:
receipts_mod.write_receipt(
path,
receipts_mod.identity("track", track.id),
stream.delivered_quality,
stream.codec,
quality,
)
except Exception: # noqa: BLE001 - garnish, never fatal
logger.debug("receipt write failed for %s", path, exc_info=True)
self._emit(
on_progress,
_base_event(
resource.kind,
index,
total,
track,
status="ok",
path=str(path),
# The real-time channel reports what actually landed:
# a downgraded run (LOSSLESS asked, HI_RES delivered)
# is visible in the events, not only in the report.
delivered_quality=stream.delivered_quality,
codec=stream.codec,
),
)
return ItemReport(
index=index,
kind="track",
id=track.id,
title=track.title,
status="ok",
path=str(path),
delivered_quality=stream.delivered_quality,
size=size,
)
[docs]
def dest_for(
self,
resource: Resource,
track: Track,
stream: StreamInfo,
index: int = 0,
dest: str | Path | None = None,
) -> Path:
"""The canonical destination path for one item.
The exact computation ``download`` uses, exposed so hosts can
get the path without downloading (pre-flight checks, resume
planning). ``index`` is the 1-based position inside the
collection (matters only for playlist/mix numbering).
"""
return self._item_dest(
resource, track, stream, self._dest_dir(resource, dest), index
)
def _skip_existing(
self,
dest: Path,
stream: StreamInfo,
track_id: int | str,
quality: Quality,
) -> bool:
"""Whether an existing file at the final path satisfies the
skip contract.
By default a file at the final path can only exist after a
completed download (atomic promotion), so a non-empty file is
complete. The remote size is an extra truncation check *when
it can be determined*: all probes succeed and the local file
is smaller than the remote -> re-download. A failed probe
never forces a re-download (legacy bug: one dead HEAD used to
re-download entire playlists).
With receipts on (save_receipts) the receipt is the check:
its recorded size and fingerprint are local and authoritative,
and a missing/unsatisfied receipt always re-downloads (the
receipt is an optimisation, never a reason to keep a file we
cannot prove).
"""
if self._config.save_receipts:
return receipts_mod.satisfied(
dest, receipts_mod.identity("track", track_id), quality
)
try:
local_size = dest.stat().st_size
except OSError:
return False
if local_size <= 0:
return False
sizes = [self._transport.probe_size(url) for url in stream.urls]
if all(size >= 0 for size in sizes):
total = sum(sizes)
if 0 < total and local_size < total:
logger.warning(
"%s: local file (%d bytes) is smaller than remote "
"(%d bytes); re-downloading",
dest.name,
local_size,
total,
)
return False
return True
def _planned_dests(
self,
resource: Resource,
track: Track,
dest_dir: Path,
index: int,
quality: Quality,
) -> list[Path]:
"""Destination candidates from catalog data only (no stream).
Feeds the stream-free skip check in :meth:`_download_item`:
a non-empty file at one of these paths means the track is
already on disk, and the rate-limited playback request can be
skipped entirely. The candidates cover every extension and
quality label a delivery could plausibly produce:
- extension: ``.m4a`` is always possible (AAC delivery, or
FLAC-in-MP4 without remux); ``.flac`` as well when the
playback ladder for this request can deliver lossless and
``save_as_flac`` remuxes it;
- quality label (only when the track template uses
``{quality}``): every quality of the playback attempt ladder
for this request, which is exactly the set the delivery may
land on.
"""
base = dest_dir
if resource.kind == "artist" and track.album is not None:
# Same layout rule as _item_dest (per-album folders).
base = dest_dir / naming.album_dir(track.album, self._config.album_format)
position = track.position or (
index if resource.kind in ("playlist", "mix") else None
)
attempts = playback_attempts(quality, track.media_tags, track.audio_quality)
# Both containers are possible whenever the ladder can deliver
# lossless: a pure-FLAC rendition lands on .flac either way, and
# a FLAC-in-MP4 rendition lands on .m4a (or .flac after remux).
extensions = {".m4a"}
if any(item in LOSSLESS_API_QUALITIES for item in attempts):
extensions.add(".flac")
quality_values: tuple[str | None, ...]
if "{quality}" in self._config.track_format:
quality_values = tuple(attempts)
else:
quality_values = (None,)
candidates: list[Path] = []
seen: set[Path] = set()
for extension in sorted(extensions):
for value in quality_values:
path = naming.track_path(
track,
extension,
base,
self._config.track_format,
album=track.album,
position=position,
quality=value,
)
if path not in seen:
seen.add(path)
candidates.append(path)
return candidates
def _item_dest(
self,
resource: Resource,
track: Track,
stream: StreamInfo,
dest_dir: Path,
index: int,
) -> Path:
codec = stream.codec
extension = naming.track_extension(codec, stream.url)
if media_mod.flac_in_mp4(stream) and not self._config.save_as_flac:
extension = ".m4a"
position = track.position or (
index if resource.kind in ("playlist", "mix") else None
)
base = dest_dir
if resource.kind == "artist" and track.album is not None:
# Artist download: every album lands in its own folder under
# the output root (TMD parity — no artist wrapper folder).
base = dest_dir / naming.album_dir(track.album, self._config.album_format)
return naming.track_path(
track,
extension,
base,
self._config.track_format,
album=track.album,
position=position,
quality=stream.delivered_quality,
)
@staticmethod
def _claim_dest(
dest: Path, track_id: int | str, index: int, claimed: dict[Path, object]
) -> Path:
"""Make ``dest`` unique within this run (data-loss guard).
Two *different* tracks can sanitise to the identical canonical
file name (duplicate titles with the same number). Without a
guard the later item would find the earlier item's file and be
skipped as "file exists" — silently losing an entire track.
Disambiguation is deterministic: the item's position in the
collection, so re-running the same collection maps every item
to the same path and the skip-existing/resume contract holds.
A track repeated in the collection keeps the shared canonical
file (the later occurrence skips, as intended).
"""
candidate = dest
number = index
while claimed.get(candidate) not in (None, track_id):
candidate = dest.with_name(f"{dest.stem} ({number}){dest.suffix}")
number += 1
claimed[candidate] = track_id
return candidate
def _cover_bytes(self, track: Track, resource: Resource) -> bytes | None:
if not self._config.download_cover:
return None
artwork = None
album = track.album
if album is not None and album.artwork_id:
artwork = album.artwork_id
elif resource.kind == "album" and isinstance(resource.data, Album):
artwork = resource.data.artwork_id
if not artwork:
return None
try:
url = artwork_url(artwork, COVER_WIDTH, COVER_HEIGHT)
return self._transport.get_bytes(url)
except Exception: # noqa: BLE001 - cover art is garnish, never fatal
logger.debug("cover fetch failed for %s", artwork, exc_info=True)
return None
def _composer_names(self, track_id: int | str) -> list[str] | None:
"""The Composer credits for tagging (TMD parity); ``None`` if none."""
names: list[str] = []
for item in self._api.get_track_contributors(track_id):
if item.get("role") == "Composer":
name = item.get("name")
if name and name not in names:
names.append(name)
return names or None
@staticmethod
def _write_lyrics_sidecar(dest: Path, lyrics: dict[str, str]) -> bool:
"""Write the lyrics next to the media file (best-effort).
``.lrc`` when timed subtitles exist (most players and DJ tools
understand them), ``.txt`` for plain text. Written via a temp
file + ``os.replace`` so a crash mid-write never leaves a torn
sidecar; a write failure only logs (post-processing contract:
it never fails the item — the media itself is complete).
"""
text = lyrics.get("subtitles") or lyrics.get("lyrics") or ""
if not text:
return False
sidecar = dest.with_suffix(".lrc" if lyrics.get("subtitles") else ".txt")
tmp = sidecar.with_name(f"{sidecar.name}.tmp")
try:
tmp.write_text(text, encoding="utf-8")
os.replace(tmp, sidecar)
return True
except OSError:
logger.warning("could not write lyrics sidecar %s", sidecar, exc_info=True)
try:
tmp.unlink(missing_ok=True)
except OSError: # noqa: BLE001 - cleanup is best-effort
pass
return False
# -- report helpers ------------------------------------------------------------
@staticmethod
def _check_stop(stop_event: threading.Event) -> None:
if stop_event.is_set():
raise CancelledError("download cancelled")
@staticmethod
def _emit(on_progress: ProgressCallback | None, event: dict) -> None:
if on_progress is None:
return
try:
on_progress(event)
except Exception: # noqa: BLE001 - a host callback must not kill the run
logger.warning("progress callback failed", exc_info=True)
@staticmethod
def _record(report: DownloadReport, item: ItemReport) -> None:
report.items.append(item)
def _failed(
self,
index: int,
total: int,
track: Track,
exc: Exception,
code: str | None = None,
on_progress: ProgressCallback | None = None,
) -> ItemReport:
error = (
exc
if isinstance(exc, TidalError)
else TidalError(str(exc), code or "internal")
)
item = ItemReport(
index=index,
kind="track",
id=track.id,
title=track.title,
status="failed",
reason=str(exc),
code=code or error.code,
retryable=getattr(error, "retryable", False),
)
logger.info("item %s failed: %s", track.id, exc)
self._emit(
on_progress,
_base_event(
"",
index,
total,
track,
status="failed",
reason=str(exc),
code=error.code,
),
)
return item
def _skipped(
self,
index: int,
total: int,
track: Track | None,
reason: str,
status: str = "skipped",
kind: str = "track",
) -> ItemReport:
return ItemReport(
index=index,
kind=kind,
id=track.id if track is not None else None,
title=track.title if track is not None else None,
status=status,
reason=reason,
)