"""Catalog and playback API (v1 gust + v2 OpenAPI fallback).
Tidal runs two catalog APIs in parallel:
- **v1** (``api.tidal.com/v1``) resolves the *legacy* entity ids and
answers playback info;
- **v2** (``openapi.tidal.com/v2``, JSON:API) resolves *both* legacy
and the new entity ids used by the current web app.
So a fetch goes v1 first and, on 404, retries v2 (top-level entities
only — sub-resources like playback info accept new ids directly).
v2 has a hard 20-item cap on album/playlist items with no working
pagination; when the declared item count exceeds what came back we
raise instead of silently enumerating a truncated collection.
"""
from __future__ import annotations
import logging
import re
import time
from collections.abc import Callable, Sequence
from dataclasses import replace
from typing import Any
from urllib.parse import quote, unquote, urlsplit
from . import manifests as mf
from .config import Config
from .errors import (
CancelledError,
ConfigError,
ManifestError,
NotFoundError,
PreviewOnlyError,
QualityUnavailableError,
StreamError,
TidalApiError,
TidalError,
UnsupportedType,
)
from .models import Album, Artist, Playlist, Resource, StreamInfo, Track
from .oauth import OAuth
from .quality import (
Quality,
needs_stereo_fallback,
normalize_codec,
parse_iso8601_duration,
playback_attempts,
quality_meets_profile,
select_representation,
stereo_media_tags,
)
from .transport import Transport
logger = logging.getLogger("tidalpy.api")
# Catalog hosts this library accepts, matched *exactly* (lowercased):
# a link to not-tidal.com must never be reinterpreted as a Tidal id.
_TIDAL_HOSTS = frozenset({"tidal.com", "www.tidal.com", "listen.tidal.com"})
# Schemeless links that carry one of these hosts are a supported form.
_TIDAL_HOST_PREFIX_RE = re.compile(
r"^(?:www\.)?(?:listen\.)?tidal\.com(?=[:/]|$)", re.IGNORECASE
)
# Kind names in a catalog path and the id shape each kind accepts:
# numeric for the catalog ids, opaque (UUID-ish) for user-generated
# ones. Validated before any network call.
_PATH_KINDS = frozenset({"album", "track", "playlist", "video", "mix", "artist"})
_NUMERIC_KINDS = frozenset({"track", "album", "video", "artist"})
_NUMERIC_ID_RE = re.compile(r"[0-9]+\Z")
_OPAQUE_ID_RE = re.compile(r"[A-Za-z0-9_-]+\Z")
_PLAY_TIDAL_MIX_RE = re.compile(r"(?:https?://)?play\.tidal\.com/mc/([A-Za-z0-9_-]+)")
_ARTWORK_IMAGE_ID_RE = re.compile(r"resources\.tidal\.com/images/((?:[^/]+/){4}[^/]+)")
_V2_ITEMS_PAGE_CAP = 20
OPENAPI_MANIFEST_USAGES = ("DOWNLOAD", "PLAYBACK")
# Resolved-stream cache policy. The TTL is deliberately short: signed
# CDN URLs expire in minutes, and a stale URL would surface as a
# download failure while a fresh resolve would have fixed it. The
# bound keeps a long-lived client (an app resolving many tracks) from
# accumulating one entry per track forever.
STREAM_CACHE_TTL_SECONDS = 90.0
STREAM_CACHE_MAX_ITEMS = 256
# A playback quality the account is *not entitled* to (the API answers
# 403 CLIENT_NOT_ENTITLED) is not entitled for every track: the
# rejection is cached briefly so later resolutions skip the doomed
# probe instead of spending one rate-limited request per track
# (reference-fork TTL).
PLAYBACK_BLOCK_TTL_SECONDS = 300.0
[docs]
def parse_tidal_url(url: str) -> tuple[str | None, str | None]:
"""``(kind, id)`` from a Tidal link; ``(None, url)`` when it is
not a Tidal link (bare ids, foreign links).
Host matching is exact — ``tidal.com``, ``www.``, ``listen.`` —
http(s) only, no embedded credentials, ports 80/443 only, any
locale prefix and nesting, query strings and fragments ignored.
A link to *not*-tidal.com therefore never resolves to a Tidal
resource (reference-fork parity). Identifiers are validated per
kind before any network call: numeric for track/album/video/
artist, opaque (UUID-ish) for playlist/mix. Nested paths resolve
to the innermost known kind. ``play.tidal.com/mc/<id>`` yields a
mix.
"""
text = (url or "").strip()
if not text:
return None, None
mix_match = _PLAY_TIDAL_MIX_RE.match(text)
if mix_match and len(mix_match.group(1)) <= 128:
return "mix", mix_match.group(1)
candidate = text
if _TIDAL_HOST_PREFIX_RE.match(candidate):
# Schemeless catalog links are a supported form.
candidate = "https://" + candidate
try:
parsed = urlsplit(candidate)
valid = (
parsed.scheme in ("http", "https")
and (parsed.hostname or "").lower() in _TIDAL_HOSTS
and parsed.username is None
and parsed.password is None
and parsed.port in (None, 80, 443)
)
except ValueError: # malformed port and friends
valid = False
if not valid:
return None, text
parts = [unquote(part) for part in parsed.path.split("/") if part]
lowered = [part.lower() for part in parts]
# The innermost known kind wins; an invalid id for it rejects the
# whole link (no fallback to an outer kind).
for index in range(len(lowered) - 2, -1, -1):
kind = lowered[index]
if kind not in _PATH_KINDS:
continue
identifier = parts[index + 1]
pattern = _NUMERIC_ID_RE if kind in _NUMERIC_KINDS else _OPAQUE_ID_RE
if len(identifier) <= 128 and pattern.fullmatch(identifier):
return kind, identifier
return None, text
return None, text
def _is_foreign_url(text: str) -> bool:
"""True when the text is an http(s) URL for a host that is not a
Tidal host — a paste mistake that should fail fast."""
parsed = urlsplit(text)
return parsed.scheme in ("http", "https") and (
(parsed.hostname or "").lower() not in _TIDAL_HOSTS
)
# The legacy image ids (search rows, Album.artwork_id, artist pictures)
# are five-segment UUIDs; the CDN path is the same id with the dashes
# turned into slashes.
_ARTWORK_ID_RE = re.compile(
r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\Z"
)
[docs]
def artwork_url(artwork_id: str, width: int = 1280, height: int = 1280) -> str:
"""The static-CDN URL for a five-segment artwork id.
The ids carried by search rows (``artwork``), ``Album.artwork_id``
and ``Artist`` pictures are the legacy image ids: five dash-
separated segments. The image path is the same id with the dashes
turned into slashes — the exact pattern the reference tool uses
(tagHelper). Pure URL construction: no network, no auth, safe for
any client to render thumbnails from search rows.
Raises :class:`~tidalpy.errors.ConfigError` for ids that are not
five-segment image ids (a paste mistake or a v2 artwork uuid that
needs the ``artworks/{id}`` lookup first).
"""
text = (artwork_id or "").strip()
if not _ARTWORK_ID_RE.match(text):
raise ConfigError(f"not a Tidal image id: {artwork_id!r}")
if not (1 <= width <= 8192 and 1 <= height <= 8192):
raise ConfigError("artwork dimensions must be 1..8192")
return (
f"https://resources.tidal.com/images/"
f"{text.replace('-', '/')}/{width}x{height}.jpg"
)
def _search_type_name(item_type: str) -> str:
"""Plural API type name; the v1 search endpoint rejects singulars."""
name = item_type.strip().upper()
return name if name.endswith("S") else name + "S"
# The genre-browse endpoint segments (live API verified: ``artists``
# exists on no genre — the 404 matches the ``hasArtists: False``
# flags — so it is not a member). The plural names match both the
# URL segments and the genre rows' ``has_<item_type>`` flags.
GENRE_ITEM_TYPES: tuple[str, ...] = ("tracks", "albums", "playlists", "videos")
# The bulk-fetch page size (the catalog's standard page size — the
# same clamp the search endpoints use): one v2 call per 50 ids.
_BULK_PAGE_SIZE = 50
# The autocomplete row's ``type`` discriminator (live API verified:
# uppercase group names, same vocabulary as the search ``types``
# parameter) -> the uniform search row's lowercase kind.
_AUTOCOMPLETE_KINDS: dict[str, str] = {
"ARTIST": "artist",
"ALBUM": "album",
"TRACK": "track",
"PLAYLIST": "playlist",
"VIDEO": "video",
}
def _search_row(kind: str, item: dict) -> dict[str, Any]:
"""The uniform search row for one catalog item (or ``topHit``).
``{"kind", "id", "title"}`` plus ``artist`` for albums and tracks
(the same rows the reference client's ``search`` RPC/CLI emit —
the ``id`` values build ``https://tidal.com/<kind>/<id>`` links
straight away; playlist ids are the ``uuid``). The catalog's
garnish rides along *when present* (for result-rendering
clients): ``version`` (track/album editions), ``duration`` in
seconds (track/video length, album total), ``quality`` (track/
album, the raw API value — map it with
:func:`tidalpy.quality.quality_label` for display), ``isrc`` and
the album title/artwork for tracks, the album's own cover and
``releaseDate`` for albums, the artist's picture for artists.
"""
if kind == "playlist":
# The link is /playlist/<uuid>; the numeric id is
# not linkable (reference-client parity).
row: dict[str, Any] = {
"kind": kind,
"id": item.get("uuid") or item.get("id"),
"title": item.get("title") or "",
}
elif kind == "artist":
# Artist rows carry the display name in ``name``.
row = {
"kind": kind,
"id": item.get("id"),
"title": item.get("name") or item.get("title") or "",
}
else:
row = {
"kind": kind,
"id": item.get("id"),
"title": item.get("title") or "",
}
if kind in ("album", "track"):
row["artist"] = _search_artists_name(item.get("artists"))
if kind in ("track", "album"):
version = item.get("version")
if isinstance(version, str) and version.strip():
row["version"] = version
if kind in ("track", "album", "video"):
duration = item.get("duration")
if (
isinstance(duration, (int, float))
and not isinstance(duration, bool)
and duration > 0
):
row["duration"] = int(duration)
if kind in ("track", "album"):
audio_quality = item.get("audioQuality")
if isinstance(audio_quality, str) and audio_quality:
row["quality"] = audio_quality
if kind == "track":
# ``isrc`` (the master-recording id) and the album
# title: the timed-lyrics fallback scores candidates
# by them (same ISRC = same master recording), and
# they are useful display facts for search clients.
isrc = item.get("isrc")
if isinstance(isrc, str) and isrc.strip():
row["isrc"] = isrc
album = item.get("album")
if isinstance(album, dict):
if album.get("title"):
row["album"] = album["title"]
if isinstance(album.get("cover"), str):
row["artwork"] = album["cover"]
# Availability facts (live API verified on search rows):
# a pay-to-stream track is surfaced only when it is one
# (an indexer badges the row before attempting a resolve),
# and audio modes only when they are non-stereo (a 5.1/
# ATMOS badge; STEREO is the norm, not a fact worth a key).
if item.get("payToStream") is True:
row["payToStream"] = True
audio_modes = [m for m in (item.get("audioModes") or []) if isinstance(m, str)]
if audio_modes and audio_modes != ["STEREO"]:
row["audioModes"] = audio_modes
elif kind == "album":
# Albums carry their own cover (the five-segment
# legacy image id, consumed by artwork_url()).
if isinstance(item.get("cover"), str) and item["cover"]:
row["artwork"] = item["cover"]
# The catalog release date (live API verified:
# "1997-07-01" shape): album cards in a real-time
# search UI show the year.
if isinstance(item.get("releaseDate"), str) and item["releaseDate"].strip():
row["releaseDate"] = item["releaseDate"]
# The catalog type (live API verified: "ALBUM"/"SINGLE"/
# ...): the "Single"/"EP" badge of an album card.
if isinstance(item.get("type"), str) and item["type"]:
row["type"] = item["type"]
# The purchase-required badge (live API verified on album
# rows): surfaced only when true, as on track rows.
if item.get("payToStream") is True:
row["payToStream"] = True
elif kind == "artist":
if isinstance(item.get("picture"), str) and item["picture"]:
row["artwork"] = item["picture"]
return row
def _top_hit_kind(top_type: Any) -> str | None:
"""The catalog's ``topHit.type`` name (``"ARTISTS"``, plural)
mapped to the row ``kind`` (``"artist"``); ``None`` when the
catalog names a group the row builder doesn't know."""
if not isinstance(top_type, str):
return None
kind = top_type.strip().lower().rstrip("s")
known = ("artist", "album", "track", "video", "playlist")
return kind if kind in known else None
def _search_artists_name(artists: Any) -> str:
"""Display names of a search row's ``artists`` list (", "-joined,
reference-client parity; entries without a name are skipped)."""
if not isinstance(artists, list):
return ""
return ", ".join(
a["name"] for a in artists if isinstance(a, dict) and a.get("name")
)
def _normalize_match_text(value: Any) -> str:
"""Case/whitespace-insensitive comparison key for lyrics matching."""
return " ".join(str(value or "").casefold().split())
def _optional_number(value: Any) -> int | float | None:
"""A JSON number or ``None`` (strings/booleans/None are not numbers)."""
if isinstance(value, bool) or not isinstance(value, (int, float)):
return None
return value
def _optional_text(value: Any) -> str | None:
"""A non-empty string or ``None`` (the catalog omits absent facts)."""
if isinstance(value, str) and value.strip():
return value
return None
def _score_lyrics_candidate(track: Track, candidate: dict) -> int:
"""How well a search candidate matches ``track`` as the same song
in another version (reference-fork scoring; ``< 0`` rejects).
A candidate must share the normalized title *and* at least one
artist name; the score then rewards the stronger signals:
identical ISRC (same master recording), near-identical duration,
same album, same lead artist, more artist overlap.
"""
if candidate.get("id") == track.id:
return -1
track_title = _normalize_match_text(track.title)
candidate_title = _normalize_match_text(candidate.get("title"))
if not track_title or track_title != candidate_title:
return -1
track_artists = {_normalize_match_text(a.name) for a in track.artists if a.name}
# Search rows carry the ", "-joined display names (the same
# convention the reference client joins with); the order is the
# credit order, so the first entry is the lead artist.
candidate_artists = [
_normalize_match_text(name)
for name in str(candidate.get("artist") or "").split(", ")
if name.strip()
]
candidate_set = set(candidate_artists)
if not track_artists or not candidate_set:
return -1
overlap = track_artists & candidate_set
if not overlap:
return -1
score = 100 + min(len(overlap), 3) * 20
# Lead-artist order: the first name is the primary credit.
track_first = next((a.name for a in track.artists if a.name), None)
if track_first and candidate_artists:
if track_first.casefold() == candidate_artists[0].casefold():
score += 15
track_isrc = str(track.isrc or "").strip()
candidate_isrc = str(candidate.get("isrc") or "").strip()
if track_isrc and candidate_isrc:
score += 50 if track_isrc == candidate_isrc else -5
track_duration = track.duration or 0
candidate_duration = candidate.get("duration") or 0
if isinstance(candidate_duration, (int, float)) and not isinstance(
candidate_duration, bool
):
delta = abs(track_duration - int(candidate_duration))
if track_duration:
if delta <= 2:
score += 20
elif delta <= 5:
score += 10
track_album = _normalize_match_text(
track.album.title if track.album is not None else None
)
candidate_album = _normalize_match_text(candidate.get("album"))
if track_album and candidate_album and track_album == candidate_album:
score += 10
return score
def _artist_from_v1(data: dict | None) -> Artist | None:
if not isinstance(data, dict):
return None
return Artist(
id=data.get("id"),
name=data.get("name"),
picture=data.get("picture"),
popularity=float(_optional_number(data.get("popularity")) or 0.0),
handle=_optional_text(data.get("handle")),
mixes=_v1_mixes(data.get("mixes")),
)
def _artists_from_v1(data: dict | None) -> list[Artist]:
if not isinstance(data, list):
return []
artists = [_artist_from_v1(item) for item in data]
return [a for a in artists if a is not None]
def _v1_track_replaygain(data: dict) -> dict[str, str]:
"""The v1 track's own loudness facts (track scope only).
The v1 payload carries ``replayGain`` (EBU R128 track gain, dB)
and ``peak`` (true peak, 0..1) on the track object itself (live
API verified — on every items entry, not just the track
endpoint). The tag writers want the same pre-formatted strings as
the v2 normalization data produces (``+X.XX dB`` gains, plain
floats for peaks); both consumers parse them identically.
"""
facts: dict[str, str] = {}
gain = _optional_number(data.get("replayGain"))
if gain is not None:
facts["trackReplayGain"] = f"{float(gain):+.2f} dB"
peak = _optional_number(data.get("peak"))
if peak is not None:
facts["trackPeakAmplitude"] = str(peak)
return facts
def _v1_audio_modes(raw: Any) -> tuple[str, ...]:
"""The catalog's audio modes (a list of str); anything else is absent.
A masquerading string must not be iterated character by character
(``"STEREO"`` is not ``("S", "T", "E", ...)``).
"""
if not isinstance(raw, (list, tuple)):
return ()
return tuple(m for m in raw if isinstance(m, str))
def _v1_promoted_artists(raw: Any) -> list[Artist]:
"""The playlist's promoted artists (live API verified, v1 only).
A list of ``{"id", "name", "handle", "picture", "type"}`` rows —
reuses the :class:`~tidalpy.models.Artist` model (the row's
``type`` badge, ``MAIN``, is dropped: it is not a fact a host
needs). Rows without a name are unusable for a card and are
skipped.
"""
if not isinstance(raw, list):
return []
artists = []
for item in raw:
name = item.get("name") if isinstance(item, dict) else None
if not (isinstance(name, str) and name):
continue
artists.append(
Artist(
id=item.get("id"),
name=item.get("name"),
picture=item.get("picture"),
handle=_optional_text(item.get("handle")),
)
)
return artists
def _v1_mixes(raw: Any) -> dict[str, str]:
"""The catalog's editorial-mix map (mix kind -> mix id); anything
else is absent. Live API verified on v1 track and artist objects
(v2 carries no mixes): ``{"TRACK_MIX": "<id>"}`` /
``{"ARTIST_MIX": "<id>"}`` — the ids build
``https://play.tidal.com/mc/<id>`` links, which resolve through
the normal mix path."""
if not isinstance(raw, dict):
return {}
return {k: v for k, v in raw.items() if isinstance(k, str) and isinstance(v, str)}
def _v1_track(data: dict) -> Track:
artists = _artists_from_v1(data.get("artists"))
primary = _artist_from_v1(data.get("primaryArtist"))
if primary is not None and (not artists or artists[0].id != primary.id):
artists.insert(0, primary)
media = data.get("mediaMetadata") or {}
album_data = data.get("album")
album = None
if isinstance(album_data, dict) and album_data.get("id") is not None:
album = _v1_album(album_data)
return Track(
id=data.get("id"),
title=data.get("title"),
version=data.get("version"),
duration=int(data.get("duration") or 0),
track_number=int(data.get("trackNumber") or 0),
volume_number=int(data.get("volumeNumber") or 0),
bpm=_optional_number(data.get("bpm")),
key=_optional_text(data.get("key")),
key_scale=_optional_text(data.get("keyScale")),
position=int(data.get("trackNumberOnPlaylist") or 0),
explicit=bool(data.get("explicit")),
isrc=data.get("isrc"),
copyright_text=(
(data.get("copyright") or {}).get("text")
if isinstance(data.get("copyright"), dict)
else None
),
audio_quality=data.get("audioQuality"),
media_tags=list(media.get("tags") or []),
artists=artists,
album=album,
replaygain=_v1_track_replaygain(data),
pay_to_stream=bool(data.get("payToStream") is True),
audio_modes=_v1_audio_modes(data.get("audioModes")),
stream_start_date=_optional_text(data.get("streamStartDate")),
mixes=_v1_mixes(data.get("mixes")),
)
def _v1_album(data: dict) -> Album:
artists = _artists_from_v1(data.get("artists"))
primary = _artist_from_v1(data.get("primaryArtist"))
if primary is not None and (not artists or artists[0].id != primary.id):
artists.insert(0, primary)
return Album(
id=data.get("id"),
title=data.get("title"),
version=data.get("version"),
release_date=data.get("releaseDate"),
# The v1 catalog's album copyright is a PLAIN STRING (live API
# verified: "XL Recordings Ltd") — the old dict-shaped read
# left every live album's label empty (the v1 *track* object
# is the one that carries a dict). A masquerading dict
# degrades to absent.
copyright_label=_optional_text(data.get("copyright")),
number_of_tracks=int(data.get("numberOfTracks") or 0),
number_of_volumes=int(data.get("numberOfVolumes") or 0),
# The v1 key is ``type`` (live API verified: "ALBUM"/"SINGLE";
# the v2 JSON:API attributes block is the one that uses
# ``albumType``) — reading ``albumType`` here left every v1
# album's type empty.
album_type=data.get("type"),
explicit=bool(data.get("explicit")),
audio_quality=data.get("audioQuality"),
artwork_id=data.get("cover"),
artists=artists,
popularity=float(_optional_number(data.get("popularity")) or 0.0),
upc=_optional_text(data.get("upc")),
vibrant_color=_optional_text(data.get("vibrantColor")),
duration=parse_iso8601_duration(data.get("duration")),
pay_to_stream=bool(data.get("payToStream") is True),
stream_start_date=_optional_text(data.get("streamStartDate")),
)
def _v1_playlist_creator(data: dict) -> str | None:
"""The creator's display name (v1 playlists only).
User playlists carry ``{"id", "name", "username"}``; editorial
ones carry ``{"id": 0}`` (Tidal itself) — the name is absent, so
the field is ``None`` rather than a bogus "0".
"""
creator = data.get("creator")
if not isinstance(creator, dict):
return None
return _optional_text(creator.get("name")) or _optional_text(
creator.get("username")
)
def _v1_playlist(data: dict) -> Playlist:
return Playlist(
id=data.get("uuid") or data.get("id"),
title=data.get("title") or data.get("name"),
description=data.get("description"),
number_of_tracks=int(data.get("numberOfTracks") or 0),
explicit=bool(data.get("explicit")),
# ``image`` is the five-segment artwork id (the same shape the
# album's ``cover`` and the search rows' ``picture`` carry).
artwork_id=_optional_text(data.get("image")),
creator=_v1_playlist_creator(data),
creation_time=_optional_text(data.get("created")),
last_modified=_optional_text(data.get("lastUpdated")),
# The playlist's featured artist(s) (live API verified:
# [{"id", "name", "handle", "picture", "type": "MAIN"}]) —
# a playlist card's attribution, and artist-link deep links.
promoted_artists=_v1_promoted_artists(data.get("promotedArtists")),
last_item_added=_optional_text(data.get("lastItemAddedAt")),
duration=int(data.get("duration") or 0),
playlist_type=_optional_text(data.get("type")),
follower_count=int(data.get("fanCount") or 0),
popularity=float(_optional_number(data.get("popularity")) or 0.0),
is_public=(
bool(data["publicPlaylist"])
if isinstance(data.get("publicPlaylist"), bool)
else None
),
)
def _copy_stream(stream: StreamInfo) -> StreamInfo:
"""A copy with the mutable fields (urls list, replaygain dict)
detached, so a caller mutating its result cannot corrupt the
stream cache's shared resolution (reference-fork parity)."""
return replace(stream, urls=list(stream.urls), replaygain=dict(stream.replaygain))
[docs]
class TidalApi:
"""Stateless catalog/playback access on top of :class:`Transport`."""
[docs]
def __init__(self, config: Config, oauth: OAuth, transport: Transport) -> None:
self._config = config
self._oauth = oauth
self._transport = transport
# Per-instance item cache: resolve() then download() must not
# re-enumerate the same collection. No global state. Keys are
# (kind, id[, scope flag]) so differently-scoped artist
# listings can coexist.
self._items_cache: dict[tuple, list[Track | None]] = {}
# Short-lived resolved-stream cache (insertion-ordered, LRU by
# re-insertion): a duplicate track in a playlist, or a host
# previewing the same track twice, must not spend a second
# rate-limited playback request. The key carries the account
# identity, so adopting a different login on this client
# cannot serve the previous account's streams. Entries are
# stored as copies; hits return copies, so no caller can
# corrupt a shared resolution.
self._stream_cache: dict[tuple, tuple[float, StreamInfo]] = {}
# Playback qualities this account is not entitled to, as
# (account identity, audioquality) -> (deadline, rejection):
# skipping them saves one rate-limited probe per track for
# the TTL window, and the stored rejection keeps the eventual
# error actionable (the original 403, not "no stream found").
# Only capability rejections are cached — a track-specific
# 403 must not taint other tracks.
self._blocked_params: dict[tuple, tuple[float, TidalApiError]] = {}
# ------------------------------------------------------------------
# session info
# ------------------------------------------------------------------
[docs]
def subscription(self) -> dict:
record = self._oauth.record
if record is None or not record.user_id:
return {}
try:
body = self._transport.get_json(f"users/{record.user_id}/subscription")
except TidalError:
# Subscription facts are garnish for auth_status(); any
# rejection degrades to "unknown" instead of failing the caller.
return {}
return body if isinstance(body, dict) else {}
# ------------------------------------------------------------------
# v1 catalog (with v2 fallback on 404)
# ------------------------------------------------------------------
[docs]
def get_track(self, track_id: int | str) -> Track:
try:
data = self._transport.get_json(f"tracks/{track_id}")
return _v1_track(data)
except NotFoundError:
pass
data = self._v2_get(f"tracks/{track_id}", {"include": "artists,albums"})
track = self._v2_track(data)
# The v2 track resource carries no position; resolve it from the
# album's item list (the v2 item meta) so a single-track download
# gets the real album number in the file name (reference-client
# parity: TMD's getTrack does the same follow-up).
album = track.album
if (
track.id is not None
and not track.track_number
and album is not None
and album.id is not None
):
try:
_album, items = self._v2_album_with_items(album.id)
except TidalError as exc:
logger.debug(
"could not resolve position of track %s in album %s: %s",
track.id,
album.id,
exc,
)
items = []
for item in items:
if item is not None and item.id == track.id:
if item.track_number:
track.track_number = item.track_number
if item.volume_number and not track.volume_number:
track.volume_number = item.volume_number
break
return track
[docs]
def get_tracks_by_ids(self, track_ids: Sequence[int | str]) -> list[Track]:
"""Bulk track fetch (live API verified, v2 only): N ids in,
one catalog call per 50 — the indexer surface.
A sync job that has track ids from elsewhere (a playlist
export, a library audit, a web-app deep link) resolves them
in one round trip per page, instead of N single-track fetches
(each of which starts as a rate-limited v1 attempt). The v1
catalog has no bulk endpoint (live 404), so this goes straight
to v2 — which knows both the legacy numeric ids and the new
web-app ids.
Result order is the request order (duplicates collapse to the
first occurrence); ids the catalog does not know are skipped,
not fabricated, so a host diffs the result against the request
to find the gaps. No per-track position follow-up (the
single-track fetch does that one; the bulk fetch stays one
call per page).
"""
return self._bulk_get(
"tracks", track_ids, self._bulk_track_row, include="artists,albums"
)
[docs]
def get_track_by_isrc(self, isrc: str) -> Track | None:
"""Resolve a track by its ISRC (live API verified, v2 only):
``tracks?filter[isrc]=`` is the catalog's ISRC index — one
call, the matching track's full row.
The indexer surface for hosts that hold ISRCs from elsewhere
(CDDB, a licensing feed, another catalog's export) instead of
Tidal ids: the v1 catalog cannot answer this (live verified:
an ISRC typed into search returns nothing), so the call goes
straight to v2.
A lookup, not a fetch: the ISRC the catalog does not know
returns ``None`` (the live contract: 200 with an empty row
list, never a 404 — a defensive catch of the unexpected 404
degrades the same way), so a host diffing a library of ISRCs
pays no exception per miss. Blank input returns ``None``
without a request. The input is normalised before the call
(whitespace removed, upper-cased — the catalog stores ISRCs
upper-case and rejects any other case with a 400, live
verified); a malformed ISRC is the catalog's to reject
(``TidalApiError``), which propagates.
When the catalog lists more than one track for the ISRC (the
same recording on several releases), the first is returned.
"""
key = "".join(str(isrc or "").split()).upper()
if not key:
return None
try:
env = self._v2_get(
"tracks", {"filter[isrc]": key, "include": "artists,albums"}
)
except NotFoundError:
return None
artists_by_id, albums_by_id = self._v2_row_context(env)
for row in env.get("data") or []:
if isinstance(row, dict) and row.get("id"):
model = self._bulk_track_row(row, artists_by_id, albums_by_id)
if model is not None:
return model
return None
[docs]
def get_tracks_by_isrcs(self, isrcs: Sequence[str]) -> list[Track]:
"""Bulk ISRC lookup (live API verified, v2 only): N ISRCs in,
one catalog call per 50 — ``filter[isrc]`` accepts a
comma-separated OR list (live verified with two real ISRCs).
The bulk sibling of :meth:`get_track_by_isrc` for hosts that
hold a library of ISRCs (a CDDB export, a licensing feed):
N keys in ``ceil(N/50)`` rate-limited calls instead of N
single lookups. Same contract as :meth:`get_tracks_by_ids`:
request order, duplicates collapse, keys the catalog does
not know are skipped, not fabricated — diff the result
against the request to find the gaps.
Each ISRC is normalised as in :meth:`get_track_by_isrc`
(whitespace removed, upper-cased); blank entries are
skipped. When the catalog lists more than one track for an
ISRC (the same recording on several releases), the first is
returned for that key.
"""
keys = ["".join(str(item or "").split()).upper() for item in isrcs]
return self._bulk_get(
"tracks",
keys,
self._bulk_track_row,
include="artists,albums",
filter_key="filter[isrc]",
key_from_row=lambda row: (row.get("attributes") or {}).get("isrc"),
)
[docs]
def get_albums_by_ids(self, album_ids: Sequence[int | str]) -> list[Album]:
"""Bulk album fetch (live API verified, v2 only): N ids in,
one catalog call per 50.
The album counterpart of :meth:`get_tracks_by_ids` (an
indexer importing a discography from ids). Same contract:
request order, duplicates collapse, unknown ids skipped not
fabricated. v2-only for the same reason (the v1 catalog has
no bulk endpoint).
The v2 bulk rows carry no ``coverArt`` relationship (live
verified: the single-album resource does), so bulk albums
have no artwork — a host that needs the cover for a card
resolves that album singly, or uses the row artwork of the
search/browse that surfaced the id.
"""
return self._bulk_get(
"albums", album_ids, self._bulk_album_row, include="artists"
)
[docs]
def get_album_by_upc(self, upc: str) -> Album | None:
"""Resolve an album by its UPC (live API verified, v2 only):
``albums?filter[barcodeId]=`` is the catalog's barcode index
— one call, the matching album's full row.
The album counterpart of :meth:`get_track_by_isrc` for hosts
that hold UPCs from elsewhere (the back of a CD case, a
licensing feed) instead of Tidal ids: the v1 catalog cannot
answer this (live verified: a UPC typed into search returns
nothing), so the call goes straight to v2, where the UPC is
the ``barcodeId`` attribute — the same value the v1 album's
``upc`` carries.
A lookup, not a fetch: the barcode the catalog does not know
returns ``None`` (the live contract: 200 with an empty row
list; a defensive catch of the unexpected 404 degrades the
same way), so a host diffing a library of barcodes pays no
exception per miss. Blank input returns ``None`` without a
request. The input is stripped and otherwise passed through
(UPCs run 12-14 digits, no reformatting); a malformed value
is the catalog's to reject (``TidalApiError``), which
propagates.
The match is exact on the stored barcode (live verified: a
prefix matches nothing), but placeholder barcodes are shared
by real catalog entries (a barcode of zeros and the classic
1234567890128 each match several junk albums, live
verified) — when the catalog lists more than one, the first
is returned; a host that cares diffs the title.
"""
key = str(upc or "").strip()
if not key:
return None
try:
env = self._v2_get(
"albums", {"filter[barcodeId]": key, "include": "artists"}
)
except NotFoundError:
return None
artists_by_id, _albums_by_id = self._v2_row_context(env)
for row in env.get("data") or []:
if isinstance(row, dict) and row.get("id"):
model = self._bulk_album_row(row, artists_by_id, {})
if model is not None:
return model
return None
[docs]
def get_albums_by_upcs(self, upcs: Sequence[str]) -> list[Album]:
"""Bulk UPC lookup (live API verified, v2 only): N UPCs in,
one catalog call per 50 — ``filter[barcodeId]`` accepts a
comma-separated OR list.
The bulk sibling of :meth:`get_album_by_upc` for hosts that
hold a library of barcodes (the back of a shelf of CD cases,
a licensing feed): N keys in ``ceil(N/50)`` rate-limited
calls instead of N single lookups. Same contract as
:meth:`get_albums_by_ids`: request order, duplicates
collapse, barcodes the catalog does not know are skipped,
not fabricated — diff the result against the request to
find the gaps.
Each UPC is stripped and otherwise passed through (no
reformatting); blank entries are skipped. Rows are matched
back to the request by their stored ``barcodeId`` — when the
catalog's matching normalises a placeholder barcode
differently than the stored value (live verified: twelve
zeros also answer with rows whose barcodeId carries
thirteen), such a row cannot be attributed to a key and is
dropped, the same as a miss.
"""
keys = [str(item or "").strip() for item in upcs]
return self._bulk_get(
"albums",
keys,
self._bulk_album_row,
include="artists",
filter_key="filter[barcodeId]",
key_from_row=lambda row: (row.get("attributes") or {}).get("barcodeId"),
)
[docs]
def get_artists_by_ids(self, artist_ids: Sequence[int | str]) -> list[Artist]:
"""Bulk artist fetch (live API verified, v2 only): N ids in,
one catalog call per 50.
The artist counterpart of :meth:`get_tracks_by_ids` — also
the only v2 artist resolution path (the single artist fetch
is v1), so web-app ids that the v1 catalog does not know are
resolvable here. Same contract: request order, duplicates
collapse, unknown ids skipped not fabricated.
The v2 bulk rows carry no picture/handle (live verified: the
attributes block is name/popularity/numberOfFollowers), so
those stay ``None`` on bulk artists.
"""
return self._bulk_get("artists", artist_ids, self._bulk_artist_row)
[docs]
def get_playlists_by_ids(self, playlist_ids: Sequence[int | str]) -> list[Playlist]:
"""Bulk playlist fetch (live API verified, v2 only): N
playlist UUIDs in, one catalog call per 50.
The playlist counterpart of :meth:`get_tracks_by_ids` — the
ids are the linkable playlist UUIDs (the v2 playlist
resource is uuid-keyed; editorial mixes are a v1 concept
and not reachable here). Same contract: request order,
duplicates collapse, unknown UUIDs skipped not fabricated —
diff the result against the request to find the gaps.
The rows carry the playlist card facts (the v2 title key is
``name``, the duration is ISO-8601, the track count is
``numberOfTrackItems``) but no artwork: resolving the row's
v2 artwork id costs one extra catalog call per row, which
would defeat the bulk fetch — a host that needs a cover
resolves that playlist singly (the same contract as bulk
albums).
"""
return self._bulk_get("playlists", playlist_ids, self._bulk_playlist_row)
def _bulk_get(
self,
resource: str,
ids: Sequence[int | str],
row_to_model: Callable[[dict, dict[Any, Artist], dict[Any, dict]], Any | None],
include: str | None = None,
filter_key: str = "filter[id]",
key_from_row: Callable[[dict], str | None] | None = None,
) -> list[Any]:
"""The shared bulk-fetch engine (a v2 ``?filter[...]=`` OR
list — live API verified on all four resource types and on
the ISRC/barcodeId key filters): dedupe + order, chunk at
the page size, map each row, return in request order with
unknown keys skipped.
Rows are attributed back to the request by the entity id by
default (the ``filter[id]`` fetches, where the key IS the
id); ``key_from_row`` overrides the attribution for the
key-based lookups, where the key is a catalog attribute
(``isrc`` / ``barcodeId``) and several rows may carry the
same key (first row wins)."""
keys: list[str] = []
seen: set[str] = set()
for item in ids:
if item is None:
continue
key = str(item).strip()
if key and key not in seen:
seen.add(key)
keys.append(key)
if not keys:
return []
by_key: dict[str, Any] = {}
for start in range(0, len(keys), _BULK_PAGE_SIZE):
chunk = keys[start : start + _BULK_PAGE_SIZE]
params: dict[str, str] = {filter_key: ",".join(chunk)}
if include is not None:
params["include"] = include
body = self._v2_get(resource, params)
artists_by_id, albums_by_id = self._v2_row_context(body)
for row in body.get("data") or []:
if not isinstance(row, dict) or not row.get("id"):
continue
model = row_to_model(row, artists_by_id, albums_by_id)
if model is None:
continue
if key_from_row is not None:
row_key = key_from_row(row)
else:
row_key = str(row["id"])
if row_key is not None and row_key not in by_key:
by_key[row_key] = model
return [by_key[key] for key in keys if key in by_key]
def _bulk_album_row(
self,
row: dict,
artists_by_id: dict[Any, Artist],
albums_by_id: dict[Any, dict],
) -> Album:
"""One bulk album row -> Album (the shared v2 field mapping;
no coverArt relationship on bulk rows, so no artwork)."""
rels = row.get("relationships") or {}
rel_artists = (rels.get("artists") or {}).get("data") or []
artists = [
artists_by_id[rel["id"]]
for rel in rel_artists
if isinstance(rel, dict) and rel.get("id") in artists_by_id
]
return self._v2_album_from_attrs(
row.get("id"), row.get("attributes") or {}, None, artists
)
def _bulk_track_row(
self,
row: dict,
artists_by_id: dict[Any, Artist],
albums_by_id: dict[Any, dict],
) -> Track:
"""One bulk track row -> Track (the shared v2 field mapping)."""
rels = row.get("relationships") or {}
rel_artists = (rels.get("artists") or {}).get("data") or []
rel_albums = (rels.get("albums") or {}).get("data") or []
artists = [
artists_by_id[rel["id"]]
for rel in rel_artists
if isinstance(rel, dict) and rel.get("id") in artists_by_id
]
first_album = (
rel_albums[0].get("id")
if rel_albums and isinstance(rel_albums[0], dict)
else None
)
album_resource = (
albums_by_id.get(first_album) if first_album is not None else None
)
album = (
self._v2_album_from_resource(album_resource, artists)
if album_resource is not None
else None
)
return self._v2_track_from_data(row, artists, album)
@staticmethod
def _bulk_artist_row(
row: dict,
_artists_by_id: dict[Any, Artist],
_albums_by_id: dict[Any, dict],
) -> Artist | None:
"""One bulk artist row -> Artist (the v2 artist resource IS
the row — there is no relationship to resolve)."""
return TidalApi._v2_artist_from_resource(row)
@staticmethod
def _bulk_playlist_row(
row: dict,
_artists_by_id: dict[Any, Artist],
_albums_by_id: dict[Any, dict],
) -> Playlist:
"""One bulk playlist row -> Playlist (the shared v2 field
mapping: the v2 playlist's title is ``name``, the duration
is ISO-8601, the track count is ``numberOfTrackItems``).
No artwork: the row's coverArt relationship carries a v2
artwork id, and resolving it to the image id costs one extra
catalog call per row — which would defeat the bulk fetch,
so a host that needs a cover resolves that playlist singly
(the same contract as bulk albums).
"""
attrs = row.get("attributes") or {}
return Playlist(
id=row.get("id"),
title=attrs.get("title") or attrs.get("name"),
description=attrs.get("description"),
number_of_tracks=int(attrs.get("numberOfTrackItems") or 0),
explicit=bool(attrs.get("explicit")),
artwork_id=None,
creation_time=_optional_text(attrs.get("createdAt")),
last_modified=_optional_text(attrs.get("lastModifiedAt")),
duration=parse_iso8601_duration(attrs.get("duration")),
playlist_type=_optional_text(attrs.get("playlistType")),
follower_count=int(attrs.get("numberOfFollowers") or 0),
)
[docs]
def get_lyrics(
self, track_id: int | str, track: Track | None = None
) -> dict[str, str]:
"""The track's lyrics: ``{"lyrics", "subtitles"}``, each ``""``
when Tidal has none.
``lyrics`` is the plain text, ``subtitles`` the timed (LRC)
version. Lyrics are garnish: the provider service rejects
tracks without lyrics (a 400, not a 404) and any other
rejection — rate limit, stale session — degrades to the empty
strings instead of an error (a cancellation is not a
rejection: it propagates).
When ``track`` is given (a :class:`~tidalpy.models.Track` with
title/artists) and Tidal has *no timed subtitles* for the
track itself, a fallback searches the catalog for matching
track versions (remaster, live, radio edit — same title and
artists, scored by ISRC/duration/album) and adopts the first
candidate's timed lyrics, keeping the primary plain text
(reference-fork parity: the plain text of a version can drift,
the timing is what a display client needs). The fallback only
runs when it can be scored — a title-less ``track_id`` call
costs no extra requests.
"""
primary = self._lyrics_fetch(track_id)
if track is None or primary["subtitles"]:
return primary
timed = self._find_timed_lyrics(track)
if timed is None:
return primary
return {
"lyrics": primary["lyrics"] or timed["lyrics"],
"subtitles": timed["subtitles"],
}
def _lyrics_fetch(self, track_id: int | str) -> dict[str, str]:
try:
data = self._transport.get_json(f"tracks/{track_id}/lyrics")
except CancelledError:
raise
except TidalError:
return {"lyrics": "", "subtitles": ""}
result: dict[str, str] = {}
for field in ("lyrics", "subtitles"):
value = data.get(field)
text = value.strip() if isinstance(value, str) else ""
result[field] = text
return result
def _find_timed_lyrics(self, track: Track) -> dict[str, str] | None:
"""The scored-candidate search for a track without timed
lyrics (see :meth:`get_lyrics`); ``None`` when no candidate
carries timed subtitles."""
title = track.title
if not title or not title.strip():
return None
artist_names = [a.name for a in track.artists if a.name]
# One query per artist (the reference cap), then the bare
# title: a feature-heavy track may only be findable as one
# of its artists.
queries = [f"{title} {artist}" for artist in artist_names[:5]]
if len(artist_names) > 1:
queries.append(f"{title} {' '.join(artist_names[:2])}")
queries.append(title)
candidates: dict[int | str, dict] = {}
for query in dict.fromkeys(q for q in queries if q.strip()):
try:
items, _total, _top_hit = self.search(query, ("track",), limit=10)
except CancelledError:
raise
except TidalError as exc:
logger.debug("lyrics fallback search failed (%s): %s", query, exc)
continue
for item in items:
item_id = item.get("id")
if item_id is not None:
candidates[item_id] = item
scored = sorted(
(
(_score_lyrics_candidate(track, item), item)
for item in candidates.values()
),
key=lambda pair: pair[0],
reverse=True,
)
for _score, candidate in scored:
if _score < 0:
break # sorted descending: the rest are worse
lyrics = self._lyrics_fetch(candidate["id"])
if lyrics["subtitles"]:
return lyrics
return None
[docs]
def get_track_contributors(self, track_id: int | str) -> list[dict[str, Any]]:
"""The track's credited contributors, or ``[]`` when unavailable.
Garnish for tagging (TMD parity: the ``Composer`` credit lands
in the audio tags): any failure degrades to an empty list.
Each item is a ``{"name": str, "role": str}`` dict.
"""
try:
data = self._transport.get_json(f"tracks/{track_id}/contributors")
except TidalError:
return []
items = data.get("items")
if not isinstance(items, list):
return []
return [
item
for item in items
if isinstance(item, dict) and item.get("name") and item.get("role")
]
[docs]
def get_album(
self, album_id: int | str, with_items: bool = True
) -> tuple[Album, list[Track | None]]:
try:
data = self._transport.get_json(f"albums/{album_id}")
album = _v1_album(data)
items = self._v1_album_items(album_id) if with_items else []
return album, items
except NotFoundError:
pass
album, items = self._v2_album_with_items(album_id)
return album, items
[docs]
def get_playlist(
self, playlist_id: int | str, is_mix: bool = False
) -> tuple[Playlist, list[Track | None]]:
path = ("mixes" if is_mix else "playlists") + f"/{playlist_id}"
try:
data = self._transport.get_json(path)
playlist = _v1_playlist(data)
playlist.is_mix = is_mix
items = self._v1_playlist_items(playlist_id, is_mix)
return playlist, items
except NotFoundError:
pass
playlist, items = self._v2_playlist_with_items(playlist_id, is_mix)
return playlist, items
[docs]
def get_artist(self, artist_id: int | str) -> Artist:
"""Artist metadata (v1)."""
data = self._transport.get_json(f"artists/{artist_id}")
return _artist_from_v1(data) or Artist(
id=artist_id,
)
[docs]
def get_artist_albums(
self, artist_id: int | str, include_ep: bool = False
) -> list[Album]:
"""The artist's discography (full albums; EPs/singles opt-in).
Mirrors the reference client: the normal album listing plus,
when ``include_ep`` is set, a second listing with the
``EPSANDSINGLES`` filter.
"""
albums = self._artist_albums_page(artist_id)
if include_ep:
albums.extend(self._artist_albums_page(artist_id, eps_and_singles=True))
return albums
[docs]
def get_artist_top_tracks(
self, artist_id: int | str, limit: int = 20
) -> list[Track]:
"""The artist's top tracks, ranked by the catalog (v1
``artists/{id}/toptracks``; live API verified: paged, full v1
track payloads, none of the reference clients expose it).
A single ranked page: a real-time "top tracks" tab wants the
top N, and the endpoint already orders them — the host pages
with repeated calls if it needs more.
"""
data = self._transport.get_json(
f"artists/{artist_id}/toptracks",
{"limit": min(max(1, limit), 50), "offset": 0},
)
tracks: list[Track] = []
for item in data.get("items") or []:
if not isinstance(item, dict):
continue
track = _v1_track(item)
tracks.append(self._refresh_stale_snapshot(track, item))
return tracks
def _artist_albums_page(
self, artist_id: int | str, eps_and_singles: bool = False
) -> list[Album]:
albums: list[Album] = []
offset = 0
limit = 100
while True:
params: dict[str, Any] = {"limit": limit, "offset": offset}
if eps_and_singles:
params["filter"] = "EPSANDSINGLES"
data = self._transport.get_json(f"artists/{artist_id}/albums", params)
page = data.get("items") or []
albums.extend(_v1_album(item) for item in page if isinstance(item, dict))
offset += len(page)
if len(page) < limit:
return albums
total = data.get("totalNumberOfItems")
if isinstance(total, int) and total > 0 and offset >= total:
return albums
def _items_key(self, *parts: Any) -> tuple:
"""Items-cache key, keyed on the account identity.
Collection contents are per-account (a private playlist, a
region-specific catalog), so the same id under a different
login must not serve the previous account's cached items —
the same rule the stream cache applies.
"""
return (self._account_identity(),) + tuple(parts)
[docs]
def get_items(self, resource: Resource) -> list[Track | None]:
"""The tracks of a resolved collection (cached per instance).
Albums/playlists/mixes return their own tracks; an artist
returns the tracks of every album in the discography, each with
its ``album`` back-reference set (the pipeline needs it for the
per-album destination folders).
"""
data = resource.data
if isinstance(data, Track):
return [data]
if resource.kind == "album" and isinstance(data, Album):
if data.id is None:
return []
key: tuple = self._items_key("album", data.id)
if key not in self._items_cache:
self._items_cache[key] = self._album_items_fresh(data.id)
return self._items_cache[key]
if resource.kind in ("playlist", "mix") and isinstance(data, Playlist):
if data.id is None:
return []
key = self._items_key(resource.kind, data.id)
if key not in self._items_cache:
_, items = self.get_playlist(data.id, is_mix=data.is_mix)
self._items_cache[key] = items
data.items = self._items_cache[key]
return self._items_cache[key]
if resource.kind == "artist" and isinstance(data, Artist):
if data.id is None:
return []
include_ep = self._config.include_ep
key = self._items_key("artist", data.id, include_ep)
if key not in self._items_cache:
tracks: list[Track | None] = []
for album in self.get_artist_albums(data.id, include_ep=include_ep):
if album.id is None:
continue
_, album_tracks = self.get_album(album.id, with_items=True)
for track in album_tracks:
if track is not None:
track.album = album
tracks.extend(album_tracks)
self._items_cache[key] = tracks
return self._items_cache[key]
return []
def _album_items_fresh(self, album_id: int | str) -> list[Track | None]:
album, items = self.get_album(album_id, with_items=True)
return items
# --- v1 pagination -------------------------------------------------
def _v1_items_page(self, path: str, limit: int = 50) -> list[dict]:
"""One page of a v1 ``/items`` listing (tolerates missing key).
A *first* empty page on a collection the catalog claims is
non-empty (``totalNumberOfItems > 0``) is a degraded listing,
not an empty collection: raise instead of returning ``[]``
(the v2 side has the matching truncation guard; without this,
a live album with a bad items endpoint would download nothing
and report ``ok=False`` with zero items, which the host
cannot distinguish from a genuinely empty playlist).
"""
offset = 0
items: list[dict] = []
while True:
data = self._transport.get_json(path, {"limit": limit, "offset": offset})
page = data.get("items") or []
items.extend(page)
if not page:
total = data.get("totalNumberOfItems")
if not items and isinstance(total, int) and total > 0:
raise TidalApiError(
f"{path}: catalog claims {total} items but returned "
f"none (degraded listing)"
)
return items
offset += len(page)
if len(page) < limit:
return items
total = data.get("totalNumberOfItems")
if isinstance(total, int) and total > 0 and len(items) >= total:
return items
@staticmethod
def _v1_item_payload(item: dict) -> dict:
"""The track data of a v1 ``/items`` entry.
Live entries nest the payload under ``item`` (``{"type":
"track", "item": {...}}``); flat entries are tolerated for
replayed fixtures.
"""
nested = item.get("item")
return nested if isinstance(nested, dict) else item
def _v1_album_items(self, album_id: int | str) -> list[Track | None]:
raw = self._v1_items_page(f"albums/{album_id}/items")
tracks: list[Track | None] = []
for item in raw:
if item.get("type") != "track":
# A non-audio slot (music video): it takes a listing
# slot but is never downloaded. Surfaced as ``None``
# so the reports can record it — silently dropping the
# slot would hide the gap between the catalog's item
# count and the items actually processed.
tracks.append(None)
continue
payload = self._v1_item_payload(item)
track_data = dict(payload)
track_data["trackNumber"] = (
payload.get("trackNumber") or payload.get("position") or 1
)
track = _v1_track(track_data)
tracks.append(self._refresh_stale_snapshot(track, payload))
return tracks
def _v1_playlist_items(
self, playlist_id: int | str, is_mix: bool
) -> list[Track | None]:
path = ("mixes" if is_mix else "playlists") + f"/{playlist_id}/items"
raw = self._v1_items_page(path)
tracks: list[Track | None] = []
number = 0
for item in raw:
if item.get("type") != "track":
tracks.append(None)
continue
payload = self._v1_item_payload(item)
track_data = dict(payload)
# Dense index over the audio items only: video entries take
# playlist slots but never enter the download (the
# reference client numbers tracks by download position —
# same result). The API's slot number would leave gaps.
number += 1
track_data["trackNumberOnPlaylist"] = number
track = _v1_track(track_data)
track = self._refresh_stale_snapshot(track, payload)
tracks.append(track)
return tracks
def _refresh_stale_snapshot(self, track: Track, payload: dict) -> Track:
"""Re-resolve an item whose snapshot is stale.
Tidal's catalog migration leaves stale item snapshots
(``streamReady: false``): the id may still resolve with
fresher metadata, or the release may have been *replaced* —
the old entry deleted, the same recording re-issued under a
brand-new id (both v1 and v2 404 the snapshot id; the web app
404s it too). The recovery ladder (reference-fork parity)
applies exact anchors in order — deterministic, no fuzzy
matching:
1. **ID re-resolution** — a fresh ``get_track(id)`` (v1 -> v2
fallback). Recovers the "snapshot carries the new catalog
id" case.
2. **Exact ISRC lookup** — the snapshot's ISRC is a
primary-key lookup (one recording per ISRC; multiple hits
are parallel catalog entries of the same audio — identical
title/version/tags — and the first is safe). Recovers
replaced releases whose recording kept its ISRC.
3. **Exact composite predicate** — for recordings re-issued
under a *new* ISRC (old ISRC -> zero live entries): v1
search is a candidate *source* only (verbatim title query,
relevance ranking ignored), and every candidate must
match the snapshot's title exactly, its primary artist
exactly, and its duration within 1 s (re-master trim).
Uniqueness is required — an exact album-title tie-break
when the snapshot has an album, otherwise refuse (never
guess).
Each recovery is logged with the anchor used and the id
transition, so every recovered item is auditable. When an
anchor recovers the track, the live record is adopted (its
title/artists drive file names); the collection position and
the snapshot's album back-reference are carried over. When
every anchor misses, the snapshot is kept: the download then
fails for that one item with a clear report (per-item
isolation), and a later re-run sees the refreshed snapshot.
"""
if track.id is None or payload.get("streamReady", True):
return track
recovered: Track | None = None
anchor = ""
for name, attempt in (
("id re-resolution", self._recover_snapshot_by_id),
("ISRC", self._recover_snapshot_by_isrc),
("composite predicate", self._recover_snapshot_by_composite),
):
recovered = attempt(track)
if recovered is not None:
anchor = name
break
if recovered is None:
logger.debug("stale snapshot %s did not recover; keeping it", track.id)
return track
logger.info(
"stale snapshot %s recovered via %s -> %s",
track.id,
anchor,
recovered.id,
)
# The live record does not know the collection position.
recovered.position = track.position
if recovered.album is None:
recovered.album = track.album
return recovered
def _recover_snapshot_by_id(self, track: Track) -> Track | None:
"""Anchor 1: re-resolve the snapshot id (v1 -> v2 fallback)."""
if track.id is None:
return None
try:
return self.get_track(track.id)
except CancelledError:
raise
except TidalError:
return None
def _recover_snapshot_by_isrc(self, track: Track) -> Track | None:
"""Anchor 2: exact ISRC lookup — a primary key, not a search."""
if not track.isrc:
return None
try:
return self.get_track_by_isrc(track.isrc)
except CancelledError:
raise
except TidalError:
return None
def _recover_snapshot_by_composite(self, track: Track) -> Track | None:
"""Anchor 3: the exact composite predicate (see the ladder
docstring). Returns ``None`` when no candidate — or more than
one — satisfies it: refusing to guess is the contract."""
if not track.title:
return None
primary = track.artists[0].name if track.artists else None
if not primary:
return None
try:
candidates = self._v1_search_tracks(track.title, limit=20)
except CancelledError:
raise
except TidalError:
return None
matches = [
cand
for cand in candidates
if cand.title == track.title
and (cand.artists[0].name if cand.artists else None) == primary
and abs(cand.duration - track.duration) <= 1
]
if len(matches) != 1:
# The album tie-break: only when several candidates pass
# and the snapshot names its album.
album_title = track.album.title if track.album else None
if len(matches) >= 2 and album_title:
matches = [
m
for m in matches
if m.album is not None and m.album.title == album_title
]
if len(matches) != 1:
logger.debug(
"stale snapshot %s: refusing to guess "
"(%d candidates match title/artist/duration)",
track.id,
len(matches),
)
return None
return matches[0]
def _v1_search_tracks(self, query: str, limit: int = 20) -> list[Track]:
"""v1 track search as full track objects — the composite
recovery's candidate source (``search`` returns curated rows,
which lack the album back-reference the tie-break needs)."""
params: dict[str, Any] = {
"query": query.strip(),
"types": "TRACKS",
"limit": min(max(1, limit), 50),
"offset": 0,
}
country = self._oauth.country_code()
if country:
params["countryCode"] = country
data = self._transport.get_json("search", params)
if not isinstance(data, dict):
return []
group = data.get("tracks")
if not isinstance(group, dict):
return []
items = [item for item in group.get("items") or [] if isinstance(item, dict)]
return [_v1_track(item) for item in items]
# ------------------------------------------------------------------
# v2 (OpenAPI) catalog
# ------------------------------------------------------------------
def _v2_get(self, path: str, params: dict | None = None) -> dict:
body = self._transport.get_json(
path,
params,
base=self._transport.openapi_base,
accept="application/vnd.api+json",
)
if not isinstance(body, dict):
raise TidalApiError("OpenAPI v2 response is not a JSON object")
return body
@staticmethod
def _v2_data(env: dict) -> dict:
data = env.get("data")
if not isinstance(data, dict):
raise NotFoundError("OpenAPI v2 response has no data resource")
return data
@staticmethod
def _v2_included(env: dict) -> dict[str, list[dict]]:
grouped: dict[str, list[dict]] = {}
for resource in env.get("included") or []:
if not isinstance(resource, dict):
continue
type_key = resource.get("type")
if isinstance(type_key, str):
grouped.setdefault(type_key, []).append(resource)
return grouped
@classmethod
def _v2_artists_by_id(cls, included: dict[str, list[dict]]) -> dict[Any, Artist]:
artists: dict[Any, Artist] = {}
for resource in included.get("artists", []):
artist = cls._v2_artist_from_resource(resource)
if artist is not None:
artists[artist.id] = artist
return artists
def _v2_row_context(self, env: dict) -> tuple[dict[Any, Artist], dict[Any, dict]]:
"""The shared included block -> (artists by id, album resources
by id). Every v2 row fetch (the bulk fetches and the
ISRC/UPC lookups) resolves its rows' relationships against
this, so the row mappers stay the single field mapping."""
included = self._v2_included(env)
artists_by_id = self._v2_artists_by_id(included)
albums_by_id = {
res["id"]: res
for res in included.get("albums", [])
if isinstance(res, dict) and res.get("id")
}
return artists_by_id, albums_by_id
@staticmethod
def _v2_artist_from_resource(resource: dict) -> Artist | None:
"""One v2 artist resource (an included block entry, or a bulk
row — the same object shape) -> Artist; unusable rows (no id)
return ``None``."""
if not isinstance(resource, dict) or not resource.get("id"):
return None
attrs = resource.get("attributes") or {}
return Artist(
id=resource.get("id"),
name=attrs.get("name"),
popularity=float(_optional_number(attrs.get("popularity")) or 0.0),
follower_count=int(attrs.get("numberOfFollowers") or 0),
)
@staticmethod
def _v2_resource_artists(
resource: dict,
artists_by_id: dict[Any, Artist],
fallback: list[Artist] | None = None,
) -> list[Artist]:
entries = ((resource.get("relationships") or {}).get("artists") or {}).get(
"data"
) or []
artists = [
artists_by_id[entry["id"]]
for entry in entries
if isinstance(entry, dict) and entry.get("id") in artists_by_id
]
return artists or (fallback or [])
@staticmethod
def _v2_media_tags_to_quality(tags: list[str] | None) -> str | None:
available = set(tags or [])
if "HI_RES_LOSSLESS" in available or "HIRES_LOSSLESS" in available:
return "HI_RES_LOSSLESS"
if "HI_RES" in available:
return "HI_RES"
if "LOSSLESS" in available:
return "LOSSLESS"
if "HIGH" in available:
return "HIGH"
if "LOW" in available:
return "LOW"
return None
@staticmethod
def _v2_album_from_resource(resource: dict, artists: list[Artist]) -> Album:
"""The album back-reference of a v2 track (single + bulk
fetches share the shape)."""
attrs = resource.get("attributes") or {}
return Album(
id=resource.get("id"),
title=attrs.get("title"),
number_of_tracks=int(attrs.get("numberOfItems") or 0),
number_of_volumes=int(attrs.get("numberOfVolumes") or 0),
release_date=attrs.get("releaseDate"),
artists=artists,
)
def _v2_track_from_data(
self, data: dict, artists: list[Artist], album: Album | None
) -> Track:
"""The v2 track attributes -> Track fields (the single and
bulk fetches share the object shape)."""
attrs = data.get("attributes") or {}
tags = attrs.get("mediaTags") or []
return Track(
id=data.get("id"),
title=attrs.get("title"),
version=attrs.get("version"),
duration=parse_iso8601_duration(attrs.get("duration")),
explicit=bool(attrs.get("explicit")),
isrc=attrs.get("isrc"),
copyright_text=(
(attrs.get("copyright") or {}).get("text")
if isinstance(attrs.get("copyright"), dict)
else None
),
audio_quality=self._v2_media_tags_to_quality(tags),
media_tags=list(tags),
artists=artists,
album=album,
bpm=_optional_number(attrs.get("bpm")),
key=_optional_text(attrs.get("key")),
key_scale=_optional_text(attrs.get("keyScale")),
)
def _v2_track(self, env: dict) -> Track:
data = self._v2_data(env)
included = self._v2_included(env)
artists = self._v2_artists(included.get("artists"))
albums = included.get("albums") or []
album_resource = albums[0] if albums else None
album = (
self._v2_album_from_resource(album_resource, artists)
if isinstance(album_resource, dict)
else None
)
return self._v2_track_from_data(data, artists, album)
@staticmethod
def _v2_artists(resources: list[dict] | None) -> list[Artist]:
artists = []
for resource in resources or []:
if not isinstance(resource, dict) or resource.get("type") != "artists":
continue
attrs = resource.get("attributes") or {}
artists.append(
Artist(
id=resource.get("id"),
name=attrs.get("name"),
popularity=float(_optional_number(attrs.get("popularity")) or 0.0),
follower_count=int(attrs.get("numberOfFollowers") or 0),
)
)
return artists
def _v2_cover_art_id(self, data: dict) -> str | None:
cover = ((data.get("relationships") or {}).get("coverArt") or {}).get("data")
if isinstance(cover, list):
cover = cover[0] if cover else None
if isinstance(cover, dict) and cover.get("id"):
artwork_id: str | None = str(cover["id"])
return artwork_id
return None
def _resolve_artwork(self, artwork_id: str | None) -> str | None:
"""v2 artwork id -> legacy resources.tidal.com image id (or None)."""
if not artwork_id:
return None
try:
env = self._v2_get(f"artworks/{artwork_id}")
except TidalError as exc:
# Best effort: a 404 artwork (NotFoundError, not
# TidalApiError) or any other failure degrades to no
# cover — it must never break the v2 catalog fallback.
logger.debug("could not resolve artwork %s: %s", artwork_id, exc)
return None
attrs = (env.get("data") or {}).get("attributes") or {}
for item in attrs.get("files") or []:
match = _ARTWORK_IMAGE_ID_RE.search(str(item.get("href") or ""))
if match:
return match.group(1).replace("/", "-")
return None
def _v2_album_from_attrs(
self,
album_id: int | str | None,
attrs: dict,
cover_id: str | None,
artists: list[Artist],
) -> Album:
"""The v2 album attributes -> Album fields (the single and
bulk fetches share the object shape)."""
return Album(
id=album_id,
title=attrs.get("title"),
version=attrs.get("version"),
release_date=attrs.get("releaseDate"),
# The v2 copyright is a dict {"text": ...} (live API
# verified) — the v1 catalog's is a plain string.
copyright_label=(
(attrs.get("copyright") or {}).get("text")
if isinstance(attrs.get("copyright"), dict)
else None
),
number_of_tracks=int(attrs.get("numberOfItems") or 0),
number_of_volumes=int(attrs.get("numberOfVolumes") or 0),
album_type=attrs.get("albumType"),
explicit=bool(attrs.get("explicit")),
audio_quality=self._v2_media_tags_to_quality(attrs.get("mediaTags")),
artwork_id=self._resolve_artwork(cover_id),
artists=artists,
popularity=float(_optional_number(attrs.get("popularity")) or 0.0),
# v2 reports the total playtime as an ISO-8601 duration
# (live API verified: "PT53M36S" for a 54-minute album);
# the UPC is barcodeId (live API verified: the same value
# the v1 catalog's ``upc`` carries); payToStream /
# streamStartDate are v1-only facts.
duration=parse_iso8601_duration(attrs.get("duration")),
upc=_optional_text(attrs.get("barcodeId")),
)
def _v2_album_with_items(
self, album_id: int | str
) -> tuple[Album, list[Track | None]]:
env = self._v2_get(
f"albums/{album_id}", {"include": "items,artists,items.artists"}
)
data = self._v2_data(env)
attrs = data.get("attributes") or {}
included = self._v2_included(env)
artists = self._v2_artists(included.get("artists"))
cover_id = self._v2_cover_art_id(data)
album = self._v2_album_from_attrs(data.get("id"), attrs, cover_id, artists)
return album, self._v2_album_items(env, included)
def _v2_album_items(
self, env: dict, included: dict[str, list[dict]]
) -> list[Track | None]:
data = self._v2_data(env)
items = ((data.get("relationships") or {}).get("items") or {}).get("data") or []
expected = (data.get("attributes") or {}).get("numberOfItems")
self._check_v2_items_complete("album", len(items), expected)
artists_by_id = self._v2_artists_by_id(included)
included_tracks = {
r.get("id"): r for r in included.get("tracks", []) if isinstance(r, dict)
}
album_artist_ids = ((data.get("relationships") or {}).get("artists") or {}).get(
"data"
) or []
album_artists = [
artists_by_id[entry["id"]]
for entry in album_artist_ids
if isinstance(entry, dict) and entry.get("id") in artists_by_id
]
tracks: list[Track | None] = []
for item in items:
if not isinstance(item, dict) or item.get("type") != "tracks":
# A non-audio slot (music video): recorded as ``None``
# so the reports can count it (see _v1_playlist_items).
tracks.append(None)
continue
resource = included_tracks.get(item.get("id"))
if resource is None:
logger.debug(
"v2 album item %s has no resource; skipping", item.get("id")
)
continue
meta = item.get("meta") or {}
track = self._v2_track_from_resource(
resource, artists_by_id, fallback=album_artists
)
track.track_number = int(meta.get("trackNumber") or track.track_number or 0)
track.volume_number = int(
meta.get("volumeNumber") or track.volume_number or 0
)
tracks.append(track)
return tracks
def _v2_track_from_resource(
self,
resource: dict,
artists_by_id: dict[Any, Artist],
fallback: list[Artist] | None = None,
) -> Track:
attrs = resource.get("attributes") or {}
tags = attrs.get("mediaTags") or []
return Track(
id=resource.get("id"),
title=attrs.get("title"),
version=attrs.get("version"),
duration=parse_iso8601_duration(attrs.get("duration")),
explicit=bool(attrs.get("explicit")),
isrc=attrs.get("isrc"),
audio_quality=self._v2_media_tags_to_quality(tags),
media_tags=list(tags),
artists=self._v2_resource_artists(resource, artists_by_id, fallback),
bpm=_optional_number(attrs.get("bpm")),
key=_optional_text(attrs.get("key")),
key_scale=_optional_text(attrs.get("keyScale")),
)
def _v2_playlist_with_items(
self, playlist_id: int | str, is_mix: bool
) -> tuple[Playlist, list[Track | None]]:
env = self._v2_get(
f"playlists/{playlist_id}", {"include": "items,items.artists"}
)
data = self._v2_data(env)
attrs = data.get("attributes") or {}
included = self._v2_included(env)
cover_id = self._v2_cover_art_id(data)
items = self._v2_playlist_items(env, included)
number_of_tracks = int(
attrs.get("numberOfTrackItems") or attrs.get("numberOfTracks") or 0
)
if not number_of_tracks:
# Live API verified: a v2 mix carries NO count attribute
# (both are null) — the parsed items are the truth, so an
# editorial mix no longer reports 0 tracks.
number_of_tracks = sum(1 for item in items if item is not None)
playlist = Playlist(
id=data.get("id"),
title=attrs.get("title") or attrs.get("name"),
description=attrs.get("description"),
number_of_tracks=number_of_tracks,
explicit=bool(attrs.get("explicit")),
artwork_id=self._resolve_artwork(cover_id),
is_mix=is_mix,
creation_time=_optional_text(attrs.get("createdAt")),
last_modified=_optional_text(attrs.get("lastModifiedAt")),
# The v2 duration is an ISO-8601 string (live API
# verified: "PT1H56M35S") — int() raised on it; the
# parser also tolerates a plain seconds value.
duration=parse_iso8601_duration(attrs.get("duration")),
playlist_type=_optional_text(attrs.get("playlistType")),
follower_count=int(attrs.get("numberOfFollowers") or 0),
)
return playlist, items
def _v2_playlist_items(
self, env: dict, included: dict[str, list[dict]]
) -> list[Track | None]:
data = self._v2_data(env)
items = ((data.get("relationships") or {}).get("items") or {}).get("data") or []
expected = (data.get("attributes") or {}).get("numberOfItems")
self._check_v2_items_complete("playlist", len(items), expected)
artists_by_id = self._v2_artists_by_id(included)
included_tracks = {
r.get("id"): r for r in included.get("tracks", []) if isinstance(r, dict)
}
tracks: list[Track | None] = []
for item in items:
if not isinstance(item, dict) or item.get("type") != "tracks":
# A non-audio slot (music video): recorded as ``None``
# so the reports can count it (see _v1_playlist_items).
tracks.append(None)
continue
resource = included_tracks.get(item.get("id"))
if resource is None:
logger.debug(
"v2 playlist item %s has no resource; skipping", item.get("id")
)
continue
tracks.append(self._v2_track_from_resource(resource, artists_by_id))
return tracks
@staticmethod
def _check_v2_items_complete(kind: str, item_count: int, expected: Any) -> None:
"""Refuse a truncated collection (v2's hard 20-item cap)."""
try:
expected_count = int(expected)
except (TypeError, ValueError):
return
if (
expected_count > 0
and item_count >= _V2_ITEMS_PAGE_CAP
and item_count < expected_count
):
raise TidalApiError(
f"OpenAPI v2 only exposes the first {item_count} of {expected_count} "
f"items of this {kind} (API limitation, no working pagination); "
"cannot enumerate it fully",
200,
)
# ------------------------------------------------------------------
# resolve (URL -> Resource)
# ------------------------------------------------------------------
[docs]
def resolve(self, url: str) -> Resource:
"""Resolve any Tidal link (or bare id) to a catalog resource."""
kind, item_id = parse_tidal_url(url)
if not item_id:
raise NotFoundError("empty Tidal URL")
if kind == "video":
raise UnsupportedType(
"video links are not supported by this library "
"(audio-only: track/album/playlist/mix/artist)"
)
if kind is None and _is_foreign_url(item_id):
# A paste of a non-Tidal link must fail fast with a clear
# message, not burn the whole kind-probe ladder on a
# string no catalog will ever contain (reference-fork
# parity: parse before network).
raise UnsupportedType(f"not a Tidal link: {item_id}")
if kind is not None:
return self._resolve_known_kind(kind, item_id)
# Bare id: try the likely kinds in order. Only a clean 404 means
# "try the next kind"; any other error (region lock, rate limit,
# bad token) is propagated immediately. Artists are tried last:
# numeric ids collide across kinds and tracks win.
last_error: Exception | None = None
for try_kind in ("track", "album", "playlist", "mix", "artist"):
try:
return self._resolve_known_kind(try_kind, item_id)
except NotFoundError as exc:
last_error = exc
continue
raise last_error or NotFoundError(f"no Tidal resource found for {item_id!r}")
def _resolve_known_kind(self, kind: str, item_id: str) -> Resource:
if kind == "track":
return Resource("track", self.get_track(item_id))
if kind == "album":
album, items = self.get_album(item_id)
self._items_cache[self._items_key("album", item_id)] = items
return Resource("album", album)
if kind == "playlist":
playlist, items = self.get_playlist(item_id, is_mix=False)
self._items_cache[self._items_key("playlist", item_id)] = items
return Resource("playlist", playlist)
if kind == "mix":
playlist, items = self.get_playlist(item_id, is_mix=True)
self._items_cache[self._items_key("mix", item_id)] = items
return Resource("mix", playlist)
if kind == "artist":
return Resource("artist", self.get_artist(item_id))
raise UnsupportedType(f"unknown Tidal kind: {kind!r}")
# ------------------------------------------------------------------
# search (v1)
# ------------------------------------------------------------------
[docs]
def search(
self,
query: str,
types: tuple[str, ...] = ("TRACK",),
limit: int = 20,
offset: int = 0,
country: str | None = None,
) -> tuple[list[dict], int, dict | None]:
"""Catalog search; returns ``(items, total, top_hit)``.
``top_hit`` is the catalog's single best match for the query
(the ``topHit`` field, live API verified: a full v1 object of
its named group) built as the same uniform search row — or
``None`` when the catalog doesn't name one. The reference
clients ignore the field; a real-time search UI renders it as
the "best match" banner above the results.
Each item is the reference client's search row shape
(``{"kind", "id", "title"}`` plus ``artist`` where sensible) —
the same rows its ``search`` RPC/CLI emit, which the host
contract reads: lowercase ``kind`` (``https://tidal.com/
<kind>/<id>`` links build from it), playlist ids are the
``uuid``, artist titles normalise from the ``name`` field.
Rows additionally carry the catalog's garnish *when present*
(for result-rendering clients): ``version`` (track/album
editions), ``duration`` in seconds (track/video length, album
total), ``quality`` (track/album, the raw API value — map it
with :func:`tidalpy.quality.quality_label` for display).
``total`` is the sum of the catalog's per-group
``totalNumberOfItems`` (may exceed ``len(items)``).
Live API notes (probed against Tidal, 2026):
- ``types`` must be the *plural* names (``TRACKS``, ``ALBUMS``,
…); singular names are rejected;
- ``countryCode`` is mandatory (400/subStatus 1002 without it)
and is taken from the logged-in session;
- the response is *grouped* (``albums.items``, ``tracks.items``,
…), not a flat ``items`` array — items are flattened here.
"""
if not query or not query.strip():
return [], 0, None
params: dict[str, Any] = {
"query": query.strip(),
"types": ",".join(_search_type_name(t) for t in types),
"limit": min(max(1, limit), 50),
"offset": max(0, offset),
}
# The catalog is per-market: the account's country by default,
# a per-call override when the host searches a different
# market (a search UI for an account in another region).
country = country or self._oauth.country_code()
if country:
params["countryCode"] = country
data = self._transport.get_json("search", params)
if not isinstance(data, dict):
return [], 0, None
# Group order mirrors the reference client's all-types request
# string (ARTISTS,ALBUMS,TRACKS,VIDEOS,PLAYLISTS) so a default
# search returns results in the same order.
items: list[dict] = []
total = 0
for group_key, kind in (
("artists", "artist"),
("albums", "album"),
("tracks", "track"),
("videos", "video"),
("playlists", "playlist"),
):
group = data.get(group_key)
if not isinstance(group, dict):
continue
group_total = group.get("totalNumberOfItems")
if isinstance(group_total, int):
total += group_total
for item in group.get("items") or []:
if not isinstance(item, dict):
continue
items.append(_search_row(kind, item))
# The catalog's single best match for the query (live API
# verified: ``{"type": "ARTISTS", "value": <full v1 object>}``
# — the reference clients ignore it). It wears the same row
# shape as a search row (the value is a full v1 payload of the
# named group), so a real-time search UI renders it exactly
# like a result — the "best match" banner above the list.
top_row: dict[str, Any] | None = None
top_hit = data.get("topHit")
if isinstance(top_hit, dict):
top_kind = _top_hit_kind(top_hit.get("type"))
value = top_hit.get("value")
if top_kind is not None and isinstance(value, dict):
top_row = _search_row(top_kind, value)
return items, total, top_row
[docs]
def autocomplete(
self, query: str, limit: int = 10, country: str | None = None
) -> list[dict[str, Any]]:
"""Catalog autocomplete (the web app's search-box suggest).
The keystroke endpoint for real-time search: ``search/
autocomplete`` answers as the user types, across artists,
albums, tracks, and playlists, with a small result budget.
Each row is the *uniform search row* (``kind``/``id``/
``title`` plus the catalog garnish when present), so a
suggest box and a search result render with the same row
code.
Live API notes (probed against Tidal, 2026): the response
rows carry a ``type`` discriminator (``"ARTIST"``, ``"ALBUM"``,
``"TRACK"``, ``"PLAYLIST"``) plus the full v1 object under
``item``; ``countryCode`` is optional (an account in an
unknown market gets an *empty* result, not an error), so the
session's country is passed by default, like :meth:`search`.
Raises :class:`TidalError` on transport failures; a query
with no matches returns ``[]`` (not an error).
"""
if not query or not query.strip():
return []
params: dict[str, Any] = {
"query": query.strip(),
"limit": min(max(1, limit), 50),
}
country = country or self._oauth.country_code()
if country:
params["countryCode"] = country
data = self._transport.get_json("search/autocomplete", params)
rows: list[dict[str, Any]] = []
items = data.get("items") if isinstance(data, dict) else None
for row in items if isinstance(items, list) else []:
if not isinstance(row, dict):
continue
item = row.get("item")
if not isinstance(item, dict):
continue
kind = _AUTOCOMPLETE_KINDS.get(str(row.get("type")).upper())
if kind is None:
continue
rows.append(_search_row(kind, item))
return rows
[docs]
def get_user_playlists(
self,
user_id: int | str | None = None,
limit: int = 25,
offset: int = 0,
) -> tuple[list[dict[str, Any]], int]:
"""A user's playlist links (``(rows, total)``) — the "My
Playlists" tab (reference parity: TMD's ``getUserPlaylists``).
``user_id`` defaults to the logged-in session's user (the
account's own playlists); pass another user id for that
user's *public* playlists. Each row is the uniform search
row for a playlist (``kind``/``id``/``title`` — the id is the
linkable ``uuid``), so a library UI renders it with the same
row code as search/browse and builds
``https://tidal.com/playlist/<uuid>`` links straight away.
Live API verified (2026): the endpoint is the plural
``users/{id}/playlists`` (the singular ``user/...`` family is
gone — 404); the response is the standard items envelope.
"""
if user_id is None:
record = self._oauth.record
if record is None or not record.user_id:
raise ConfigError(
"no user id: pass one explicitly, or the session "
"must be logged in"
)
user_id = record.user_id
data = self._transport.get_json(
f"users/{user_id}/playlists",
{"limit": min(max(1, limit), 50), "offset": max(0, offset)},
)
if not isinstance(data, dict):
return [], 0
rows = [
_search_row("playlist", item)
for item in (data.get("items") or [])
if isinstance(item, dict)
]
total = data.get("totalNumberOfItems")
return rows, (total if isinstance(total, int) else len(rows))
# ------------------------------------------------------------------
# genre browse (the catalog's "Discover" tab)
# ------------------------------------------------------------------
[docs]
def get_genres(self) -> list[dict[str, Any]]:
"""The catalog's browse genres (live API verified: 20 fixed
genres — the web app's "Discover" grid).
Each row is ``{"name", "path", "has_playlists",
"has_artists", "has_albums", "has_tracks", "has_videos"}``
plus ``artwork`` (the card thumbnail id, consumable by
:func:`~tidalpy.client.artwork_url`) when the catalog names
one. ``path`` is the catalog's own URL segment for the genre
(equal to ``name`` on every probed genre) — pass it back to
:meth:`get_genre_items`. The ``has_*`` flags tell a browse UI
which tabs a genre card offers (no genre currently has
artists; one has no playlists).
"""
data = self._transport.get_json("genres")
entries: list[Any] = data if isinstance(data, list) else []
genres: list[dict[str, Any]] = []
for item in entries:
name = item.get("name") if isinstance(item, dict) else None
if not (isinstance(name, str) and name):
continue
genre: dict[str, Any] = {
"name": name,
"path": item.get("path") or name,
"has_playlists": item.get("hasPlaylists") is True,
"has_artists": item.get("hasArtists") is True,
"has_albums": item.get("hasAlbums") is True,
"has_tracks": item.get("hasTracks") is True,
"has_videos": item.get("hasVideos") is True,
}
image = item.get("image")
if isinstance(image, str) and image:
genre["artwork"] = image
genres.append(genre)
return genres
[docs]
def get_genre_items(
self,
genre: str,
item_type: str = "tracks",
limit: int = 25,
offset: int = 0,
) -> tuple[list[dict[str, Any]], int]:
"""Browse one genre's catalog items (``(rows, total)``).
``genre`` is the catalog's genre ``path`` (from
:meth:`get_genres`); ``item_type`` one of ``"tracks"``,
``"albums"``, ``"playlists"``, ``"videos"`` — the plural
endpoint segment, which also matches the genre row's
``has_<item_type>`` flag (browse what the card offers).
The items are full v1 objects and come back as the *uniform
search rows* (:func:`_search_row`): a browse tab and a search
result render with the same row code, and the rows'
``id``/``kind`` build links straight away.
Raises :class:`ConfigError` for an unknown ``item_type``,
:class:`NotFoundError` for an unknown genre.
"""
if item_type not in GENRE_ITEM_TYPES:
raise ConfigError(
f"unknown genre item type: {item_type!r} "
f"(expected one of {', '.join(repr(t) for t in GENRE_ITEM_TYPES)})"
)
# The genre's own URL segment, quoted (names contain spaces
# and slashes: "Hip Hop / Rap").
segment = quote(genre.strip(), safe="")
if not segment:
raise ConfigError("genre path is empty")
data = self._transport.get_json(
f"genres/{segment}/{item_type}",
{"limit": min(max(1, limit), 50), "offset": max(0, offset)},
)
if not isinstance(data, dict):
return [], 0
# The uniform search row: playlist rows carry their link id
# in ``uuid`` (the branch handles it), the catalog garnish
# rides along when present.
kind = item_type[:-1] # "tracks" -> "track" (the row's kind)
rows = [
_search_row(kind, item)
for item in (data.get("items") or [])
if isinstance(item, dict)
]
total = data.get("totalNumberOfItems")
return rows, (total if isinstance(total, int) else len(rows))
# ------------------------------------------------------------------
# playback / stream resolution
# ------------------------------------------------------------------
[docs]
def resolve_stream(
self,
track_id: int | str,
profile: Quality,
track: Track | None = None,
allow_preview: bool = False,
) -> StreamInfo:
"""Resolve a playable stream for ``track_id`` at ``profile``.
Strategy (order matters, it encodes Tidal's real behaviour):
1. lossless profiles try the OpenAPI v2 FLAC manifest first
(full fidelity, no legacy caps);
2. then the v1 playback ladder (per-track attempts);
3. if nothing meets the profile but *something* playable was
resolved, the best available stream is returned and the host
reports the downgrade — silent quality loss, loud reporting.
Successful resolutions are cached briefly per
(account, track, profile, allow_preview) — a duplicate track in
a playlist, or a host that previews the same track twice, must
not spend a second rate-limited playback request. Failures are
never cached (a transient miss may clear).
``allow_preview=True`` accepts the ~30-second preview the
catalog serves instead of the full track (free-tier accounts,
pre-release/exclusive tracks): the returned
:class:`~tidalpy.models.StreamInfo` has ``preview=True`` set,
and a preview answer short-circuits the ladder (every other
rung would return the same preview). Full streams are still
preferred: a track that has one gets it, preview or not.
Raises :class:`PreviewOnlyError` (when ``allow_preview`` is
false) / :class:`QualityUnavailableError` when nothing
acceptable can be delivered.
"""
tags = stereo_media_tags(track.media_tags if track else None)
catalog_quality = track.audio_quality if track else None
key = (
self._account_identity(),
track_id,
profile.value,
tuple(tags),
catalog_quality,
allow_preview,
)
cached = self._stream_cache_get(key)
if cached is not None:
return cached
stream = self._resolve_stream_fresh(
track_id, profile, tags, catalog_quality, allow_preview
)
self._stream_cache_put(key, stream)
return stream
def _account_identity(self) -> int | str | None:
"""The logged-in account the cache is keyed on (None when
logged out) — adopting a different login must not serve the
previous account's streams."""
record = self._oauth.record
return record.user_id if record is not None else None
def _stream_cache_get(self, key: tuple) -> StreamInfo | None:
entry = self._stream_cache.get(key)
if entry is None:
return None
created, stream = entry
if time.monotonic() - created > STREAM_CACHE_TTL_SECONDS:
del self._stream_cache[key]
return None
# Refresh recency (insertion-ordered LRU).
del self._stream_cache[key]
self._stream_cache[key] = entry
return _copy_stream(stream)
def _stream_cache_put(self, key: tuple, stream: StreamInfo) -> None:
self._stream_cache[key] = (time.monotonic(), _copy_stream(stream))
while len(self._stream_cache) > STREAM_CACHE_MAX_ITEMS:
oldest = next(iter(self._stream_cache))
del self._stream_cache[oldest]
def _playback_param_blocked(self, audioquality: str) -> TidalApiError | None:
"""The prior ``CLIENT_NOT_ENTITLED`` rejection for this
playback quality, when still within its TTL (skip the probe);
``None`` when the probe should run."""
key = (self._account_identity(), audioquality)
entry = self._blocked_params.get(key)
if entry is None:
return None
deadline, error = entry
if time.monotonic() >= deadline:
del self._blocked_params[key]
return None
return error
def _mark_playback_param_blocked(
self, audioquality: str, error: TidalApiError
) -> None:
"""Remember an account-capability rejection (TTL-bounded,
per account — a re-login may carry a different entitlement)."""
key = (self._account_identity(), audioquality)
self._blocked_params[key] = (
time.monotonic() + PLAYBACK_BLOCK_TTL_SECONDS,
error,
)
logger.debug(
"playback quality %s blocked for %.0fs (account not entitled)",
audioquality,
PLAYBACK_BLOCK_TTL_SECONDS,
)
def _resolve_stream_fresh(
self,
track_id: int | str,
profile: Quality,
tags: list[str],
catalog_quality: str | None,
allow_preview: bool = False,
) -> StreamInfo:
"""The uncached resolution ladder (see :meth:`resolve_stream`)."""
attempts = playback_attempts(profile, tags, catalog_quality)
last_stream: StreamInfo | None = None
last_error: Exception | None = None
# Audio qualities that already answered this resolution. A
# quality that answered (non-conformingly, else we would have
# returned) must not be requested again: a lossy-only track
# answers every lossless quality with the same rendition, so
# re-asking only burns rate-limited requests (reference-fork
# issue #32: a quality-mismatched answer is terminal for that
# quality).
tried: set[str] = set()
if profile in (Quality.LOSSLESS, Quality.HI_RES_LOSSLESS):
try:
flac = self._openapi_flac_stream(track_id, profile, allow_preview)
last_stream = flac
if flac.preview:
# A preview-only asset: no other rung can do
# better (the ladder would just burn requests
# for the same 30 seconds).
return flac
if quality_meets_profile(profile, flac.delivered_quality, flac.codec):
return flac
logger.debug(
"OpenAPI FLAC delivered %s/%s; falling back to playback loop",
flac.delivered_quality,
flac.codec,
)
except Exception as exc: # noqa: BLE001 - ladder continues
last_error = exc
logger.debug("OpenAPI FLAC failed: %s", exc)
for api_quality in attempts:
for variant in self._playback_param_variants(api_quality):
blocked_error = self._playback_param_blocked(variant["audioquality"])
if blocked_error is not None:
# The account is not entitled to this playback
# quality (403 CLIENT_NOT_ENTITLED, cached for
# the TTL window): re-probing would burn a
# rate-limited request for the same 403. The
# stored rejection keeps the eventual error
# actionable if every rung is blocked.
last_error = blocked_error
logger.debug(
"playback quality %s blocked this session; "
"skipping the probe",
variant["audioquality"],
)
continue
try:
stream = self._playback_stream(track_id, variant, allow_preview)
except Exception as exc: # noqa: BLE001 - ladder continues
last_error = exc
logger.debug("playback variant %s failed: %s", variant, exc)
continue
tried.add(variant["audioquality"])
last_stream = stream
if stream.preview:
# A preview-only asset: every other rung answers
# with the same preview, so stop here instead of
# burning rate-limited requests.
return stream
if quality_meets_profile(
profile, stream.delivered_quality, stream.codec
):
return stream
# A lossless-family request answered with a
# spatial/proprietary/lossy rendition: retry at
# LOSSLESS — unless LOSSLESS already answered this
# resolution, in which case the retry would repeat a
# request whose answer we already have.
if (
needs_stereo_fallback(variant["audioquality"], stream.codec)
and "LOSSLESS" not in tried
):
logger.debug(
"unexpected codec %s for %s; retrying LOSSLESS",
stream.codec,
variant["audioquality"],
)
lossless_variant = dict(variant)
lossless_variant["audioquality"] = "LOSSLESS"
try:
stream = self._playback_stream(track_id, lossless_variant)
tried.add("LOSSLESS")
last_stream = stream
if quality_meets_profile(
profile, stream.delivered_quality, stream.codec
):
return stream
except Exception as exc: # noqa: BLE001 - ladder continues
last_error = exc
logger.debug("stereo fallback failed: %s", exc)
if last_stream is not None:
logger.warning(
"requested %s not deliverable; using best available %s (%s)",
profile.value,
last_stream.delivered_quality,
last_stream.codec,
)
return last_stream
if last_error is not None:
raise last_error
raise QualityUnavailableError(
f"no stream could be resolved for track {track_id} at {profile.value}"
)
@staticmethod
def _playback_param_variants(api_quality: str) -> list[dict]:
# ``audiomode=STEREO`` is sent up front, as the reference client
# does: spatial/proprietary renditions are useless for the
# transfer use case, and requesting stereo avoids the reactive
# stereo-fallback round trip for the tracks that would
# otherwise be answered with one.
variants = [
{
"audioquality": api_quality,
"playbackmode": "STREAM",
"assetpresentation": "FULL",
"audiomode": "STEREO",
}
]
if api_quality in ("LOSSLESS", "HI_RES"):
variants.append(
{
"audioquality": "HI_RES_LOSSLESS",
"playbackmode": "STREAM",
"assetpresentation": "FULL",
"audiomode": "STEREO",
}
)
return variants
def _playback_stream(
self, track_id: int | str, params: dict, allow_preview: bool = False
) -> StreamInfo:
"""One playbackinfo attempt over the known hosts/versions."""
last_error: Exception | None = None
bases = (
(self._transport.api_base, "primary"),
(self._transport.legacy_api_base, "legacy"),
)
for base, base_label in bases:
for path in (
f"tracks/{track_id}/playbackinfopostpaywall/v4",
f"tracks/{track_id}/playbackinfopostpaywall",
):
try:
data = self._transport.get_json(path, params, base=base)
except TidalApiError as exc:
last_error = exc
# Auth / rate-limit / stale-client cannot be repaired by
# switching host/version — stop instead of burning calls.
if exc.status_code in (401, 429) or exc.sub_status == 4022:
raise
if "CLIENT_NOT_ENTITLED" in exc.codes:
# An account-capability rejection, not a
# track quirk: every other track gets the same
# 403, so cache the block and stop probing
# this quality for the TTL window.
self._mark_playback_param_blocked(
str(params.get("audioquality")), exc
)
raise
logger.debug(
"playback failed (%s, %s): %s",
path.rsplit("/", 1)[-1],
base_label,
exc,
)
continue
return self._stream_from_playback_data(
track_id, data, params["audioquality"], allow_preview
)
raise last_error or StreamError("playback info unavailable")
def _stream_from_playback_data(
self,
track_id: int | str,
data: dict,
requested: str,
allow_preview: bool = False,
) -> StreamInfo:
presentation = str(data.get("assetPresentation") or "FULL").strip().upper()
is_preview = presentation == "PREVIEW"
if is_preview and not allow_preview:
raise PreviewOnlyError(
"Tidal returned a preview-only stream instead of the full track "
"(the account may be on the free tier, or the track is "
"pre-release/exclusive; retry with allow_preview=True to "
"accept the ~30-second preview)"
)
mime = data.get("manifestMimeType") or ""
manifest = data.get("manifest") or ""
delivered = data.get("audioQuality") or ""
if "vnd.tidal.bt" in mime:
parsed = mf.parse_bento(manifest)
stream = StreamInfo(
track_id=track_id,
urls=parsed["urls"],
codec=normalize_codec(parsed["codec"]),
delivered_quality=delivered,
requested_quality=requested,
manifest_mime=mime,
encryption_key=parsed["encryption_key"],
bandwidth=parsed["bandwidth"],
sample_rate=parsed["sample_rate"],
bit_depth=parsed["bit_depth"],
channels=parsed["channels"],
preview=is_preview,
)
elif "dash+xml" in mime:
xml_text = mf.decode_manifest(manifest, mime)
reps = mf.parse_dash(xml_text)
selected = select_representation(reps, requested)
stream = StreamInfo(
track_id=track_id,
urls=selected["urls"],
codec=normalize_codec(selected["codec"]),
delivered_quality=delivered,
requested_quality=requested,
manifest_mime=mime,
bandwidth=selected["bandwidth"],
sample_rate=selected["sample_rate"],
bit_depth=selected["bit_depth"],
channels=selected["channels"],
representation_id=selected.get("representation_id"),
preview=is_preview,
)
else:
raise ManifestError(f"unsupported manifest type: {mime or 'unknown'}")
return stream
# ------------------------------------------------------------------
# OpenAPI v2 FLAC manifests
# ------------------------------------------------------------------
def _openapi_flac_stream(
self, track_id: int | str, profile: Quality, allow_preview: bool = False
) -> StreamInfo:
formats = (
["FLAC_HIRES", "FLAC"] if profile is Quality.HI_RES_LOSSLESS else ["FLAC"]
)
attrs = self._get_openapi_track_manifest(track_id, formats, allow_preview)
# The manifest's own presentation: a preview-only asset is
# flagged so the caller short-circuits (and the host knows
# the 30 seconds are all there is).
presentation = str(attrs.get("trackPresentation") or "FULL").strip().upper()
is_preview = presentation == "PREVIEW"
available = set(attrs.get("formats") or [])
if "FLAC_HIRES" in available:
sound_quality = "HI_RES_LOSSLESS"
elif "FLAC" in available:
sound_quality = "LOSSLESS"
else:
raise QualityUnavailableError(
"lossless FLAC is not available for this track"
)
xml_text = mf.decode_data_uri(attrs.get("uri") or "")
reps = mf.parse_dash(xml_text)
selected = select_representation(reps, sound_quality)
if not selected["urls"]:
raise ManifestError("no segment URLs in selected MPD representation")
return StreamInfo(
track_id=track_id,
urls=selected["urls"],
codec=normalize_codec(selected["codec"]),
delivered_quality=sound_quality,
requested_quality=sound_quality,
manifest_mime="application/dash+xml",
bandwidth=selected["bandwidth"],
sample_rate=selected["sample_rate"],
bit_depth=selected["bit_depth"],
channels=selected["channels"],
representation_id=selected.get("representation_id"),
manifest_hash=attrs.get("hash"),
replaygain=self._loudness_facts(attrs),
preview=is_preview,
)
@staticmethod
def _loudness_facts(attrs: dict) -> dict[str, str]:
facts: dict[str, str] = {}
for scope, prefix in (
("track", "trackReplayGain"),
("album", "albumReplayGain"),
):
data = attrs.get(f"{scope}AudioNormalizationData")
if isinstance(data, dict):
gain = data.get("replayGain")
peak = data.get("peakAmplitude")
if gain is not None:
facts[prefix] = f"{float(gain):+.2f} dB"
if peak is not None:
facts[f"{prefix.replace('ReplayGain', 'PeakAmplitude')}"] = str(
peak
)
return facts
def _get_openapi_track_manifest(
self, track_id: int | str, formats: list[str], allow_preview: bool = False
) -> dict:
"""trackManifests with usage/formats fallback.
403 (e.g. ``PREREQUISITE_MISSING``) and 404 fall through to the
next usage/format combination; 401/429 propagate immediately (a
dead token or an exhausted rate limit cannot be repaired by
trying another combination).
A ``PREVIEW`` presentation (a full-fidelity manifest holding
only the 30 seconds) is never returned as a full stream: by
default it records a :class:`PreviewOnlyError` and the next
usage is tried; with ``allow_preview`` the last-seen preview
attrs are returned (a FULL manifest still wins over them).
"""
format_attempts = [formats]
if len(formats) > 1:
# 403 PREREQUISITE_MISSING can be triggered by FLAC_HIRES alone;
# retry with the base format before giving up.
format_attempts.append(formats[-1:])
last_error: Exception | None = None
preview_attrs: dict | None = None
for usage in OPENAPI_MANIFEST_USAGES:
for attempt_formats in format_attempts:
try:
attrs = self._openapi_track_manifest_once(
track_id, attempt_formats, usage
)
except TidalError as exc:
if getattr(exc, "status_code", None) in (401, 429):
raise
last_error = exc
logger.debug(
"trackManifests usage=%s formats=%s failed: %s",
usage,
attempt_formats,
exc,
)
continue
presentation = (
str(attrs.get("trackPresentation") or "FULL").strip().upper()
)
if presentation == "PREVIEW":
# A full-fidelity manifest can still contain only the
# 30-second preview; never report that as completed.
last_error = PreviewOnlyError(
"OpenAPI manifest is preview-only; trying next usage"
)
if allow_preview and preview_attrs is None:
preview_attrs = attrs
logger.debug("trackManifests usage=%s is PREVIEW", usage)
break
return attrs
if allow_preview and preview_attrs is not None:
return preview_attrs
if last_error is not None:
raise last_error
raise StreamError("track manifest request returned no attributes")
def _openapi_track_manifest_once(
self, track_id: int | str, formats: list[str], usage: str
) -> dict:
# A list of tuples so repeated ``formats`` keys survive (the
# OpenAPI endpoint accepts ?formats=A&formats=B).
params: list[tuple[str, str]] = [
("manifestType", "MPEG_DASH"),
("uriScheme", "DATA"),
("usage", usage),
("adaptive", "false"),
]
for item in formats:
params.append(("formats", item))
body = self._transport.get_json(
f"trackManifests/{track_id}",
params,
base=self._transport.openapi_base,
accept="application/vnd.api+json",
)
data = body.get("data") if isinstance(body, dict) else None
attrs = data.get("attributes") if isinstance(data, dict) else None
if not isinstance(attrs, dict):
raise StreamError("track manifest response has no attributes")
return attrs