"""Typed data models for tidalpy.
Plain dataclasses with ``to_dict()`` for JSON serialisation. The
catalog-facing dicts intentionally mirror the *camelCase* key names of
the Tidal v1 JSON API (``title``, ``numberOfTracks``, ``releaseDate``,
...) because downstream consumers — in particular ``tidal_transfer``'s
``TidalTask.info`` JSONField and its admin UI — were written against
that shape. :meth:`Resource.to_dict` emits ``{"type": ..., "data": ...}``
exactly like the legacy CLI's ``--json`` mode.
"""
from __future__ import annotations
import re
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any
__all__ = [
"Artist",
"Album",
"Track",
"Playlist",
"Resource",
"SessionInfo",
"StreamInfo",
"ItemReport",
"DownloadReport",
]
def _resource_link(kind: str, identifier: int | str | None) -> str | None:
"""``https://tidal.com/<kind>/<id>`` — ``None`` without an id.
The id form (never the artist handle or playlist name): it is
the form :func:`tidalpy.api.parse_tidal_url` round-trips, so a
link a host builds this way always resolves through the library
again.
"""
if identifier is None or identifier == "":
return None
return f"https://tidal.com/{kind}/{identifier}"
def _clean(value: Any) -> Any:
"""Make a value JSON-safe (Paths -> str, dataclasses -> dict).
Untyped on purpose (it recurses through arbitrary JSON shapes);
callers that need a concrete return type annotate the assignment.
"""
if isinstance(value, Path):
return str(value)
if hasattr(value, "to_dict"):
return value.to_dict()
if isinstance(value, (list, tuple)):
return [_clean(v) for v in value]
if isinstance(value, dict):
return {str(k): _clean(v) for k, v in value.items()}
return value
[docs]
@dataclass
class Artist:
id: int | str | None = None
name: str | None = None
picture: str | None = None
# Catalog garnish (live API verified): the catalog's popularity
# (v1 reports an int 0..100, v2 a 0..1 float — normalised to
# float), the artist handle when the catalog has one (the linkable
# name in ``tidal.com/artist/<handle>``), and the follower count
# (v2 only).
popularity: float = 0.0
handle: str | None = None
follower_count: int = 0
# The catalog's editorial mixes for this artist (live API
# verified, v1 only — v2 artists carry no mixes): a map of mix
# kind to mix id ({"ARTIST_MIX": "<id>"}). The ids build
# https://play.tidal.com/mc/<id> links, which resolve through the
# normal mix path (a "listen in a mix" tab for artist pages).
mixes: dict[str, str] = field(default_factory=dict)
@property
def link(self) -> str | None:
"""``https://tidal.com/artist/<id>`` (``None`` without an id).
The id form, not the handle form: :func:`tidalpy.api.parse_tidal_url`
validates artist ids as numeric, so this is the form that
round-trips through the library.
"""
return _resource_link("artist", self.id)
[docs]
def to_dict(self) -> dict[str, Any]:
data: dict[str, Any] = _clean(asdict(self))
return data
[docs]
@dataclass
class Album:
id: int | str | None = None
title: str | None = None
version: str | None = None
release_date: str | None = None
copyright_label: str | None = None
number_of_tracks: int = 0
number_of_volumes: int = 0
album_type: str | None = None
explicit: bool = False
audio_quality: str | None = None
artwork_id: str | None = None
artists: list[Artist] = field(default_factory=list)
# Catalog garnish (live API verified): the catalog's popularity
# (v1 reports an int 0..100, v2 a 0..1 float — normalised to
# float, as on Artist), the UPC barcode (v1 only — an indexer's
# stable physical-release id), and the vibrant color (v1 only —
# a card background for rendering clients).
popularity: float = 0.0
upc: str | None = None
vibrant_color: str | None = None
# Album-card facts (live API verified on the v1 object and the
# v1/v2 search rows): the catalog's total playtime in seconds
# (an album card's "12 tracks, 54 min" without an item walk; v2
# reports it as an ISO-8601 duration, parsed to seconds), the
# purchase-required flag (an indexer badges the row *before*
# attempting a resolve, as on Track), and the stream start date
# (the pre-release fact).
duration: int = 0
pay_to_stream: bool = False
stream_start_date: str | None = None
@property
def primary_artist(self) -> Artist | None:
return self.artists[0] if self.artists else None
@property
def link(self) -> str | None:
"""``https://tidal.com/album/<id>`` (``None`` without an id)."""
return _resource_link("album", self.id)
@property
def year(self) -> int | None:
"""The release year (``release_date``'s leading four digits) —
the card's year label, ``None`` when the catalog has none."""
if not self.release_date:
return None
match = re.match(r"(\d{4})", str(self.release_date))
return int(match.group(1)) if match else None
[docs]
def to_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"title": self.title,
"version": self.version,
"releaseDate": self.release_date,
"copyrightLabel": self.copyright_label,
"numberOfTracks": self.number_of_tracks,
"numberOfVolumes": self.number_of_volumes,
"type": self.album_type,
"explicit": self.explicit,
"audioQuality": self.audio_quality,
"artworkId": self.artwork_id,
"primaryArtist": _clean(self.primary_artist),
"artists": _clean(self.artists),
"popularity": self.popularity,
"upc": self.upc,
"vibrantColor": self.vibrant_color,
"duration": self.duration,
"payToStream": self.pay_to_stream,
"streamStartDate": self.stream_start_date,
}
[docs]
@dataclass
class Track:
id: int | str | None = None
title: str | None = None
version: str | None = None
duration: int = 0
track_number: int = 0
volume_number: int = 0
position: int = 0
explicit: bool = False
isrc: str | None = None
copyright_text: str | None = None
audio_quality: str | None = None
media_tags: list[str] = field(default_factory=list)
artists: list[Artist] = field(default_factory=list)
album: Album | None = None
# ReplayGain facts (dB strings, the tag writer's key set:
# trackReplayGain/trackPeakAmplitude/albumReplayGain/
# albumPeakAmplitude). From the v1 catalog's track loudness
# fields (track scope; every items entry carries them), merged
# under the v2 normalization data the stream resolution provides
# (the stream facts win).
replaygain: dict[str, str] = field(default_factory=dict)
# Musical key facts (DJ/CDJ metadata; live API verified on the v1
# and v2 track objects — not on search rows): ``bpm`` as reported
# (int on v1, float on v2), ``key`` the pitch class ("C", "FSharp"),
# ``key_scale`` "MAJOR" / "MINOR".
bpm: int | float | None = None
key: str | None = None
key_scale: str | None = None
# Availability facts (live API verified, v1 — every track entry
# and search row carries them; v2 carries none of the three):
# ``pay_to_stream`` marks a track that must be bought before it
# can be streamed/downloaded (an indexer badges the row *before*
# attempting a resolve instead of learning it from the error),
# ``audio_modes`` the spatial-audio modes (``("STEREO",)`` is the
# norm; 5.1/ATMOS renditions get their own badge), and
# ``stream_start_date`` when streaming started (the pre-release
# fact).
pay_to_stream: bool = False
audio_modes: tuple[str, ...] = ()
stream_start_date: str | None = None
# The catalog's editorial mixes containing this track (live API
# verified, v1 only): a map of mix kind to mix id ({"TRACK_MIX":
# "<id>"}). The ids build https://play.tidal.com/mc/<id> links,
# which resolve through the normal mix path (a "listen in a mix"
# action for track pages).
mixes: dict[str, str] = field(default_factory=dict)
@property
def primary_artist(self) -> Artist | None:
return self.artists[0] if self.artists else None
@property
def link(self) -> str | None:
"""``https://tidal.com/track/<id>`` (``None`` without an id)."""
return _resource_link("track", self.id)
@property
def artist_name(self) -> str:
primary = self.primary_artist
if primary and primary.name:
return primary.name
names = [a.name for a in self.artists if a.name]
return ", ".join(names)
@property
def key_label(self) -> str:
"""A short pitch label (``"F#m"``) for players/DJ tools: the
pitch class in display form (``Sharp`` → ``#``, ``Flat`` →
``b``) with the ``"m"`` scale suffix when minor; ``""`` when
the catalog has no key facts (search rows)."""
if not self.key:
return ""
label = str(self.key).replace("Sharp", "#").replace("Flat", "b")
if (self.key_scale or "").upper() == "MINOR":
label += "m"
return label
@property
def number_label(self) -> str:
"""Zero-padded track number (TMD parity: the volume is never part
of the filename — multi-disc albums use ``CD{n}`` subfolders,
see :func:`tidalpy.naming.cd_folder`).
The reference client numbers a track by its position in the
playlist when the track is playlist-bound, and by its album
track number otherwise — hence ``position`` takes precedence
here (it is only set for playlist items).
"""
number = self.position or self.track_number or 0
return f"{number:02d}"
@property
def duration_label(self) -> str:
"""The track length in ``M:SS`` display form (``"3:38"``) — the
duration column of a player/search UI. ``""`` when the catalog
has no duration (search rows); the rare hour-plus track formats
as ``H:MM:SS``.
"""
if not self.duration:
return ""
minutes, seconds = divmod(int(self.duration), 60)
hours, minutes = divmod(minutes, 60)
if hours:
return f"{hours}:{minutes:02d}:{seconds:02d}"
return f"{minutes}:{seconds:02d}"
[docs]
def to_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"title": self.title,
"version": self.version,
"duration": self.duration,
"trackNumber": self.track_number,
"volumeNumber": self.volume_number,
"trackNumberOnPlaylist": self.position,
"explicit": self.explicit,
"isrc": self.isrc,
"copyrightText": self.copyright_text,
"audioQuality": self.audio_quality,
"mediaTags": list(self.media_tags),
"primaryArtist": _clean(self.primary_artist),
"artists": _clean(self.artists),
"album": _clean(self.album) if self.album is not None else None,
"payToStream": self.pay_to_stream,
"audioModes": list(self.audio_modes),
"streamStartDate": self.stream_start_date,
"mixes": dict(self.mixes),
"replayGain": dict(self.replaygain),
"bpm": self.bpm,
"key": self.key,
"keyScale": self.key_scale,
}
[docs]
@dataclass
class Playlist:
"""A playlist or mix (``is_mix`` distinguishes the two)."""
id: int | str | None = None
title: str | None = None
description: str | None = None
number_of_tracks: int = 0
explicit: bool = False
artwork_id: str | None = None
is_mix: bool = False
# Catalog garnish (live API verified on the v1/v2 playlist
# objects): the creator's display name (editorial playlists have
# none), ISO timestamps, total duration in seconds, the catalog's
# playlist type (EDITORIAL / COMMUNITY / PERSONAL / ...), and the
# follower count.
creator: str | None = None
creation_time: str | None = None
last_modified: str | None = None
duration: int = 0
playlist_type: str | None = None
follower_count: int = 0
# The catalog's popularity (v1 only, an int 0..100 normalised to
# float as on Artist/Album - a trending signal for UIs), and the
# public/private flag (v1 ``publicPlaylist``; ``None`` when the
# catalog doesn't say - the v2 object carries no such flag).
popularity: float = 0.0
is_public: bool | None = None
# The playlist's promoted artists (live API verified, v1 only —
# the v2 object carries none): the featured artist(s) a playlist
# card attributes it to (the row's ``type`` badge, ``MAIN``, is
# dropped). Ids build artist links for the card's deep links.
promoted_artists: list[Artist] = field(default_factory=list)
# When the most recent item was added (live API verified, v1
# only): an ISO timestamp, same shape as ``created`` /
# ``lastUpdated`` — a playlist card's "updated" fact.
last_item_added: str | None = None
# ``None`` entries are non-audio slots (music videos): the reports
# count them as skipped items instead of dropping them silently.
items: list[Track | None] = field(default_factory=list)
@property
def kind(self) -> str:
return "mix" if self.is_mix else "playlist"
@property
def link(self) -> str | None:
"""The linkable URL (``None`` without an id).
Mixes use the player host (``https://play.tidal.com/mc/<id>``),
everything else ``https://tidal.com/playlist/<id>`` (the id is
the uuid for user playlists, the numeric id for editorial
ones). Both forms round-trip through
:func:`tidalpy.api.parse_tidal_url`.
"""
if self.is_mix:
if not self.id:
return None
return f"https://play.tidal.com/mc/{self.id}"
return _resource_link("playlist", self.id)
[docs]
def to_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"uuid": self.id,
"title": self.title,
"name": self.title,
"description": self.description,
"numberOfTracks": self.number_of_tracks,
"explicit": self.explicit,
"artworkId": self.artwork_id,
"type": self.kind,
# Catalog garnish; ``type`` above is the kind (the
# playlist/mix distinction), the catalog's own type
# travels under its own name.
"creator": self.creator,
"created": self.creation_time,
"lastUpdated": self.last_modified,
"duration": self.duration,
"playlistType": self.playlist_type,
"followerCount": self.follower_count,
"popularity": self.popularity,
"isPublic": self.is_public,
"promotedArtists": _clean(self.promoted_artists),
"lastItemAddedAt": self.last_item_added,
}
# The ``type`` value of the ``metadata()`` payload. The host contract
# (tidal_transfer's TidalTask.info JSONField) was written against the
# legacy client's ``--json``/RPC output, which carries the ``Type``
# enum names verbatim — capitalised ("Track", "Album", ...).
_TYPE_NAMES: dict[str, str] = {
"track": "Track",
"album": "Album",
"playlist": "Playlist",
"mix": "Mix",
"artist": "Artist",
}
[docs]
@dataclass
class Resource:
"""A resolved catalog entry: ``album`` / ``track`` / ``playlist`` /
``mix`` / ``artist``."""
kind: str
data: Album | Track | Playlist | Artist
@property
def title(self) -> str | None:
if isinstance(self.data, Artist):
return self.data.name
return getattr(self.data, "title", None)
@property
def track_count(self) -> int | None:
if isinstance(self.data, Track):
return 1
return getattr(self.data, "number_of_tracks", None)
@property
def link(self) -> str | None:
"""The resolved entry's linkable URL (delegates to the model)."""
return self.data.link
[docs]
def to_dict(self) -> dict[str, Any]:
"""``{"type": ..., "data": {...}}`` — the legacy ``--json`` shape.
``type`` is the reference client's ``Type`` enum name (capitalised:
``"Track"``, ``"Album"``, ``"Playlist"``, ``"Mix"``, ``"Artist"``) —
the same value its CLI ``--json``, wrapper json-mode, and RPC
``metadata`` all emit, which the host contract reads. (The
*reports* use the lowercase ``kind`` instead — different
convention, different consumer, see :class:`DownloadReport`.)
"""
return {
"type": _TYPE_NAMES.get(self.kind, self.kind),
"data": self.data.to_dict(),
}
[docs]
@dataclass
class SessionInfo:
"""A live login session (device OAuth or pasted web token)."""
user_id: int | str | None = None
country_code: str | None = None
access_token: str = ""
refresh_token: str | None = None
expires_in: int | None = None
client_id: str | None = None
token_type: str = "device"
subscription: dict[str, Any] | None = None
[docs]
def to_dict(self) -> dict[str, Any]:
return {
"logged_in": bool(self.access_token),
"userId": self.user_id,
"countryCode": self.country_code,
"clientId": self.client_id,
"tokenType": self.token_type,
"expiresIn": self.expires_in,
}
[docs]
@dataclass
class StreamInfo:
"""A resolved stream: where to fetch the media bytes from and what it is.
``urls`` has one entry for bento (single file) streams and several
for DASH (initialization + segments, concatenated in order).
``encryption_key`` is the base64 *security token* from a bento
manifest (``keyId``); ``None`` means the bytes are plain.
"""
track_id: int | str
urls: list[str]
codec: str
delivered_quality: str
requested_quality: str
manifest_mime: str
encryption_key: str | None = None
bandwidth: int | None = None
sample_rate: int | None = None
bit_depth: int | None = None
channels: int | None = None
representation_id: str | None = None
manifest_hash: str | None = None
replaygain: dict[str, str] = field(default_factory=dict)
# True when the manifest carries only the ~30-second preview
# (free-tier accounts, pre-release/exclusive tracks). Set only on
# demand (allow_preview); downloads never resolve with it.
preview: bool = False
@property
def url(self) -> str:
return self.urls[0] if self.urls else ""
@property
def multi_segment(self) -> bool:
return len(self.urls) > 1
[docs]
def to_dict(self) -> dict[str, Any]:
return {
"trackId": self.track_id,
"url": self.url,
"segments": len(self.urls),
"codec": self.codec,
"deliveredQuality": self.delivered_quality,
"requestedQuality": self.requested_quality,
"manifestMime": self.manifest_mime,
"bandwidth": self.bandwidth,
"sampleRate": self.sample_rate,
"bitDepth": self.bit_depth,
"channels": self.channels,
"representationId": self.representation_id,
"manifestHash": self.manifest_hash,
"preview": self.preview,
}
[docs]
@dataclass
class ItemReport:
"""Outcome of one item (track) inside a download run."""
index: int
kind: str
id: int | str | None = None
title: str | None = None
status: str = "pending" # pending | ok | failed | skipped
reason: str = ""
path: str | None = None
delivered_quality: str | None = None
codec: str | None = None
code: str | None = None
retryable: bool = False
size: int = 0
@property
def link(self) -> str | None:
"""The Tidal URL for this item (``None`` without an id).
The same canonical, id-form, round-trip-stable shape as the
catalog models' :attr:`~tidalpy.models.Track.link` — a host
rendering a failed-item row can offer a "view on Tidal" link
without re-deriving ``https://tidal.com/<kind>/<id>``.
"""
return _resource_link(self.kind, self.id)
[docs]
def to_dict(self) -> dict[str, Any]:
"""JSON-safe row for this item (snake_case keys, matching the
legacy client's per-item diagnose rows). All fields are
present; unset optional fields are ``null``."""
data: dict[str, Any] = _clean(asdict(self))
return data
[docs]
@dataclass
class DownloadReport:
"""Structured outcome of a whole download run."""
kind: str
title: str | None = None
requested_quality: str = ""
items: list[ItemReport] = field(default_factory=list)
cancelled: bool = False
@property
def ok(self) -> bool:
if self.cancelled or not self.items:
return False
return all(item.status in ("ok", "skipped") for item in self.items)
@property
def total_bytes(self) -> int:
return sum(item.size for item in self.items if item.status == "ok")
@property
def failures(self) -> list[ItemReport]:
return [item for item in self.items if item.status == "failed"]
@property
def skipped_existing(self) -> int:
return sum(1 for item in self.items if item.status == "skipped")
[docs]
def summary(self) -> dict[str, Any]:
"""A compact, JSON-safe status line for the whole run.
Hosts that display run state (a Django admin column, a batch
UI, a log line) get the counts and the failed items' identity
without walking ``items``; use :meth:`to_dict` for the full
per-item detail. ``failures`` rows are compact: id, title,
the machine ``code``, and the human ``reason``.
Keys (camelCase, as in :meth:`to_dict`): ``kind``, ``title``,
``requestedQuality``, ``ok``, ``cancelled``, ``total``,
``okCount``, ``skippedCount``, ``failedCount``,
``totalBytes``, ``failures``.
"""
counts = {"ok": 0, "skipped": 0, "failed": 0}
for item in self.items:
if item.status in counts:
counts[item.status] += 1
return {
"kind": self.kind,
"title": self.title,
"requestedQuality": self.requested_quality,
"ok": self.ok,
"cancelled": self.cancelled,
"total": len(self.items),
"okCount": counts["ok"],
"skippedCount": counts["skipped"],
"failedCount": counts["failed"],
"totalBytes": self.total_bytes,
"failures": [
{
"id": item.id,
"title": item.title,
"code": item.code,
"reason": item.reason,
}
for item in self.failures
],
}
[docs]
def to_dict(self) -> dict[str, Any]:
"""JSON-safe dict of the whole run. Top-level keys are camelCase
(same convention as the catalog ``metadata()`` output); the
``items`` rows are snake_case (the legacy diagnose row shape)."""
return {
"kind": self.kind,
"title": self.title,
"requestedQuality": self.requested_quality,
"ok": self.ok,
"totalBytes": self.total_bytes,
"cancelled": self.cancelled,
"items": [item.to_dict() for item in self.items],
}