"""Configuration and the OAuth client registry.
All client construction is explicit: a :class:`Config` carries
credentials location, output policy, quality, network policy, and the
OAuth *client* (Tidal's API-key/device pair) to authenticate as. There
is no global state anywhere in the package.
``Config`` is a pydantic model: every field is validated at
construction (a bad value fails fast with
:class:`~tidalpy.errors.ConfigError`, not at first request), the
camelCase aliases used by :meth:`Config.as_dict` and Django's
JSONField are handled declaratively, and instances are immutable —
:meth:`Config.with_updates` returns a fresh, re-validated copy.
The client registry mirrors what the proven reference clients ship:
each entry is a public device-OAuth ``client_id``/``client_secret`` pair
with the quality tiers it is known to unlock. A Bearer token is only
valid for the client that issued it, so the active client is part of
the saved session (see :mod:`tidalpy.session`).
"""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Callable
from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator
from pydantic.alias_generators import to_camel
from .errors import ConfigError
from .quality import Quality, parse_quality
# --------------------------------------------------------------------------
# OAuth client registry
# --------------------------------------------------------------------------
# Stable ids keyed by client_id. Never renumber: a saved login is bound
# to the client that issued its token.
_ANDROID_TV_UA = "TIDAL/3.34.0 (Android TV; Linux; U)"
[docs]
@dataclass(frozen=True)
class OAuthClient:
"""A public device-OAuth client (Tidal "API key")."""
client_id: str
client_secret: str
platform: str
qualities: tuple[str, ...] = ()
user_agent: str = _ANDROID_TV_UA
note: str = ""
[docs]
def supports(self, quality: Quality) -> bool:
return quality.value in self.qualities
_CLIENTS: dict[str, OAuthClient] = {}
def _register(
client_id: str,
platform: str,
secret: str,
qualities: tuple[str, ...],
user_agent: str = _ANDROID_TV_UA,
note: str = "",
) -> None:
_CLIENTS[client_id] = OAuthClient(
client_id=client_id,
client_secret=secret,
platform=platform,
qualities=qualities,
user_agent=user_agent,
note=note,
)
_register(
"OmDtrzFgyVVL6uW56OnFA2COiabqm",
"Fire TV (legacy)",
"zxen1r3pO0hgtOC7j6twMo9UAqngGrmRiWpV7QC1zJ8=",
("LOW", "HIGH", "LOSSLESS"),
note="libopenTIDAL; legacy, playback availability varies",
)
_register(
"7m7Ap0JC9j1cOM3n",
"Fire TV (master)",
"vRAdA108tlvkJpTsGZS8rGZ7xTlbJ0qaZ2K9saEzsgY=",
("HI_RES_LOSSLESS",),
note="Dniel97/RedSea; Master-only (else error)",
)
_register(
"Pzd0ExNVHkyZLiYN",
"Android TV",
"W7X6UvBaho+XOi1MUeCX6ewv2zTdSOV3Y7qC3p3675I=",
("LOW", "HIGH", "LOSSLESS"),
note="reported invalid as of 2024",
)
_register(
"8SEZWa4J1NVC5U5Y",
"TV",
"owUYDkxddz+9FpvGX24DlxECNtFEMBxipU0lBfrbq60=",
("LOW", "HIGH", "LOSSLESS", "HI_RES_LOSSLESS"),
note="morguldir/python-tidal",
)
_register(
"cgiF7TQuB97BUIu3",
"Orpheus TV",
"1nqpgx8uvBdZigrx4hUPDV2hOwgYAAAG5DYXOr6uNf8=",
("LOW", "HIGH", "LOSSLESS", "HI_RES_LOSSLESS"),
note="OrpheusDL tidal module; Max (24/192) reported",
)
_register(
"zU4XHVVkc2tDPo4t",
"Android Auto",
"VJKhDFqJPqvsPVNBV6ukXTJmwlvbttP7wlMlrc72se4=",
("LOW", "HIGH", "LOSSLESS", "HI_RES_LOSSLESS"),
note="Tidal-Media-Downloader PR #840",
)
_register(
"fX2JxdmntZWK0ixT",
"Tidekeeper OAuth",
"1Nn9AfDAjxrgJFJbKNWLeAyKGVGmINuXPPLHVXAvxAg=",
("LOW", "HIGH", "LOSSLESS", "HI_RES_LOSSLESS"),
note="default (tidalapi 0.8.11 OAuth client)",
)
_register(
"4N3n6Q1x95LL5K7p",
"Tidal TV",
"oKOXfJW371cX6xaZ0PyhgGNBdNLlBZd4AKKYougMjik=",
("LOW", "HIGH", "LOSSLESS", "HI_RES_LOSSLESS"),
note="TV device flow (tiddl-elvigilante)",
)
DEFAULT_CLIENT_ID = "fX2JxdmntZWK0ixT"
[docs]
def get_client(client_id: str | None = None) -> OAuthClient:
"""Look up a registered client by id (default client when ``None``)."""
key = (client_id or DEFAULT_CLIENT_ID).strip()
client = _CLIENTS.get(key)
if client is None:
known = ", ".join(sorted(_CLIENTS))
raise ConfigError(f"Unknown OAuth client id: {key!r}. Known: {known}")
return client
[docs]
def list_clients() -> list[OAuthClient]:
return list(_CLIENTS.values())
[docs]
def clients_for(quality: str | Quality | None = None) -> list[OAuthClient]:
"""Registered clients that can serve ``quality``, best first.
The registry documents per-client quality support, but nothing
*used* it: a host that needs "a client that can actually deliver
this quality" (a login that should target the right client from the
start) had to walk :func:`list_clients` and :meth:`OAuthClient.supports`
itself. ``quality`` takes the same labels as the rest of the
library (aliases included, e.g. ``"24/192"``); ``None``/unknown
resolve to the default profile.
The default client comes first when it claims the quality (its
login is usually the one a host already has), then the other
claiming clients in registry order (the stable id order). Raises
:class:`~tidalpy.errors.ConfigError` when no registered client
claims the quality — a registry gap, not a host error.
"""
profile = parse_quality(quality)
supporting = [client for client in _CLIENTS.values() if client.supports(profile)]
if not supporting:
raise ConfigError(
f"no registered OAuth client claims quality {profile.value!r}; "
"extend the client registry before requesting it"
)
default = _CLIENTS[DEFAULT_CLIENT_ID]
if default.supports(profile):
return [default] + [c for c in supporting if c is not default]
return supporting
# --------------------------------------------------------------------------
# Config
# --------------------------------------------------------------------------
DEFAULT_TOKEN_PATH = Path.home() / ".tidalpy" / "token.json"
DEFAULT_OUTPUT = Path("tidal_downloads")
def _socks_available() -> bool:
from importlib.util import find_spec
return find_spec("socks") is not None
def _validate_proxy(url: str) -> None:
"""Fail fast on a proxy URL that could only fail at request time.
``http``/``https`` are always supported; ``socks4(a)``/``socks5(h)``
need the optional ``PySocks`` dependency (the ``socks`` extra) and
get a clear error at construction instead of a mid-download
``Unrecognized URL`` from urllib3.
"""
from urllib.parse import urlsplit
parts = urlsplit(str(url).strip())
if not parts.hostname:
raise ConfigError(f"proxy URL has no host: {url!r}")
if parts.scheme in ("socks4", "socks4a", "socks5", "socks5h"):
if not _socks_available():
raise ConfigError(
"socks proxy requires the optional PySocks dependency "
f"(install the 'socks' extra, e.g. `uv add "
f"'binary-tidal-py[socks]'`): {url!r}"
)
return
if parts.scheme not in ("http", "https"):
raise ConfigError(
f"proxy scheme {parts.scheme!r} is not supported "
f"(use http, https, or socks with the 'socks' extra): {url!r}"
)
[docs]
class Config(BaseModel):
"""Validated, immutable configuration for a
:class:`tidalpy.client.TidalClient`.
Both ``snake_case`` field names and the camelCase aliases used by
:meth:`as_dict` are accepted by the constructor and
:meth:`from_dict`. Invalid values raise
:class:`~tidalpy.errors.ConfigError` immediately.
"""
model_config = ConfigDict(
frozen=True,
populate_by_name=True,
extra="forbid",
alias_generator=to_camel,
)
# --- auth / session -------------------------------------------------
token_path: Path | None = None
# Which registered OAuth client (platform persona) to present. The
# registry is :func:`list_clients` (see :data:`_CLIENTS`); pick one
# with :func:`get_client` or the quality-aware :func:`clients_for`.
# A saved login is *bound* to the client that minted its token
# (switching clients here logs the host out of the saved session
# on purpose), and a 404/subStatus 4022 clears it after one
# refresh attempt. ``None`` = :data:`DEFAULT_CLIENT_ID`;
# ``TIDALPY_CLIENT_ID`` in :func:`config_from_env` overrides it.
client_id: str | None = None
# --- output -----------------------------------------------------------
output: Path | None = None
track_format: str = "{number} {artist} - {title}{explicit}"
album_format: str = "{artist} - {album} {year}"
playlist_format: str = "Playlist - {title}"
# Artist downloads/listings include EPs and singles (the
# ``EPSANDSINGLES`` filter). True matches the reference client's
# effective default.
include_ep: bool = True
# --- quality ------------------------------------------------------------
quality: Quality = Quality.LOSSLESS
# --- network -------------------------------------------------------------
# Base URL overrides (advanced / testing). Production traffic must go
# to Tidal's own hosts; these exist so hosts can point a client at a
# replay proxy or a local test server. ``None`` = the real endpoint.
api_base: str | None = None
legacy_api_base: str | None = None
openapi_base: str | None = None
auth_base: str | None = None
# ``"http://host:port"`` (applied to both schemes) or the
# requests-style mapping ``{"http": ..., "https": ...}``.
proxies: str | dict[str, str] | None = None
# TLS verification override: ``False`` disables, a CA bundle path
# enables a custom trust store (corporate MITM proxies). ``None`` =
# requests' default (certifi).
verify_tls: bool | str | None = None
# requests only supports (connect, read) timeouts; there is no
# write timeout in its API.
timeout_connect: float = Field(default=10.0, gt=0)
timeout_read: float = Field(default=60.0, gt=0)
max_retries: int = Field(default=3, ge=0)
retry_backoff: float = Field(default=1.0, ge=0)
# 429 handling budget per logical request (reference-fork parity:
# the total *time* spent waiting, not a count — an account-level
# rate limit may need several short waits to clear).
rate_limit_max_wait: float = Field(default=90.0, ge=0)
# Cumulative cap for *long* server-requested pauses (Retry-After
# beyond ``rate_limit_max_wait``), which are honoured rather than
# failed: Tidal answers every request 429 until the pause lapses,
# so failing one request would fail every following one too.
rate_limit_max_pause: float = Field(default=6 * 60 * 60, ge=0)
asset_not_ready_max_waits: int = Field(default=6, ge=0)
user_agent: str | None = None
# --- downloads ---------------------------------------------------------
download_workers: int = Field(default=8, ge=1)
part_size: int = Field(default=1024 * 1024, ge=65536)
# Throttle download throughput in bits/s. ``None`` (default) = no
# throttle; zero/negative would be meaningless, so they fail fast.
pacing_bps: float | None = Field(default=None, gt=0)
# SSRF guard for manifest-supplied media URLs. Disable only when the
# host owns the URLs itself (replay proxies, offline test servers) —
# production must leave this on.
validate_media_urls: bool = True
# --- post-processing ------------------------------------------------------
write_tags: bool = True
# Fetch lyrics per track and embed them in the tags (one extra API
# call per item; off by default).
write_lyrics: bool = False
# Write the lyrics next to the media file as a sidecar (``.lrc``
# when Tidal has timed subtitles, ``.txt`` for plain text; TMD
# parity — some players/DJ tools read sidecars). Shares the
# ``write_lyrics`` fetch, so enabling both costs no extra call.
lyric_file: bool = False
download_cover: bool = True
save_as_flac: bool = True
verify_media: bool = True
# Write a completion receipt (``<final>.tidalpy.json``) next to
# each finished media file and make the skip-existing checks
# consult it: a file is skipped only when the receipt proves it is
# this track, intact, fully post-processed, and at least the
# requested quality. Off by default: the plain skip contract
# (non-empty file at the final path = complete) stays the
# documented behaviour; hosts that re-run collections at a higher
# quality ceiling (or crash-prone long transfers) turn it on.
save_receipts: bool = False
# --- collection behaviour ---------------------------------------------------
skip_broken: bool = True
# -- validation -----------------------------------------------------------
@field_validator("quality", mode="before")
@classmethod
def _parse_quality(cls, value: object) -> object:
if value is not None and not isinstance(value, (str, Quality)):
raise ConfigError(f"invalid quality value: {value!r}")
return parse_quality(value)
@field_validator("proxies")
@classmethod
def _check_proxies(cls, value: object) -> object:
if isinstance(value, str):
_validate_proxy(value)
elif isinstance(value, dict):
for scheme, url in value.items():
if scheme not in ("http", "https"):
raise ConfigError(f"unknown proxy scheme: {scheme!r}")
_validate_proxy(url)
return value
[docs]
def __init__(self, /, **data: object) -> None:
"""Construction is validated; invalid values raise
:class:`~tidalpy.errors.ConfigError` (not a raw
``ValidationError``)."""
try:
super().__init__(**data)
except ValidationError as exc:
raise ConfigError(f"invalid configuration: {exc}") from exc
# --- resolved views ----------------------------------------------------
@property
def token_file(self) -> Path:
# ``~`` is expanded: hosts and the docs pass "~/..." strings.
return (self.token_path or DEFAULT_TOKEN_PATH).expanduser()
@property
def output_dir(self) -> Path:
return (self.output or DEFAULT_OUTPUT).expanduser()
@property
def oauth_client(self) -> OAuthClient:
return get_client(self.client_id)
[docs]
def proxies_dict(self) -> dict[str, str] | None:
if not self.proxies:
return None
if isinstance(self.proxies, str):
normalized = {"http": self.proxies, "https": self.proxies}
else:
normalized = dict(self.proxies)
if normalized.get("https") and not normalized.get("http"):
normalized["http"] = normalized["https"]
elif normalized.get("http") and not normalized.get("https"):
normalized["https"] = normalized["http"]
return normalized or None
[docs]
def user_agent_for(self, client: OAuthClient) -> str:
return self.user_agent or client.user_agent
# -- (de)serialisation ---------------------------------------------------
[docs]
def with_updates(self, **kwargs: object) -> "Config":
"""A copy of this config with the given fields replaced.
Runs the same validation as the constructor; an invalid value
raises :class:`~tidalpy.errors.ConfigError` and leaves this
object untouched.
"""
merged = {**self.model_dump(), **kwargs}
try:
return type(self).model_validate(merged)
except ConfigError:
raise
except ValidationError as exc:
raise ConfigError(f"invalid config value: {exc}") from exc
[docs]
def as_dict(self) -> dict[str, object]:
"""JSON-serialisable snapshot (for logs, task rows, admin UI)."""
data = self.model_dump(mode="json")
data["quality"] = self.quality.value
return data
[docs]
@classmethod
def from_dict(cls, data: dict[str, object]) -> "Config":
"""Build a :class:`Config` from a plain dict (JSON-safe).
Accepts both ``snake_case`` and the camelCase keys used by
:meth:`as_dict` and Django's JSONField settings. Unknown keys
raise :class:`~tidalpy.errors.ConfigError` (a typo should fail
fast, not silently pick a default). Explicit ``null`` values
are treated as "unspecified" and keep the model default
(Django form/JSON round-trips serialize unset values as
``None``; the reference client applies the same rule — a
naive ``None`` would clobber a working default).
"""
if not isinstance(data, dict):
raise ConfigError("config data must be a mapping")
for key in data:
if key not in cls._known_keys():
raise ConfigError(f"unknown config key: {key!r}")
try:
return cls.model_validate({k: v for k, v in data.items() if v is not None})
except ConfigError:
raise
except ValidationError as exc:
raise ConfigError(f"invalid configuration: {exc}") from exc
@classmethod
def _known_keys(cls) -> frozenset[str]:
keys: set[str] = set()
for name, info in cls.model_fields.items():
keys.add(name)
if info.alias:
keys.add(info.alias)
return frozenset(keys)
def _env_str(raw: str, _var: str) -> str:
return raw.strip()
def _env_bool(raw: str, var: str) -> bool:
value = raw.strip().lower()
if value in ("1", "true", "yes", "on"):
return True
if value in ("0", "false", "no", "off"):
return False
raise ConfigError(
f"{var} must be a boolean (1/0, true/false, yes/no, on/off): {raw!r}"
)
def _env_int(raw: str, var: str) -> int:
try:
return int(raw.strip())
except ValueError as exc:
raise ConfigError(f"{var} is not a valid number: {raw!r}") from exc
def _env_float(raw: str, var: str) -> float:
try:
return float(raw.strip())
except ValueError as exc:
raise ConfigError(f"{var} is not a valid number: {raw!r}") from exc
def _env_proxies(raw: str, _var: str) -> dict[str, str]:
proxy = raw.strip()
return {"http": proxy, "https": proxy}
def _env_quality(raw: str, _var: str) -> Quality:
return parse_quality(raw.strip())
def _env_verify_tls(raw: str, var: str) -> bool | str:
"""A truthy/falsy value, or a CA bundle path (corporate MITM)."""
try:
return _env_bool(raw, var)
except ConfigError:
return raw.strip()
#: The full set of recognised environment overrides:
#: ``(variable, Config field, parser)``. One entry per configurable
#: field, so the env surface cannot drift from the model — adding a
#: variable is a one-liner, and the docstring stays honest by
#: construction.
_ENV_OVERRIDES: tuple[tuple[str, str, Callable[[str, str], object]], ...] = (
("TIDALPY_TOKEN_PATH", "token_path", _env_str),
("TIDALPY_OUTPUT", "output", _env_str),
("TIDALPY_QUALITY", "quality", _env_quality),
("TIDALPY_CLIENT_ID", "client_id", _env_str),
("TIDALPY_TRACK_FORMAT", "track_format", _env_str),
("TIDALPY_ALBUM_FORMAT", "album_format", _env_str),
("TIDALPY_PLAYLIST_FORMAT", "playlist_format", _env_str),
("TIDALPY_INCLUDE_EP", "include_ep", _env_bool),
("TIDALPY_API_BASE", "api_base", _env_str),
("TIDALPY_LEGACY_API_BASE", "legacy_api_base", _env_str),
("TIDALPY_OPENAPI_BASE", "openapi_base", _env_str),
("TIDALPY_AUTH_BASE", "auth_base", _env_str),
("TIDALPY_PROXY", "proxies", _env_proxies),
("TIDALPY_VERIFY_TLS", "verify_tls", _env_verify_tls),
("TIDALPY_TIMEOUT_CONNECT", "timeout_connect", _env_float),
("TIDALPY_TIMEOUT_READ", "timeout_read", _env_float),
("TIDALPY_MAX_RETRIES", "max_retries", _env_int),
("TIDALPY_RETRY_BACKOFF", "retry_backoff", _env_float),
("TIDALPY_RATE_LIMIT_MAX_WAIT", "rate_limit_max_wait", _env_float),
("TIDALPY_RATE_LIMIT_MAX_PAUSE", "rate_limit_max_pause", _env_float),
(
"TIDALPY_ASSET_NOT_READY_MAX_WAITS",
"asset_not_ready_max_waits",
_env_int,
),
("TIDALPY_USER_AGENT", "user_agent", _env_str),
("TIDALPY_DOWNLOAD_WORKERS", "download_workers", _env_int),
("TIDALPY_PART_SIZE", "part_size", _env_int),
("TIDALPY_PACING_BPS", "pacing_bps", _env_float),
("TIDALPY_VALIDATE_MEDIA_URLS", "validate_media_urls", _env_bool),
("TIDALPY_WRITE_TAGS", "write_tags", _env_bool),
("TIDALPY_SAVE_RECEIPTS", "save_receipts", _env_bool),
("TIDALPY_WRITE_LYRICS", "write_lyrics", _env_bool),
("TIDALPY_LYRIC_FILE", "lyric_file", _env_bool),
("TIDALPY_DOWNLOAD_COVER", "download_cover", _env_bool),
("TIDALPY_SAVE_AS_FLAC", "save_as_flac", _env_bool),
("TIDALPY_VERIFY_MEDIA", "verify_media", _env_bool),
("TIDALPY_SKIP_BROKEN", "skip_broken", _env_bool),
)
[docs]
def config_from_env(env: dict[str, str] | None = None) -> Config:
"""Build a :class:`Config` with ``TIDALPY_``-prefixed environment overrides.
Every configurable field has a ``TIDALPY_`` variable (the full
table is :data:`_ENV_OVERRIDES`): ``TIDALPY_TOKEN_PATH``,
``TIDALPY_OUTPUT``, ``TIDALPY_QUALITY``, ``TIDALPY_PROXY``,
``TIDALPY_WRITE_TAGS``, ``TIDALPY_WRITE_LYRICS``,
``TIDALPY_INCLUDE_EP``, ``TIDALPY_DOWNLOAD_WORKERS``,
``TIDALPY_PART_SIZE``, ``TIDALPY_MAX_RETRIES``,
``TIDALPY_TIMEOUT_CONNECT``, ``TIDALPY_TIMEOUT_READ``, and so on.
Handy for the Django side to inject per-deployment settings
without code changes.
Values are parsed with the same strictness as constructor
arguments: an unparseable value raises :class:`ConfigError`
(``TIDALPY_WRITE_TAGS=ture`` is a deploy bug, not a silent
``False``). Unset or empty variables keep the default.
``env`` defaults to :data:`os.environ`; tests pass a plain dict.
"""
import os
source = os.environ if env is None else env
kwargs: dict[str, object] = {}
for var, field, parse in _ENV_OVERRIDES:
raw = source.get(var)
if raw is None or not raw.strip():
continue # unset or empty: keep the model default
kwargs[field] = parse(raw, var)
return Config(**kwargs)