music-assistant-server

1"""API client wrapper for Yandex Music."""
2
3from __future__ import annotations
4
5import asyncio
6import base64
7import hashlib
8import hmac
9import logging
10import random
11import re
12import time
13from collections import OrderedDict, defaultdict, deque
14from collections.abc import Awaitable, Callable
15from datetime import datetime
16from typing import TYPE_CHECKING, Any, Final, Literal, TypeVar, cast
17
18from music_assistant_models.errors import (
19    LoginFailed,
20    ProviderUnavailableError,
21    RateLimited,
22    ResourceTemporarilyUnavailable,
23)
24from yandex_music import Album as YandexAlbum
25from yandex_music import Artist as YandexArtist
26from yandex_music import ClientAsync, MixLink, Search, TrackShort
27from yandex_music import Playlist as YandexPlaylist
28from yandex_music import Track as YandexTrack
29from yandex_music.exceptions import BadRequestError, NetworkError, UnauthorizedError
30from yandex_music.utils.sign_request import DEFAULT_SIGN_KEY
31
32from music_assistant.helpers.datetime import utc
33from music_assistant.helpers.throttle_retry import BYPASS_THROTTLER, Throttler
34
35if TYPE_CHECKING:
36    from ya_passport_auth import SecretStr
37    from yandex_music import DownloadInfo
38    from yandex_music.feed.feed import Feed
39    from yandex_music.landing.chart_info import ChartInfo
40    from yandex_music.landing.landing import Landing
41    from yandex_music.landing.landing_list import LandingList
42    from yandex_music.rotor.dashboard import Dashboard
43    from yandex_music.rotor.station_result import StationResult
44
45from .constants import (
46    CAPTCHA_COOLDOWN_LADDER_S,
47    CAPTCHA_STRIKE_RETENTION_S,
48    DEFAULT_LIMIT,
49    FILE_INFO_CACHE_MAX,
50    FILE_INFO_CACHE_TTL_S,
51    INITIAL_SYNC_JITTER_S,
52    INITIAL_SYNC_WINDOW_S,
53    LIKED_BATCH_JITTER_MIN_S,
54    LIKED_BATCH_JITTER_SPAN_S,
55    RATE_LIMIT_COOLDOWN_S,
56    RESTRICTIVE_GLOBAL_CONCURRENCY,
57    THROTTLE_DEFAULT_RPS,
58    THROTTLE_FILE_INFO_RPS,
59    THROTTLE_METADATA_RPS,
60    THROTTLE_ROTOR_RPS,
61)
62
63_CAPTCHA_MARKERS: Final = ("smart-captcha", "captcha_smart_qrcode", "about-429.html")
64
65# get-file-info with quality=lossless returns FLAC; default /tracks/.../download-info often does not
66# Prefer flac-mp4/aac-mp4 (Yandex API moved to these formats around 2025)
67GET_FILE_INFO_CODECS = "flac-mp4,flac,aac-mp4,aac,he-aac,mp3,he-aac-mp4"
68
69LOGGER = logging.getLogger(__name__)
70
71_T = TypeVar("_T")
72
73
74def _liked_track_sort_key(track: Any) -> datetime:
75    """
76    Return a naive ``datetime`` for sorting liked tracks chronologically.
77
78    Yandex's ``TrackShort.timestamp`` is sometimes tz-aware and sometimes
79    tz-naive depending on the upstream library version; mixing the two
80    triggers ``TypeError`` in ``sorted``. Strip ``tzinfo`` and fall back to
81    ``datetime.min`` when the field is missing.
82    """
83    ts = getattr(track, "timestamp", None)
84    if not isinstance(ts, datetime):
85        return datetime.min  # noqa: DTZ901 — naive sentinel by design (see docstring)
86    if ts.tzinfo is not None:
87        return ts.replace(tzinfo=None)
88    return ts
89
90
91class YandexMusicClient:
92    """Wrapper around yandex-music-api ClientAsync."""
93
94    def __init__(
95        self,
96        token: SecretStr,
97        base_url: str | None = None,
98        *,
99        restrictive_rate_limits: bool = False,
100    ) -> None:
101        """
102        Initialize the Yandex Music client.
103
104        :param token: Yandex Music OAuth token (wrapped in SecretStr).
105        :param base_url: Optional API base URL (defaults to Yandex Music API).
106        :param restrictive_rate_limits: When True, applies a token-wide
107            concurrency cap (``RESTRICTIVE_GLOBAL_CONCURRENCY``) on top of
108            the per-kind throttler and per-endpoint lock — for users on
109            VPS / datacenter / VPN IPs where Yandex's edge enforces a
110            tighter anti-scraper concurrency limit.
111        """
112        self._token = token
113        self._base_url = base_url
114        self._client: ClientAsync | None = None
115        self._user_id: int | None = None
116        self._last_reconnect_at: float = -30.0  # allow first reconnect immediately
117        self._reconnect_lock = asyncio.Lock()
118        # Per-kind throttlers. Yandex's smart-captcha quota is per-endpoint-family,
119        # so we keep a separate token bucket per logical class and let one kind
120        # back off independently of the others. `metadata` covers the artist/album
121        # refresh burst MA fires during initial sync (see #146).
122        self._throttlers: dict[str, Throttler] = {
123            "default": Throttler(rate_limit=THROTTLE_DEFAULT_RPS, period=1.0),
124            "metadata": Throttler(rate_limit=THROTTLE_METADATA_RPS, period=1.0),
125            "file_info": Throttler(rate_limit=THROTTLE_FILE_INFO_RPS, period=1.0),
126            "rotor": Throttler(rate_limit=THROTTLE_ROTOR_RPS, period=1.0),
127        }
128        # Per-kind captcha quarantine deadlines (monotonic). Only the explicit
129        # smart-captcha page sets a deadline; plain 429 leaves these at 0.
130        self._block_until: dict[str, float] = dict.fromkeys(self._throttlers, 0.0)
131        # Per-kind captcha strike timestamps (monotonic), trimmed to the
132        # CAPTCHA_STRIKE_RETENTION_S window on every push. Drives the
133        # CAPTCHA_COOLDOWN_LADDER_S escalation.
134        self._captcha_strikes: dict[str, deque[float]] = defaultdict(deque)
135        # Set when connect() succeeds. Drives the initial-sync jitter window.
136        self._connected_at: float | None = None
137        # Short-TTL cache for /get-file-info results, keyed by
138        # (track_id, quality, codecs, transport). Bounded by FILE_INFO_CACHE_MAX (LRU).
139        self._file_info_cache: OrderedDict[
140            tuple[str, str, str, str], tuple[float, dict[str, Any]]
141        ] = OrderedDict()
142        # Per-endpoint concurrency locks. Yandex's edge layer reacts to
143        # concurrent requests to the same URL family (per-endpoint scraper
144        # signature), not steady-state RPS. Defense-in-depth on top of the
145        # per-kind throttler: even if a caller fans out via
146        # ``asyncio.gather`` over the same method, the lock serialises the
147        # actual HTTP requests to ≤1 concurrent per endpoint. Created lazily
148        # on first use to keep the dict small. Lifetime tied to the client
149        # instance (rebuilt on reconnect / token rotation).
150        self._endpoint_locks: dict[str, asyncio.Lock] = {}
151        # Restrictive mode: optional global token-wide concurrency cap.
152        # When set, every call through ``_call_with_retry`` must acquire
153        # this semaphore before firing — so the total in-flight count
154        # across all kinds and endpoints can never exceed
155        # ``RESTRICTIVE_GLOBAL_CONCURRENCY``. Lives at the client level
156        # because Yandex's edge enforces the cap per-token, not per-kind.
157        self._global_concurrency: asyncio.Semaphore | None = (
158            asyncio.Semaphore(RESTRICTIVE_GLOBAL_CONCURRENCY) if restrictive_rate_limits else None
159        )
160
161    def _get_throttler(self, kind: str) -> Throttler:
162        return self._throttlers.get(kind, self._throttlers["default"])
163
164    def _get_endpoint_lock(self, endpoint: str) -> asyncio.Lock:
165        """Return (creating on demand) the per-endpoint serialization lock."""
166        lock = self._endpoint_locks.get(endpoint)
167        if lock is None:
168            lock = asyncio.Lock()
169            self._endpoint_locks[endpoint] = lock
170        return lock
171
172    @staticmethod
173    def _derive_endpoint(func: Callable[..., Any]) -> str | None:
174        """
175        Extract a stable endpoint key from a lambda's enclosing method.
176
177        Most ``_call_with_retry`` callers pass a lambda defined inside a
178        ``YandexMusicClient.<method>``; the lambda's ``__qualname__`` reads as
179        ``YandexMusicClient.<method>.<locals>.<lambda>``. We trim the
180        ``<locals>...`` suffix to get a per-method endpoint key, which
181        mirrors Yandex's per-URL-family edge limit. Returns ``None`` when
182        the qualname is missing or not in lambda form — in that case the
183        per-endpoint lock is skipped (no behaviour change for that call).
184        """
185        qn = getattr(func, "__qualname__", "")
186        if not qn:
187            return None
188        if ".<locals>." in qn:
189            return qn.split(".<locals>.", 1)[0]
190        return qn
191
192    @property
193    def user_id(self) -> int:
194        """Return the user ID."""
195        if self._user_id is None:
196            raise ProviderUnavailableError("Client not initialized, call connect() first")
197        return self._user_id
198
199    async def connect(self) -> bool:
200        """
201        Initialize the client and verify token validity.
202
203        :return: True if connection was successful.
204        :raises LoginFailed: If the token is invalid.
205        """
206        try:
207            self._client = await ClientAsync(
208                self._token.get_secret(), base_url=self._base_url
209            ).init()
210            if self._client.me is None or self._client.me.account is None:
211                raise LoginFailed("Failed to get account info")
212            self._user_id = self._client.me.account.uid
213            self._connected_at = time.monotonic()
214            LOGGER.debug("Connected to Yandex Music as user %s", self._user_id)
215            return True
216        except UnauthorizedError as err:
217            raise LoginFailed("Invalid Yandex Music token") from err
218        except NetworkError as err:
219            msg = "Network error connecting to Yandex Music"
220            raise ResourceTemporarilyUnavailable(msg) from err
221
222    async def disconnect(self) -> None:
223        """Disconnect the client."""
224        self._client = None
225        self._user_id = None
226        self._connected_at = None
227
228    async def _ensure_connected(self) -> ClientAsync:
229        """Ensure the client is connected, attempting reconnect if needed."""
230        if self._client is not None:
231            return self._client
232        async with self._reconnect_lock:
233            # Re-check after acquiring lock — another task may have connected already
234            if self._client is not None:
235                return self._client  # type: ignore[unreachable]
236            LOGGER.info("Client disconnected, attempting to reconnect...")
237            try:
238                await self.connect()
239            except LoginFailed:
240                raise
241            except Exception as err:
242                raise ProviderUnavailableError("Client not connected and reconnect failed") from err
243        return cast("ClientAsync", self._client)
244
245    def _is_connection_error(self, err: Exception) -> bool:
246        """
247        Return True if the exception indicates a connection or server drop.
248
249        ``BadRequestError`` upstream extends ``NetworkError`` but represents a
250        terminal 4xx response (malformed query, geo-block) — retry-on-reconnect
251        would just reproduce the same failure and waste a connection cycle,
252        so it is explicitly excluded.
253        """
254        if isinstance(err, BadRequestError):
255            return False
256        if isinstance(err, NetworkError) and not self._is_rate_limit_error(err):
257            return True
258        msg = str(err).lower()
259        return "disconnect" in msg or "connection" in msg or "timeout" in msg
260
261    def _classify_429(self, err: Exception) -> Literal["captcha", "rate_limit", "other"]:
262        """
263        Classify a 429-ish error: smart-captcha edge block vs plain rate-limit.
264
265        Yandex returns an HTML smart-captcha page when its anti-bot edge layer
266        decides an endpoint family is too hot. That page is per-endpoint, not
267        per-IP, and warrants a longer cooldown than an ordinary 429.
268        """
269        if not isinstance(err, NetworkError):
270            return "other"
271        low = str(err).lower()
272        is_429 = "429" in low or "too many requests" in low or "rate limit" in low
273        if not is_429:
274            return "other"
275        # 429 payload dump for forensics: which markers actually matched and
276        # the first 2000 chars of the body. Captured at DEBUG so a single
277        # captcha trip in production can be reconstructed by flipping the
278        # provider log level — without flooding steady-state logs.
279        if LOGGER.isEnabledFor(logging.DEBUG):
280            matched_markers = [m for m in _CAPTCHA_MARKERS if m in low]
281            LOGGER.debug(
282                "429 classify forensics: markers_matched=%s body[:2000]=%r",
283                matched_markers,
284                str(err)[:2000],
285            )
286        return "captcha" if any(m in low for m in _CAPTCHA_MARKERS) else "rate_limit"
287
288    def _is_rate_limit_error(self, err: Exception) -> bool:
289        """Return True if the exception indicates a rate-limit response from Yandex."""
290        return self._classify_429(err) != "other"
291
292    @staticmethod
293    def _truncate_err_msg(err: Exception, limit: int = 200) -> str:
294        """Cap a NetworkError message so the captcha HTML body never lands in logs."""
295        msg = str(err)
296        return msg if len(msg) <= limit else msg[:limit] + "...[truncated]"
297
298    async def _reconnect(self) -> None:
299        """
300        Disconnect and connect again to recover from Server disconnected / connection errors.
301
302        Enforces a 30-second cooldown between reconnect attempts to avoid hammering Yandex
303        and triggering rate limiting. A lock ensures concurrent callers don't bypass the cooldown.
304        """
305        async with self._reconnect_lock:
306            now = time.monotonic()
307            if now - self._last_reconnect_at < 30.0:
308                raise ProviderUnavailableError("Reconnect cooldown active, skipping")
309            self._last_reconnect_at = now
310            await self.disconnect()
311            await self.connect()
312
313    def _check_block(self, kind: str) -> None:
314        """
315        Raise immediately if `kind` is under a captcha quarantine.
316
317        BYPASS_THROTTLER callers (stream URL refresh) must skip this check so a
318        currently playing track isn't dropped mid-stream when an unrelated
319        endpoint family trips smart-captcha.
320        """
321        deadline = self._block_until.get(kind, 0.0)
322        remaining = deadline - time.monotonic()
323        if remaining > 0:
324            raise ResourceTemporarilyUnavailable(
325                f"Yandex Music {kind} cooldown active",
326                backoff_time=int(remaining) + 1,
327            )
328
329    def _trigger_captcha_block(self, kind: str) -> int:
330        """
331        Quarantine the given throttler kind using the captcha-cooldown ladder.
332
333        Only called when _classify_429 == "captcha". Plain rate-limit responses
334        do NOT trigger this, since Yandex's smart-captcha bucket is per
335        endpoint family and we don't want to gate unrelated traffic.
336
337        :param kind: Throttler bucket name (e.g. "default", "metadata").
338        :return: The cooldown duration in seconds (rounded down to int).
339        """
340        now = time.monotonic()
341        strikes = self._captcha_strikes[kind]
342        cutoff = now - CAPTCHA_STRIKE_RETENTION_S
343        while strikes and strikes[0] < cutoff:
344            strikes.popleft()
345        strikes.append(now)
346        ladder = CAPTCHA_COOLDOWN_LADDER_S
347        idx = min(len(strikes), len(ladder)) - 1
348        cooldown = ladder[idx]
349        self._block_until[kind] = max(self._block_until.get(kind, 0.0), now + cooldown)
350        LOGGER.warning(
351            "Yandex Music %s captcha cooldown engaged: %.0fs (strike %d/%d in last %.0fs)",
352            kind,
353            cooldown,
354            len(strikes),
355            len(ladder),
356            CAPTCHA_STRIKE_RETENTION_S,
357        )
358        return int(cooldown)
359
360    def _maybe_handle_429(self, err: Exception, kind: str) -> ResourceTemporarilyUnavailable | None:
361        """
362        Classify a 429 error and build the user-facing exception.
363
364        Returns the prepared `ResourceTemporarilyUnavailable` to raise, or
365        ``None`` if the error isn't a 429 (caller should re-raise or fall
366        through to connection-error handling). Always truncates the message
367        so the smart-captcha HTML body never lands in logs.
368
369        Side effect: a captcha-classified result engages the per-kind block
370        deadline. Plain 429 leaves block deadlines untouched.
371        """
372        classified = self._classify_429(err)
373        if classified == "captcha":
374            backoff = self._trigger_captcha_block(kind)
375            return ResourceTemporarilyUnavailable(
376                f"Yandex Music captcha ({kind})",
377                backoff_time=backoff,
378            )
379        if classified == "rate_limit":
380            LOGGER.debug("Yandex Music plain 429 on kind=%s", kind)
381            return RateLimited(
382                "Yandex Music rate limit",
383                backoff_time=int(RATE_LIMIT_COOLDOWN_S),
384            )
385        return None
386
387    def _file_info_cache_get(self, key: tuple[str, str, str, str]) -> dict[str, Any] | None:
388        entry = self._file_info_cache.get(key)
389        if entry is None:
390            return None
391        expires_at, value = entry
392        if time.monotonic() >= expires_at:
393            self._file_info_cache.pop(key, None)
394            return None
395        self._file_info_cache.move_to_end(key)
396        return value
397
398    def _file_info_cache_put(self, key: tuple[str, str, str, str], value: dict[str, Any]) -> None:
399        self._file_info_cache[key] = (
400            time.monotonic() + FILE_INFO_CACHE_TTL_S,
401            value,
402        )
403        self._file_info_cache.move_to_end(key)
404        while len(self._file_info_cache) > FILE_INFO_CACHE_MAX:
405            self._file_info_cache.popitem(last=False)
406
407    def _file_info_cache_invalidate(self, track_id: str) -> None:
408        for k in [k for k in self._file_info_cache if k[0] == track_id]:
409            self._file_info_cache.pop(k, None)
410
411    async def _initial_sync_jitter(self, kind: str) -> None:
412        """
413        Sleep a small random delay during the first-sync window.
414
415        Smooths out the parallel metadata-refresh burst MA fires immediately
416        after a fresh install + auth, which is what triggers smart-captcha
417        in #146. After INITIAL_SYNC_WINDOW_S the helper is a no-op — no
418        steady-state overhead.
419
420        Only active for the `default` and `metadata` kinds. `file_info` is
421        on the streaming hot path (latency matters), and `rotor` has its
422        own bucket already tuned for its cadence.
423
424        :param kind: Throttler bucket name.
425        """
426        if kind not in ("default", "metadata"):
427            return
428        connected_at = self._connected_at
429        if connected_at is None:
430            return
431        if time.monotonic() - connected_at >= INITIAL_SYNC_WINDOW_S:
432            return
433        delay = random.uniform(0.0, INITIAL_SYNC_JITTER_S)
434        if delay > 0:
435            await asyncio.sleep(delay)
436
437    async def _call_with_retry(
438        self,
439        func: Callable[[ClientAsync], Awaitable[_T]],
440        *,
441        kind: str = "default",
442    ) -> _T:
443        """
444        Execute an async API call with throttling and one reconnect attempt on connection error.
445
446        Three layers of rate-control apply, outermost first:
447
448        * **Global concurrency cap** (restrictive mode only) — a
449          token-wide ``asyncio.Semaphore`` sized to ``RESTRICTIVE_GLOBAL_
450          CONCURRENCY``. Keeps total in-flight requests under Yandex's
451          per-token edge limit observed on datacenter / VPN IPs (~6).
452        * **Per-kind throttler** — a token bucket shared by all calls of a
453          given logical class (``default``, ``metadata``, ``file_info``,
454          ``rotor``). Caps sustained RPS per kind.
455        * **Per-endpoint lock** — derived from ``func.__qualname__`` so each
456          ``YandexMusicClient`` method gets its own ``asyncio.Lock``. Caps
457          concurrency to 1 per endpoint family — cheap defense-in-depth
458          against future ``asyncio.gather`` regressions; near-zero cost
459          when there's no contention.
460
461        :param func: Async callable that takes a ClientAsync and returns a result.
462        :param kind: Throttler bucket — one of the keys registered in
463            ``self._throttlers`` ("default", "metadata", "file_info",
464            "rotor"). Falls back to "default" if unknown.
465        :return: The result of the API call.
466        """
467        if self._global_concurrency is not None and not BYPASS_THROTTLER.get():
468            async with self._global_concurrency:
469                return await self._call_with_retry_inner(func, kind=kind)
470        return await self._call_with_retry_inner(func, kind=kind)
471
472    async def _call_with_retry_inner(
473        self,
474        func: Callable[[ClientAsync], Awaitable[_T]],
475        *,
476        kind: str,
477    ) -> _T:
478        """Per-kind throttler + per-endpoint lock layer of ``_call_with_retry``."""
479        # Per-request diagnostic — emits caller + kind so a DEBUG-level capture
480        # can reconstruct request density before any captcha trip. Stays at
481        # DEBUG so steady-state logs are clean.
482        if LOGGER.isEnabledFor(logging.DEBUG):
483            caller = getattr(func, "__qualname__", "?")
484            if ".<locals>." in caller:
485                caller = caller.split(".<locals>.")[0]
486            LOGGER.debug(
487                "req: kind=%s caller=%s bypass=%s",
488                kind,
489                caller,
490                BYPASS_THROTTLER.get(),
491            )
492        if not BYPASS_THROTTLER.get():
493            # Fast path: short-circuit before queueing if the kind is already
494            # blocked. Re-check after acquire() — another concurrent request
495            # may have engaged the cooldown while we were queued.
496            self._check_block(kind)
497            await self._initial_sync_jitter(kind)
498            await self._get_throttler(kind).acquire()
499            self._check_block(kind)
500        client = await self._ensure_connected()
501        endpoint = self._derive_endpoint(func)
502        try:
503            return await self._invoke_under_endpoint_lock(func, client, endpoint)
504        except Exception as err:
505            rate_limit_exc = self._maybe_handle_429(err, kind)
506            if rate_limit_exc is not None:
507                raise rate_limit_exc from NetworkError(self._truncate_err_msg(err))
508            if not self._is_connection_error(err):
509                raise
510            LOGGER.warning("Connection error, reconnecting and retrying: %s", err)
511            try:
512                await self._reconnect()
513            except Exception as recon_err:
514                raise ProviderUnavailableError("Reconnect failed") from recon_err
515            client = cast("ClientAsync", self._client)
516            # Re-check the block AND re-acquire a throttler token before the
517            # retry. Skipping ``acquire()`` here lets reconnect-retries
518            # bypass rate-limiting, doubling the effective request rate
519            # during connection flap — the conditions that already increase
520            # captcha-trip risk. BYPASS_THROTTLER paths skip this so an
521            # in-flight stream refresh can still attempt the retry.
522            if not BYPASS_THROTTLER.get():
523                self._check_block(kind)
524                await self._get_throttler(kind).acquire()
525                self._check_block(kind)
526            # Reconnect-retry must also go through 429 classification —
527            # otherwise a captcha on the retry attempt bypasses the cooldown
528            # logic and propagates the raw HTML body.
529            try:
530                return await self._invoke_under_endpoint_lock(func, client, endpoint)
531            except Exception as retry_err:
532                retry_exc = self._maybe_handle_429(retry_err, kind)
533                if retry_exc is not None:
534                    raise retry_exc from NetworkError(self._truncate_err_msg(retry_err))
535                raise
536
537    async def _invoke_under_endpoint_lock(
538        self,
539        func: Callable[[ClientAsync], Awaitable[_T]],
540        client: ClientAsync,
541        endpoint: str | None,
542    ) -> _T:
543        """Run ``func(client)`` serialised by the per-endpoint lock when set."""
544        if endpoint is None:
545            return await func(client)
546        async with self._get_endpoint_lock(endpoint):
547            return await func(client)
548
549    async def _call_no_retry(
550        self,
551        func: Callable[[ClientAsync], Awaitable[_T]],
552        *,
553        kind: str = "default",
554    ) -> _T:
555        """
556        Execute an async API call without reconnect retry on call failure.
557
558        Used for fire-and-forget calls (e.g. rotor feedback) where a failed request
559        should be silently dropped rather than triggering a reconnect cycle that
560        could cause rate limiting. Note: _ensure_connected() is still called to
561        establish the initial connection if needed; only the reconnect-on-error
562        path is skipped.
563
564        :param func: Async callable that takes a ClientAsync and returns a result.
565        :param kind: Throttler bucket — one of the keys registered in
566            ``self._throttlers`` ("default", "metadata", "file_info",
567            "rotor"). Falls back to "default" if unknown.
568        :return: The result of the API call.
569        """
570        if not BYPASS_THROTTLER.get():
571            # Same dual check as _call_with_retry — see comment there.
572            self._check_block(kind)
573            await self._initial_sync_jitter(kind)
574            await self._get_throttler(kind).acquire()
575            self._check_block(kind)
576        client = await self._ensure_connected()
577        try:
578            return await func(client)
579        except Exception as err:
580            # Even on the fire-and-forget path we want to classify 429s: a
581            # captcha hit on rotor feedback must still quarantine the rotor
582            # kind so the rest of the provider stops poking Yandex's edge
583            # while it's hot. Truncation also prevents callers' broad
584            # `except NetworkError` from logging multi-KB HTML payloads.
585            rate_limit_exc = self._maybe_handle_429(err, kind)
586            if rate_limit_exc is not None:
587                raise rate_limit_exc from NetworkError(self._truncate_err_msg(err))
588            raise
589
590    # Rotor (radio station) methods
591
592    async def get_rotor_station_tracks(
593        self,
594        station_id: str,
595        queue: str | int | None = None,
596    ) -> tuple[list[YandexTrack], str | None]:
597        """
598        Get tracks from a rotor station (e.g. user:onyourwave or track:1234).
599
600        :param station_id: Station ID (e.g. ROTOR_STATION_MY_WAVE or "track:1234" for similar).
601        :param queue: Optional track ID for pagination (first track of previous batch).
602        :return: Tuple of (list of track objects, batch_id for feedback or None).
603        """
604        try:
605            result = await self._call_with_retry(
606                lambda c: c.rotor_station_tracks(station_id, settings2=True, queue=queue),
607                kind="rotor",
608            )
609        except BadRequestError as err:
610            LOGGER.warning("Error fetching rotor station %s tracks: %s", station_id, err)
611            return ([], None)
612        except (NetworkError, ProviderUnavailableError) as err:
613            LOGGER.warning("Error fetching rotor station tracks: %s", err)
614            return ([], None)
615
616        if not result or not result.sequence:
617            return ([], result.batch_id if result else None)
618        track_ids = []
619        for seq in result.sequence:
620            if seq.track is None:
621                continue
622            tid = getattr(seq.track, "id", None) or getattr(seq.track, "track_id", None)
623            if tid is not None:
624                track_ids.append(str(tid))
625        if not track_ids:
626            return ([], result.batch_id if result else None)
627        try:
628            full_tracks = await self.get_tracks(track_ids)
629        except ResourceTemporarilyUnavailable as err:
630            LOGGER.warning("Error fetching rotor station track details: %s", err)
631            return ([], result.batch_id if result else None)
632        order_map = {str(t.id): t for t in full_tracks if hasattr(t, "id") and t.id}
633        ordered = [order_map[tid] for tid in track_ids if tid in order_map]
634        return (ordered, result.batch_id if result else None)
635
636    async def send_rotor_station_feedback(
637        self,
638        station_id: str,
639        feedback_type: str,
640        *,
641        batch_id: str | None = None,
642        track_id: str | None = None,
643        total_played_seconds: int | None = None,
644    ) -> bool:
645        """
646        Send rotor station feedback for My Wave recommendations.
647
648        Used to report radioStarted, trackStarted, trackFinished, skip so that
649        Yandex can improve subsequent recommendations.
650
651        :param station_id: Station ID (e.g. ROTOR_STATION_MY_WAVE).
652        :param feedback_type: One of 'radioStarted', 'trackStarted', 'trackFinished', 'skip'.
653        :param batch_id: Optional batch ID from the last get_my_wave_tracks response.
654        :param track_id: Track ID (required for trackStarted, trackFinished, skip).
655        :param total_played_seconds: Seconds played (for trackFinished, skip).
656        :return: True if the request succeeded.
657        """
658        timestamp = utc().isoformat().replace("+00:00", "Z")
659
660        async def _send(c: ClientAsync) -> bool:
661            if feedback_type == "radioStarted":
662                return bool(
663                    await c.rotor_station_feedback_radio_started(
664                        station_id,
665                        from_="YandexMusicDesktopAppWindows",
666                        batch_id=batch_id,
667                        timestamp=timestamp,
668                    )
669                )
670            if feedback_type == "trackStarted":
671                if track_id is None:
672                    return False
673                return bool(
674                    await c.rotor_station_feedback_track_started(
675                        station_id,
676                        track_id=track_id,
677                        batch_id=batch_id,
678                        timestamp=timestamp,
679                    )
680                )
681            if feedback_type == "trackFinished":
682                if track_id is None:
683                    return False
684                return bool(
685                    await c.rotor_station_feedback_track_finished(
686                        station_id,
687                        track_id=track_id,
688                        total_played_seconds=float(total_played_seconds or 0),
689                        batch_id=batch_id,
690                        timestamp=timestamp,
691                    )
692                )
693            if feedback_type == "skip":
694                if track_id is None:
695                    return False
696                return bool(
697                    await c.rotor_station_feedback_skip(
698                        station_id,
699                        track_id=track_id,
700                        total_played_seconds=float(total_played_seconds or 0),
701                        batch_id=batch_id,
702                        timestamp=timestamp,
703                    )
704                )
705            return bool(
706                await c.rotor_station_feedback(
707                    station_id,
708                    type_=feedback_type,
709                    timestamp=timestamp,
710                    track_id=track_id,
711                    total_played_seconds=total_played_seconds,
712                    batch_id=batch_id,
713                )
714            )
715
716        try:
717            result = await self._call_no_retry(_send, kind="rotor")
718            LOGGER.debug(
719                "Rotor feedback %s track_id=%s total_played_seconds=%s",
720                feedback_type,
721                track_id,
722                total_played_seconds,
723            )
724            return result
725        except BadRequestError as err:
726            LOGGER.warning("Rotor feedback %s failed: %s", feedback_type, err)
727            return False
728        except ResourceTemporarilyUnavailable as err:
729            # 429/captcha already truncated + block engaged inside _call_no_retry.
730            LOGGER.warning("Rotor feedback %s rate-limited: %s", feedback_type, err)
731            return False
732        except (NetworkError, ProviderUnavailableError) as err:
733            LOGGER.warning(
734                "Rotor feedback %s failed: %s",
735                feedback_type,
736                self._truncate_err_msg(err),
737            )
738            return False
739
740    # Rotor session API (new session-based endpoints)
741    #
742    # Yandex's newer rotor API models a wave as a long-lived session:
743    #   POST /rotor/session/new                     → {radioSessionId, sequence, batchId}
744    #   POST /rotor/session/{sessionId}/tracks      → {sequence, batchId}
745    #   POST /rotor/session/{sessionId}/feedback    → {result: "ok"}
746    # All feedback events carry the same sessionId, so we no longer need to
747    # thread per-batch batch_ids through call sites the way the stations-based
748    # API forced us to.
749
750    async def _rotor_session_request(
751        self, path: str, body: dict[str, Any], *, with_retry: bool = True
752    ) -> dict[str, Any] | None:
753        """
754        POST a JSON body to /rotor/session/{path} and return parsed result.
755
756        Reuses the MarshalX ClientAsync internal request object so we inherit
757        its auth headers and parsing. `json=` is forwarded to `aiohttp.request`
758        by MarshalX's `**kwargs` passthrough.
759
760        :param path: Path suffix after /rotor/session/ (e.g. "new",
761            "{session_id}/tracks", "{session_id}/feedback").
762        :param body: JSON body to send.
763        :param with_retry: When True (default), uses the same reconnect-on-
764            transient-connection-error path as normal data fetches —
765            appropriate for ``new`` and ``tracks`` which sit on the
766            user-facing browse/play path. Set to False for ``feedback``,
767            where a dropped request should be silently lost rather than
768            hammered against a potentially rate-limiting server.
769        :return: Parsed result dict, or None on failure.
770        """
771
772        async def _do(c: ClientAsync) -> dict[str, Any] | None:
773            base = getattr(c, "base_url", "https://api.music.yandex.net")
774            url = f"{base}/rotor/session/{path}"
775            LOGGER.debug("Rotor session POST %s body_keys=%s", path, list(body.keys()))
776            try:
777                result = await c._request.post(url, json=body)
778            except NetworkError as err:
779                # Let the outer retry wrapper see transient drops. On the
780                # no-retry path swallow ordinary network blips silently, but
781                # 429/captcha errors MUST propagate so _call_no_retry can
782                # engage the rotor cooldown — otherwise feedback keeps
783                # hammering Yandex during an active edge ban.
784                if with_retry or self._is_rate_limit_error(err):
785                    raise
786                LOGGER.debug("Rotor session POST %s: network error (no retry)", path)
787                return None
788            except BadRequestError as err:
789                # 4xx is terminal — server rejected the body; retry would only
790                # reproduce the same failure.
791                LOGGER.warning("Rotor session POST %s failed: %s", path, err)
792                return None
793            if isinstance(result, dict):
794                LOGGER.debug("Rotor session POST %s → result keys=%s", path, list(result.keys()))
795                return result
796            LOGGER.debug("Rotor session POST %s → non-dict result: %r", path, result)
797            return None
798
799        runner = self._call_with_retry if with_retry else self._call_no_retry
800        try:
801            return await runner(_do, kind="rotor")
802        except UnauthorizedError as err:
803            # Expired/invalidated token. Surface as LoginFailed so MA prompts
804            # for re-auth instead of the raw yandex_music exception bubbling
805            # through browse / play and crashing the caller.
806            LOGGER.warning("Rotor session POST %s: token no longer valid", path)
807            raise LoginFailed("Invalid Yandex Music token") from err
808        except ResourceTemporarilyUnavailable as err:
809            LOGGER.warning("Rotor session POST %s rate-limited: %s", path, err)
810            return None
811        except (NetworkError, ProviderUnavailableError) as err:
812            LOGGER.warning("Rotor session POST %s failed: %s", path, self._truncate_err_msg(err))
813            return None
814
815    async def rotor_session_new(
816        self,
817        station_id: str,
818        *,
819        settings: dict[str, str] | None = None,
820        queue: list[str] | None = None,
821    ) -> tuple[str | None, list[YandexTrack], str | None]:
822        """
823        Create a new rotor session.
824
825        Sends `includeWaveModel: true` so Yandex applies its wave ML model and
826        `interactive: true` so the session is treated as foreground user play.
827
828        :param station_id: Station ID (e.g. "user:onyourwave" or "track:123").
829        :param settings: Optional {diversity, moodEnergy, language} — each
830            becomes an additional seed like "settingDiversity:discover".
831        :param queue: Optional initial track IDs in the queue; usually empty.
832        :return: Tuple of (radio_session_id, list of tracks, batch_id).
833            Any element may be None/[] on failure.
834        """
835        seeds: list[str] = [station_id]
836        if settings:
837            for key, seed_name in (
838                ("diversity", "settingDiversity"),
839                ("moodEnergy", "settingMoodEnergy"),
840                ("language", "settingLanguage"),
841            ):
842                val = settings.get(key)
843                if val:
844                    seeds.append(f"{seed_name}:{val}")
845        body: dict[str, Any] = {
846            "seeds": seeds,
847            "queue": queue or [],
848            "includeTracksInResponse": True,
849            "includeWaveModel": True,
850            "interactive": True,
851        }
852        result = await self._rotor_session_request("new", body)
853        if not result:
854            return (None, [], None)
855        session_id = result.get("radioSessionId")
856        batch_id = result.get("batchId")
857        tracks = await self._hydrate_session_tracks(result.get("sequence") or [])
858        return (session_id, tracks, batch_id)
859
860    async def rotor_session_tracks(
861        self, session_id: str, *, current_track_id: str
862    ) -> tuple[list[YandexTrack], str | None]:
863        """
864        Fetch the next batch of tracks for an active rotor session.
865
866        :param session_id: radioSessionId from rotor_session_new().
867        :param current_track_id: Track ID just consumed from the previous batch
868            (Yandex uses it to decide what to return next).
869        :return: Tuple of (list of tracks, new batch_id).
870        """
871        body = {"queue": [str(current_track_id)]}
872        result = await self._rotor_session_request(f"{session_id}/tracks", body)
873        if not result:
874            return ([], None)
875        batch_id = result.get("batchId")
876        tracks = await self._hydrate_session_tracks(result.get("sequence") or [])
877        return (tracks, batch_id)
878
879    async def rotor_session_feedback(
880        self,
881        session_id: str,
882        event_type: str,
883        *,
884        track_id: str | None = None,
885        total_played_seconds: int | None = None,
886        batch_id: str | None = None,
887    ) -> bool:
888        """
889        Send a feedback event for an active rotor session.
890
891        Supports the Yandex rotor event types: radioStarted, trackStarted,
892        trackFinished, skip, like, dislike. For radioStarted the track_id goes
893        into `event.from`; all other types use `event.trackId`. Only
894        trackFinished and skip carry `totalPlayedSeconds`.
895
896        :param session_id: radioSessionId.
897        :param event_type: rotor event type string.
898        :param track_id: Yandex track ID the event refers to (required for
899            everything except radioStarted without a seed).
900        :param total_played_seconds: seconds of the track that were played
901            (only meaningful for trackFinished / skip).
902        :param batch_id: batchId from the most recent rotor_session_{new,tracks}
903            response; anchors the event to a specific batch.
904        :return: True if the POST succeeded.
905        """
906        timestamp = utc().isoformat().replace("+00:00", "Z")
907        event: dict[str, Any] = {"type": event_type, "timestamp": timestamp}
908        if event_type == "radioStarted":
909            if track_id is not None:
910                event["from"] = str(track_id)
911        elif track_id is not None:
912            event["trackId"] = str(track_id)
913        if event_type in ("trackFinished", "skip") and total_played_seconds is not None:
914            event["totalPlayedSeconds"] = int(total_played_seconds)
915        body: dict[str, Any] = {"event": event}
916        if batch_id:
917            body["batchId"] = batch_id
918        LOGGER.debug(
919            "Rotor session feedback: session=%s event=%s track=%s secs=%s batch=%s",
920            session_id,
921            event_type,
922            track_id,
923            total_played_seconds,
924            batch_id,
925        )
926        result = await self._rotor_session_request(f"{session_id}/feedback", body, with_retry=False)
927        return result is not None
928
929    async def _hydrate_session_tracks(self, sequence: list[dict[str, Any]]) -> list[YandexTrack]:
930        """
931        Extract track IDs from a rotor session sequence and hydrate via get_tracks.
932
933        The session endpoints return tracks inline when includeTracksInResponse
934        is true, but full track objects (with download info, covers, etc.) are
935        fetched separately so parsed Track objects have the same shape as in
936        the rest of the provider.
937
938        :param sequence: List of sequence items from a rotor session response.
939        :return: List of full track objects in the same order as `sequence`.
940        """
941        track_ids: list[str] = []
942        for seq in sequence:
943            tr = seq.get("track") if isinstance(seq, dict) else None
944            tid = None
945            if isinstance(tr, dict):
946                tid = tr.get("id") or tr.get("track_id")
947            if tid is not None:
948                track_ids.append(str(tid))
949        if not track_ids:
950            return []
951        try:
952            full_tracks = await self.get_tracks(track_ids)
953        except ResourceTemporarilyUnavailable as err:
954            LOGGER.warning("Rotor session track hydration failed: %s", err)
955            return []
956        order_map = {str(t.id): t for t in full_tracks if hasattr(t, "id") and t.id}
957        return [order_map[tid] for tid in track_ids if tid in order_map]
958
959    async def play_audio(
960        self,
961        *,
962        track_id: str,
963        album_id: str,
964        play_id: str,
965        track_length_seconds: int,
966        total_played_seconds: int,
967        end_position_seconds: int,
968        from_: str = "music_assistant-audiobook",
969    ) -> bool:
970        """
971        Report playback progress for an audiobook chapter or podcast episode.
972
973        Yandex persists this server-side so progress is visible across its
974        other clients. Failures are swallowed — progress sync is advisory and
975        must never abort pause/stop handling — so auth failures, rate-limits
976        and network blips all log at debug and return False.
977        """
978        try:
979            return bool(
980                await self._call_no_retry(
981                    lambda c: c.play_audio(
982                        track_id=track_id,
983                        album_id=album_id,
984                        from_=from_,
985                        play_id=play_id,
986                        track_length_seconds=track_length_seconds,
987                        total_played_seconds=total_played_seconds,
988                        end_position_seconds=end_position_seconds,
989                    )
990                )
991            )
992        except (
993            BadRequestError,
994            NetworkError,
995            ProviderUnavailableError,
996            UnauthorizedError,
997            LoginFailed,
998            ResourceTemporarilyUnavailable,
999        ) as err:
1000            LOGGER.debug("play_audio failed for %s: %s", track_id, err)
1001            return False
1002
1003    # Library methods
1004
1005    async def get_liked_tracks(self) -> list[TrackShort]:
1006        """
1007        Get user's liked tracks sorted by timestamp (most recent first).
1008
1009        :return: List of liked track objects sorted in reverse chronological order.
1010        """
1011        try:
1012            result = await self._call_with_retry(lambda c: c.users_likes_tracks())
1013            if result is None:
1014                return []
1015            tracks = result.tracks or []
1016            # Sort by timestamp in descending order (most recently liked first).
1017            # ``TrackShort.timestamp`` is sometimes tz-aware and sometimes
1018            # tz-naive depending on the upstream library version, so we
1019            # normalise to naive before comparing.
1020            return sorted(tracks, key=_liked_track_sort_key, reverse=True)
1021        except BadRequestError as err:
1022            # 4xx is terminal — do not signal retry. MA would otherwise loop.
1023            LOGGER.warning("Liked tracks unavailable (4xx): %s", err)
1024            return []
1025        except (NetworkError, ProviderUnavailableError) as err:
1026            LOGGER.warning("Error fetching liked tracks: %s", err)
1027            raise ResourceTemporarilyUnavailable("Failed to fetch liked tracks") from err
1028
1029    async def get_liked_albums(self, batch_size: int = 50) -> list[YandexAlbum]:
1030        """
1031        Get user's liked albums with full details (including cover art).
1032
1033        The users_likes_albums endpoint returns minimal album data without
1034        cover_uri, so we fetch full album details in batches afterwards.
1035
1036        :return: List of liked album objects with full details.
1037        """
1038        try:
1039            result = await self._call_with_retry(lambda c: c.users_likes_albums())
1040        except BadRequestError as err:
1041            LOGGER.warning("Liked albums unavailable (4xx): %s", err)
1042            return []
1043        except (NetworkError, ProviderUnavailableError) as err:
1044            LOGGER.warning("Error fetching liked albums: %s", err)
1045            raise ResourceTemporarilyUnavailable("Failed to fetch liked albums") from err
1046
1047        if result is None:
1048            return []
1049        album_ids = [
1050            str(like.album.id) for like in result if like.album is not None and like.album.id
1051        ]
1052        if not album_ids:
1053            return []
1054        # Fetch full album details in batches to get cover_uri and other metadata
1055        full_albums: list[YandexAlbum] = []
1056        for i in range(0, len(album_ids), batch_size):
1057            batch = album_ids[i : i + batch_size]
1058            try:
1059                batch_result = await self._call_with_retry(
1060                    lambda c, _b=batch: c.albums(_b)  # type: ignore[misc]
1061                )
1062                if batch_result:
1063                    full_albums.extend(batch_result)
1064            except (BadRequestError, NetworkError, ProviderUnavailableError) as batch_err:
1065                LOGGER.warning("Error fetching album details batch: %s", batch_err)
1066                # Fall back to minimal data for this batch
1067                batch_set = set(batch)
1068                for like in result:
1069                    if like.album is not None and like.album.id and str(like.album.id) in batch_set:
1070                        full_albums.append(like.album)
1071            # Spread bursts: small jittered pause before next batch.
1072            if i + batch_size < len(album_ids):
1073                await asyncio.sleep(
1074                    LIKED_BATCH_JITTER_MIN_S + random.random() * LIKED_BATCH_JITTER_SPAN_S
1075                )
1076        return full_albums
1077
1078    async def get_liked_artists(self) -> list[YandexArtist]:
1079        """
1080        Get user's liked artists.
1081
1082        :return: List of liked artist objects.
1083        """
1084        try:
1085            result = await self._call_with_retry(lambda c: c.users_likes_artists())
1086            if result is None:
1087                return []
1088            return [like.artist for like in result if like.artist is not None]
1089        except BadRequestError as err:
1090            LOGGER.error("Error fetching liked artists: %s", err)
1091            raise ResourceTemporarilyUnavailable("Failed to fetch liked artists") from err
1092        except (NetworkError, ProviderUnavailableError) as err:
1093            LOGGER.error("Error fetching liked artists: %s", err)
1094            raise ResourceTemporarilyUnavailable("Failed to fetch liked artists") from err
1095
1096    async def get_user_playlists(self) -> list[YandexPlaylist]:
1097        """
1098        Get user's playlists.
1099
1100        :return: List of playlist objects.
1101        """
1102        try:
1103            result = await self._call_with_retry(lambda c: c.users_playlists_list())
1104            if result is None:
1105                return []
1106            return list(result)
1107        except BadRequestError as err:
1108            LOGGER.error("Error fetching playlists: %s", err)
1109            raise ResourceTemporarilyUnavailable("Failed to fetch playlists") from err
1110        except (NetworkError, ProviderUnavailableError) as err:
1111            LOGGER.error("Error fetching playlists: %s", err)
1112            raise ResourceTemporarilyUnavailable("Failed to fetch playlists") from err
1113
1114    async def get_liked_playlists(self) -> list[YandexPlaylist]:
1115        """
1116        Get user's liked/saved editorial playlists.
1117
1118        :return: List of liked playlist objects.
1119        """
1120        try:
1121            result = await self._call_with_retry(lambda c: c.users_likes_playlists())
1122            if result is None:
1123                return []
1124            playlists = []
1125            for like in result:
1126                if like.playlist is not None:
1127                    playlists.append(like.playlist)
1128            return playlists
1129        except BadRequestError as err:
1130            LOGGER.error("Error fetching liked playlists: %s", err)
1131            raise ResourceTemporarilyUnavailable("Failed to fetch liked playlists") from err
1132        except (NetworkError, ProviderUnavailableError) as err:
1133            LOGGER.error("Error fetching liked playlists: %s", err)
1134            raise ResourceTemporarilyUnavailable("Failed to fetch liked playlists") from err
1135
1136    # Search
1137
1138    async def search(
1139        self,
1140        query: str,
1141        search_type: str = "all",
1142    ) -> Search | None:
1143        """
1144        Search for tracks, albums, artists, or playlists.
1145
1146        The upstream ``yandex-music`` client does not accept a per-type result
1147        cap at this layer — callers slice the parsed buckets to whatever
1148        ``limit`` they need after classification.
1149
1150        :param query: Search query string.
1151        :param search_type: Type of search ('all', 'track', 'album', 'artist', 'playlist').
1152        :return: Search results object.
1153        """
1154        try:
1155            return await self._call_with_retry(
1156                lambda c: c.search(query, type_=search_type, page=0, nocorrect=False)
1157            )
1158        except BadRequestError as err:
1159            # 4xx is terminal (malformed query, geo-block) — return None so MA
1160            # surfaces "no results" instead of retrying the same failure.
1161            LOGGER.warning("Search rejected by Yandex (4xx): %s", err)
1162            return None
1163        except (NetworkError, ProviderUnavailableError) as err:
1164            LOGGER.warning("Search error: %s", err)
1165            raise ResourceTemporarilyUnavailable("Search failed") from err
1166
1167    # Get single items
1168
1169    async def get_track(self, track_id: str) -> YandexTrack | None:
1170        """
1171        Get a single track by ID.
1172
1173        :param track_id: Track ID.
1174        :return: Track object or None if not found.
1175        """
1176        try:
1177            tracks = await self._call_with_retry(lambda c: c.tracks([track_id]))
1178            return tracks[0] if tracks else None
1179        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1180            LOGGER.error("Error fetching track %s: %s", track_id, err)
1181            return None
1182
1183    async def get_track_lyrics(self, track_id: str) -> tuple[str | None, bool]:
1184        """
1185        Get lyrics for a track.
1186
1187        Fetches lyrics from Yandex Music API. Returns the lyrics text and whether
1188        it's in synced LRC format (with timestamps) or plain text.
1189
1190        Note: This method fetches the track first to check lyrics_available. If you
1191        already have the YandexTrack object, use get_track_lyrics_from_track() to
1192        avoid a redundant API call.
1193
1194        :param track_id: Track ID.
1195        :return: Tuple of (lyrics_text, is_synced). Returns (None, False) if unavailable.
1196        """
1197        try:
1198            tracks = await self._call_with_retry(lambda c: c.tracks([track_id]))
1199            if not tracks:
1200                return None, False
1201
1202            return await self.get_track_lyrics_from_track(tracks[0])
1203
1204        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1205            LOGGER.debug("Error fetching lyrics for track %s: %s", track_id, err)
1206            return None, False
1207        except Exception as err:
1208            # Catch any other errors (e.g., geo-restrictions, API changes)
1209            LOGGER.debug("Unexpected error fetching lyrics for track %s: %s", track_id, err)
1210            return None, False
1211
1212    async def get_track_lyrics_from_track(self, track: YandexTrack) -> tuple[str | None, bool]:
1213        """
1214        Get lyrics for an already-fetched track.
1215
1216        Avoids the extra tracks([track_id]) API call when the YandexTrack object
1217        is already available.
1218
1219        :param track: YandexTrack object (already fetched).
1220        :return: Tuple of (lyrics_text, is_synced). Returns (None, False) if unavailable.
1221        """
1222        track_id = getattr(track, "id", None) or getattr(track, "track_id", "unknown")
1223        try:
1224            if not getattr(track, "lyrics_available", False):
1225                LOGGER.debug("Lyrics not available for track %s", track_id)
1226                return None, False
1227
1228            track_lyrics = await track.get_lyrics_async()
1229            if not track_lyrics:
1230                LOGGER.debug("Failed to get lyrics metadata for track %s", track_id)
1231                return None, False
1232
1233            lyrics_text = await track_lyrics.fetch_lyrics_async()
1234            if not lyrics_text:
1235                return None, False
1236
1237            # Check if it's LRC format (synced lyrics have timestamps like [00:12.34])
1238            # Use re.search without ^ so metadata lines like [ar:Artist] don't prevent detection
1239            is_synced = bool(re.search(r"\[\d{2}:\d{2}(?:\.\d{2,3})?\]", lyrics_text))
1240            return lyrics_text, is_synced
1241
1242        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1243            LOGGER.debug("Error fetching lyrics for track %s: %s", track_id, err)
1244            return None, False
1245        except Exception as err:
1246            # Catch any other errors (e.g., geo-restrictions, API changes)
1247            LOGGER.debug("Unexpected error fetching lyrics for track %s: %s", track_id, err)
1248            return None, False
1249
1250    async def get_tracks(self, track_ids: list[str]) -> list[YandexTrack]:
1251        """
1252        Get multiple tracks by IDs.
1253
1254        :param track_ids: List of track IDs.
1255        :return: List of track objects.
1256        :raises ResourceTemporarilyUnavailable: On network errors after retry.
1257        """
1258        try:
1259            result = await self._call_with_retry(lambda c: c.tracks(track_ids))
1260            return result or []
1261        except BadRequestError as err:
1262            LOGGER.error("Error fetching tracks: %s", err)
1263            return []
1264        except (NetworkError, ProviderUnavailableError) as err:
1265            LOGGER.error("Error fetching tracks (retry failed): %s", err)
1266            raise ResourceTemporarilyUnavailable("Failed to fetch tracks") from err
1267
1268    async def get_album(self, album_id: str) -> YandexAlbum | None:
1269        """
1270        Get a single album by ID.
1271
1272        :param album_id: Album ID.
1273        :return: Album object or None if not found.
1274        """
1275        try:
1276            albums = await self._call_with_retry(lambda c: c.albums([album_id]), kind="metadata")
1277            return albums[0] if albums else None
1278        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1279            LOGGER.error("Error fetching album %s: %s", album_id, err)
1280            return None
1281
1282    async def get_album_with_tracks(self, album_id: str) -> YandexAlbum | None:
1283        """
1284        Get an album with its tracks.
1285
1286        Uses the same semantics as the web client: albums/{id}/with-tracks
1287        with resumeStream, richTracks, withListeningFinished.
1288
1289        :param album_id: Album ID.
1290        :return: Album object with tracks or None if not found.
1291        """
1292        try:
1293            return await self._call_with_retry(
1294                lambda c: c.albums_with_tracks(
1295                    album_id,
1296                    params={
1297                        "resumeStream": "true",
1298                        "richTracks": "true",
1299                        "withListeningFinished": "true",
1300                    },
1301                ),
1302                kind="metadata",
1303            )
1304        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1305            LOGGER.error("Error fetching album with tracks %s: %s", album_id, err)
1306            return None
1307
1308    async def get_artist(self, artist_id: str) -> YandexArtist | None:
1309        """
1310        Get a single artist by ID.
1311
1312        :param artist_id: Artist ID.
1313        :return: Artist object or None if not found.
1314        """
1315        try:
1316            artists = await self._call_with_retry(lambda c: c.artists([artist_id]), kind="metadata")
1317            return artists[0] if artists else None
1318        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1319            LOGGER.error("Error fetching artist %s: %s", artist_id, err)
1320            return None
1321
1322    async def get_artist_albums(
1323        self, artist_id: str, limit: int = DEFAULT_LIMIT
1324    ) -> list[YandexAlbum]:
1325        """
1326        Get artist's albums.
1327
1328        :param artist_id: Artist ID.
1329        :param limit: Maximum number of albums.
1330        :return: List of album objects.
1331        """
1332        try:
1333            result = await self._call_with_retry(
1334                lambda c: c.artists_direct_albums(artist_id, page=0, page_size=limit),
1335                kind="metadata",
1336            )
1337            if result is None:
1338                return []
1339            return result.albums or []
1340        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1341            LOGGER.error("Error fetching artist albums %s: %s", artist_id, err)
1342            return []
1343
1344    async def get_pins(self) -> Any | None:
1345        """
1346        Get the user's pinned items (artists/albums/playlists/waves).
1347
1348        :return: PinsList object or None on error.
1349        """
1350        try:
1351            return await self._call_with_retry(lambda c: c.pins())
1352        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1353            LOGGER.error("Error fetching pins: %s", err)
1354            return None
1355
1356    async def get_music_history(self) -> Any | None:
1357        """
1358        Get the user's listening history (grouped by day).
1359
1360        :return: MusicHistory object or None on error.
1361        """
1362        try:
1363            return await self._call_with_retry(lambda c: c.music_history())
1364        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1365            LOGGER.error("Error fetching music history: %s", err)
1366            return None
1367
1368    async def get_artist_about(self, artist_id: str) -> Any | None:
1369        """
1370        Get artist enrichment info: description, monthly listeners, links.
1371
1372        :param artist_id: Artist ID.
1373        :return: ArtistAbout object or None on error/missing.
1374        """
1375        try:
1376            return await self._call_with_retry(
1377                lambda c: c.artists_about(artist_id), kind="metadata"
1378            )
1379        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1380            LOGGER.error("Error fetching artist about %s: %s", artist_id, err)
1381            return None
1382
1383    async def get_similar_artists(
1384        self, artist_id: str, limit: int = DEFAULT_LIMIT
1385    ) -> list[YandexArtist]:
1386        """
1387        Get artists similar to the given one.
1388
1389        :param artist_id: Artist ID.
1390        :param limit: Maximum number of artists.
1391        :return: List of similar artist objects.
1392        """
1393        try:
1394            result = await self._call_with_retry(lambda c: c.artists_similar(artist_id))
1395            if result is None or not result.similar_artists:
1396                return []
1397            similar: list[YandexArtist] = result.similar_artists
1398            return similar[:limit]
1399        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1400            LOGGER.error("Error fetching similar artists %s: %s", artist_id, err)
1401            return []
1402
1403    async def get_artist_tracks(
1404        self, artist_id: str, limit: int = DEFAULT_LIMIT
1405    ) -> list[YandexTrack]:
1406        """
1407        Get artist's top tracks.
1408
1409        :param artist_id: Artist ID.
1410        :param limit: Maximum number of tracks.
1411        :return: List of track objects.
1412        """
1413        try:
1414            result = await self._call_with_retry(
1415                lambda c: c.artists_tracks(artist_id, page=0, page_size=limit),
1416                kind="metadata",
1417            )
1418            if result is None:
1419                return []
1420            return result.tracks or []
1421        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1422            LOGGER.error("Error fetching artist tracks %s: %s", artist_id, err)
1423            return []
1424
1425    async def get_playlist(self, user_id: str, playlist_id: str) -> YandexPlaylist | None:
1426        """
1427        Get a playlist by ID.
1428
1429        :param user_id: User ID (owner of the playlist).
1430        :param playlist_id: Playlist ID (kind).
1431        :return: Playlist object or None if not found.
1432        :raises ResourceTemporarilyUnavailable: On network errors.
1433        """
1434        try:
1435            result = await self._call_with_retry(
1436                lambda c: c.users_playlists(kind=int(playlist_id), user_id=user_id)
1437            )
1438            if isinstance(result, list):
1439                return result[0] if result else None
1440            return result
1441        except BadRequestError as err:
1442            LOGGER.error("Error fetching playlist %s/%s: %s", user_id, playlist_id, err)
1443            return None
1444        except (NetworkError, ProviderUnavailableError) as err:
1445            LOGGER.warning("Network error fetching playlist %s/%s: %s", user_id, playlist_id, err)
1446            raise ResourceTemporarilyUnavailable("Failed to fetch playlist") from err
1447
1448    # Streaming
1449
1450    async def get_track_download_info(
1451        self, track_id: str, get_direct_links: bool = True
1452    ) -> list[DownloadInfo]:
1453        """
1454        Get download info for a track.
1455
1456        :param track_id: Track ID.
1457        :param get_direct_links: Whether to get direct download links.
1458        :return: List of download info objects.
1459        """
1460        try:
1461            result = await self._call_with_retry(
1462                lambda c: c.tracks_download_info(track_id, get_direct_links=get_direct_links)
1463            )
1464            return result or []
1465        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1466            LOGGER.error("Error fetching download info for track %s: %s", track_id, err)
1467            return []
1468
1469    async def get_track_file_info(  # noqa: PLR0915
1470        self,
1471        track_id: str,
1472        quality: str = "lossless",
1473        codecs: str = GET_FILE_INFO_CODECS,
1474        transport: str = "raw",
1475    ) -> dict[str, Any] | None:
1476        """
1477        Request stream via get-file-info for any quality tier.
1478
1479        The /get-file-info endpoint supports all quality tiers (lossless, nq, lq)
1480        and returns the best available codec based on the codecs parameter order.
1481
1482        With transport="raw", returns a direct unencrypted URL.
1483        With transport="encraw", returns an AES-CTR encrypted URL with decryption key.
1484
1485        Uses _call_with_retry for automatic reconnection on transient failures.
1486
1487        :param track_id: Track ID.
1488        :param quality: Quality tier ("lossless", "nq", "lq").
1489        :param codecs: Comma-separated codec preference list.
1490        :param transport: Transport mode ("raw" or "encraw").
1491        :return: Parsed downloadInfo dict (url, codec, key?, ...) or None on error.
1492        """
1493        # Normalize codecs: strip whitespace from each token to prevent HMAC mismatches
1494        codecs = ",".join(c.strip() for c in codecs.split(",") if c.strip())
1495
1496        # Short-TTL cache to absorb repeat calls from MA's streaming retry loop.
1497        # Bypass when refresh is in progress (BYPASS_THROTTLER): a refresh fires
1498        # specifically because the previous URL expired on the CDN side, so the
1499        # cached entry is useless.
1500        # Include `codecs` in the key: the server may pick a different codec
1501        # (and URL) based on the codec preference order, so two calls with the
1502        # same (track, quality, transport) but different codec lists must not
1503        # share a cache slot.
1504        cache_key = (track_id, quality, codecs, transport)
1505        if not BYPASS_THROTTLER.get():
1506            # Check the file_info circuit-breaker BEFORE the cache lookup —
1507            # otherwise a cooldown-period caller could be served a stale URL
1508            # from before the block was engaged. Fail fast (return None) so
1509            # MA's streaming layer treats the track as unavailable.
1510            try:
1511                self._check_block("file_info")
1512            except ResourceTemporarilyUnavailable as err:
1513                LOGGER.debug(
1514                    "get-file-info for track %s: file_info cooldown active (%s)",
1515                    track_id,
1516                    err,
1517                )
1518                return None
1519            cached = self._file_info_cache_get(cache_key)
1520            if cached is not None:
1521                LOGGER.debug(
1522                    "get-file-info for track %s: cache hit (transport=%s)",
1523                    track_id,
1524                    transport,
1525                )
1526                return cached
1527
1528        def _build_signed_params(client: ClientAsync) -> tuple[str, dict[str, Any]]:
1529            """
1530            Build URL and signed params using current client and timestamp.
1531
1532            Called on each attempt by _call_with_retry, so the HMAC signature
1533            is recomputed with a fresh timestamp on every retry.
1534            """
1535            timestamp = int(time.time())
1536            params = {
1537                "ts": timestamp,
1538                "trackId": track_id,
1539                "quality": quality,
1540                "codecs": codecs,
1541                "transports": transport,
1542            }
1543            # Build sign string: ts + trackId + quality + codecs (commas stripped) + transports.
1544            codecs_for_sign = codecs.replace(",", "")
1545            param_string = f"{timestamp}{track_id}{quality}{codecs_for_sign}{transport}"
1546            hmac_sign = hmac.new(
1547                DEFAULT_SIGN_KEY.encode(),
1548                param_string.encode(),
1549                hashlib.sha256,
1550            )
1551            # SHA-256 (32 bytes) -> base64 = 44 chars with "=" padding.
1552            # Yandex API expects exactly 43 chars (one "=" removed).
1553            params["sign"] = base64.b64encode(hmac_sign.digest()).decode()[:-1]
1554            url = f"{client.base_url}/get-file-info"
1555            return url, params
1556
1557        def _parse_file_info_result(raw: dict[str, Any] | None) -> dict[str, Any] | None:
1558            if not raw or not isinstance(raw, dict):
1559                return None
1560            # yandex-music v3 no longer normalises camelCase keys inside
1561            # Response.result, so /get-file-info returns "downloadInfo" as-is.
1562            download_info = raw.get("download_info") or raw.get("downloadInfo")
1563            if not download_info or not download_info.get("url"):
1564                return None
1565
1566            result = cast("dict[str, Any]", download_info)
1567
1568            if "key" in download_info:
1569                result["needs_decryption"] = True
1570                LOGGER.debug(
1571                    "Encrypted URL received for track %s, will require decryption",
1572                    track_id,
1573                )
1574            else:
1575                result["needs_decryption"] = False
1576
1577            return result
1578
1579        async def _do_request(c: ClientAsync) -> dict[str, Any] | None:
1580            url, params = _build_signed_params(c)
1581            return await c._request.get(url, params=params)  # type: ignore[no-any-return]
1582
1583        try:
1584            result = await self._call_with_retry(_do_request, kind="file_info")
1585            parsed = _parse_file_info_result(result)
1586            if parsed:
1587                LOGGER.debug(
1588                    "get-file-info for track %s: Success, codec=%s, transport=%s",
1589                    track_id,
1590                    parsed.get("codec"),
1591                    transport,
1592                )
1593                # Always store the freshest URL — including under BYPASS_THROTTLER.
1594                # A successful refresh proves the previously cached entry was
1595                # stale, so replacing it avoids serving the old URL to the next
1596                # non-bypass caller until its TTL expires.
1597                self._file_info_cache_put(cache_key, parsed)
1598                return parsed
1599        except BadRequestError as err:
1600            # 4xx is terminal for this URL/quality. Drop any cached entry so we
1601            # don't replay a now-rejected response.
1602            self._file_info_cache_invalidate(track_id)
1603            LOGGER.debug(
1604                "get-file-info for track %s: BadRequestError %s",
1605                track_id,
1606                getattr(err, "message", str(err)) or repr(err),
1607            )
1608        except (
1609            NetworkError,
1610            ProviderUnavailableError,
1611            ResourceTemporarilyUnavailable,
1612        ) as err:
1613            LOGGER.debug(
1614                "get-file-info for track %s: %s %s",
1615                track_id,
1616                type(err).__name__,
1617                getattr(err, "message", str(err)) or repr(err),
1618            )
1619        except UnauthorizedError as err:
1620            # Auth expired — invalidate any cached URL so the post-re-auth call
1621            # doesn't replay a stale entry tied to the old session.
1622            self._file_info_cache_invalidate(track_id)
1623            LOGGER.debug(
1624                "get-file-info for track %s: UnauthorizedError %s",
1625                track_id,
1626                getattr(err, "message", str(err)) or repr(err),
1627            )
1628        except asyncio.CancelledError:
1629            raise
1630        except Exception as err:
1631            LOGGER.warning(
1632                "get-file-info for track %s: Unexpected %s: %s",
1633                track_id,
1634                type(err).__name__,
1635                err,
1636            )
1637
1638        return None
1639
1640    # Discovery / recommendations
1641
1642    async def get_feed(self) -> Feed | None:
1643        """
1644        Get personalized feed with generated playlists (Playlist of the Day, etc.).
1645
1646        :return: Feed object with generated_playlists, or None on error.
1647        """
1648        try:
1649            return await self._call_with_retry(lambda c: c.feed())
1650        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1651            LOGGER.debug("Error fetching feed: %s", err)
1652            return None
1653
1654    async def get_chart(self, chart_option: str = "") -> ChartInfo | None:
1655        """
1656        Get chart data.
1657
1658        :param chart_option: Optional chart variant (e.g. 'world', 'russia').
1659        :return: ChartInfo object or None on error.
1660        """
1661        try:
1662            return await self._call_with_retry(lambda c: c.chart(chart_option))
1663        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1664            LOGGER.debug("Error fetching chart: %s", err)
1665            return None
1666
1667    async def get_new_releases(self) -> LandingList | None:
1668        """
1669        Get new album releases.
1670
1671        :return: LandingList with new_releases (list of album IDs) or None on error.
1672        """
1673        try:
1674            return await self._call_with_retry(lambda c: c.new_releases())
1675        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1676            LOGGER.debug("Error fetching new releases: %s", err)
1677            return None
1678
1679    async def get_new_playlists(self) -> LandingList | None:
1680        """
1681        Get new editorial playlists.
1682
1683        :return: LandingList with new_playlists (list of PlaylistId) or None on error.
1684        """
1685        try:
1686            return await self._call_with_retry(lambda c: c.new_playlists())
1687        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1688            LOGGER.debug("Error fetching new playlists: %s", err)
1689            return None
1690
1691    async def get_albums(self, album_ids: list[str]) -> list[YandexAlbum]:
1692        """
1693        Get multiple albums by IDs.
1694
1695        :param album_ids: List of album IDs.
1696        :return: List of album objects.
1697        """
1698        try:
1699            result = await self._call_with_retry(lambda c: c.albums(album_ids))
1700            return result or []
1701        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1702            LOGGER.debug("Error fetching albums: %s", err)
1703            return []
1704
1705    async def get_playlists(self, playlist_ids: list[str]) -> list[YandexPlaylist]:
1706        """
1707        Get multiple playlists by IDs (format: 'uid:kind').
1708
1709        :param playlist_ids: List of playlist IDs in 'uid:kind' format.
1710        :return: List of playlist objects.
1711        """
1712        try:
1713            result = await self._call_with_retry(lambda c: c.playlists_list(playlist_ids))
1714            return result or []
1715        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1716            LOGGER.debug("Error fetching playlists: %s", err)
1717            return []
1718
1719    async def get_tag_playlists(self, tag_id: str) -> list[YandexPlaylist]:
1720        """
1721        Get playlists for a specific tag (mood, era, activity, genre, etc.).
1722
1723        Tags are used for curated collections like 'chill', '80s', 'workout', 'rock', etc.
1724        The API returns playlist IDs which are then fetched in full.
1725
1726        :param tag_id: Tag identifier (e.g. 'chill', '80s', 'workout', 'rock').
1727        :return: List of playlist objects with full details.
1728        """
1729        try:
1730            tag_result = await self._call_with_retry(lambda c: c.tags(tag_id))
1731            if not tag_result or not tag_result.ids:
1732                LOGGER.debug("No playlists found for tag: %s", tag_id)
1733                return []
1734
1735            # Convert PlaylistId objects to 'uid:kind' format
1736            playlist_ids = [f"{pid.uid}:{pid.kind}" for pid in tag_result.ids]
1737
1738            # Fetch full playlist details
1739            return await self.get_playlists(playlist_ids)
1740        except BadRequestError as err:
1741            LOGGER.debug("Tag %s not found: %s", tag_id, err)
1742            return []
1743        except (NetworkError, ProviderUnavailableError) as err:
1744            LOGGER.debug("Error fetching tag %s playlists: %s", tag_id, err)
1745            return []
1746
1747    async def get_landing_tags(self) -> list[tuple[str, str]]:
1748        """
1749        Discover available tag slugs from the landing mixes block.
1750
1751        Uses the landing("mixes") API which returns MixLink entities
1752        containing tag URLs (e.g., /tag/chill/) and display titles.
1753        Filters out editorial post entries (/post/ URLs) which have no playlists.
1754
1755        :return: List of (tag_slug, title) tuples for real tag entries only.
1756        """
1757        try:
1758            landing: Landing | None = await self._call_with_retry(lambda c: c.landing("mixes"))
1759            if not landing or not landing.blocks:
1760                return []
1761        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1762            LOGGER.debug("Error fetching landing tags: %s", err)
1763            return []
1764
1765        tags: list[tuple[str, str]] = []
1766        for block in landing.blocks:
1767            if not block.entities:
1768                continue
1769            for entity in block.entities:
1770                if entity.type == "mix-link" and isinstance(entity.data, MixLink):
1771                    url = entity.data.url  # e.g., "/tag/chill/" or "/post/..."
1772                    # Filter out editorial posts — only include /tag/ URLs
1773                    if not url.startswith("/tag/"):
1774                        continue
1775                    slug = url.strip("/").split("/")[-1]
1776                    if slug:
1777                        tags.append((slug, entity.data.title))
1778        return tags
1779
1780    async def get_mixes_waves(self) -> list[dict[str, Any]] | None:
1781        """
1782        Get AI Wave Set stations from /landing-blocks/mixes-waves endpoint.
1783
1784        Returns structured mix data with categories and station items, each
1785        containing station_id, title, seeds, and visual metadata.
1786
1787        :return: List of mix category dicts, or None on error.
1788        """
1789        return await self._get_landing_waves("mixes-waves")
1790
1791    async def get_waves_landing(self) -> list[dict[str, Any]] | None:
1792        """
1793        Get featured wave stations from /landing-blocks/waves endpoint.
1794
1795        Returns Yandex-curated wave categories with station items — the "Волны"
1796        landing page content, separate from the full rotor/stations/list and from
1797        the AI mixes-waves sets.
1798
1799        :return: List of wave category dicts, or None on error.
1800        """
1801        return await self._get_landing_waves("waves")
1802
1803    async def _get_landing_waves(self, block: str) -> list[dict[str, Any]] | None:
1804        """
1805        Fetch wave categories from a /landing-blocks/<block> endpoint.
1806
1807        Note: Response keys are auto-converted from camelCase to snake_case
1808        by the yandex-music library's JSON parser.
1809
1810        :param block: Block name, e.g. 'waves' or 'mixes-waves'.
1811        :return: List of wave category dicts, or None on error.
1812        """
1813
1814        async def _get(c: ClientAsync) -> dict[str, Any]:
1815            # ``base_url`` is not part of the public ``ClientAsync`` contract;
1816            # mirror ``_rotor_session_request`` and fall back defensively so a
1817            # library rename does not crash this endpoint with AttributeError.
1818            base = getattr(c, "base_url", "https://api.music.yandex.net")
1819            url = f"{base}/landing-blocks/{block}"
1820            return await c._request.get(url)  # type: ignore[no-any-return]
1821
1822        try:
1823            result = await self._call_with_retry(_get)
1824            if result and isinstance(result, dict):
1825                waves = result.get("waves", [])
1826                LOGGER.debug(
1827                    "landing-blocks/%s returned %d categories",
1828                    block,
1829                    len(waves) if isinstance(waves, list) else -1,
1830                )
1831                return waves if isinstance(waves, list) else []
1832            return None
1833        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1834            LOGGER.debug("Error fetching landing-blocks/%s: %s", block, err)
1835            return None
1836
1837    async def get_wave_stations(
1838        self, language: str | None = None
1839    ) -> list[tuple[str, str, str, str | None]]:
1840        """
1841        Get available rotor wave stations grouped by category.
1842
1843        Calls rotor_stations_list() — equivalent to the rotor/stations/list API endpoint.
1844        Filters out personal stations (type 'user') since My Wave is handled separately.
1845
1846        :param language: Language for station names (e.g. 'ru', 'en'). Defaults to API default.
1847        :return: List of (station_id, category, name, image_url) tuples,
1848                 e.g. ('genre:rock', 'genre', 'Рок', 'https://...').
1849        """
1850        try:
1851            results: list[StationResult] = await self._call_with_retry(
1852                lambda c: c.rotor_stations_list(language),
1853                kind="rotor",
1854            )
1855        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1856            LOGGER.warning("Error fetching wave stations: %s", err)
1857            return []
1858
1859        stations: list[tuple[str, str, str, str | None]] = []
1860        for result in results or []:
1861            station = result.station
1862            if station is None or station.id is None:
1863                continue
1864            category = station.id.type
1865            tag = station.id.tag
1866            if not category or not tag:
1867                continue
1868            if category in ("user", "local-language"):
1869                # Skip personal stations (My Wave is handled separately)
1870                # and local-language stations (Yandex returns overlapping tracks across them)
1871                continue
1872            station_id = f"{category}:{tag}"
1873            name = station.name or result.rup_title or tag
1874            image_url: str | None = None
1875            raw_url = station.full_image_url or (station.icon.image_url if station.icon else None)
1876            if raw_url:
1877                # Yandex avatar URIs use '%%' as a size placeholder; replace it with
1878                # the desired size. If no placeholder, append the size as a suffix
1879                # since these URLs return HTTP 400 without a size component.
1880                if not raw_url.startswith("http"):
1881                    raw_url = f"https://{raw_url}"
1882                if "%%" in raw_url:
1883                    image_url = raw_url.replace("%%", "400x400")
1884                else:
1885                    image_url = f"{raw_url}/400x400"
1886            stations.append((station_id, category, name, image_url))
1887        return stations
1888
1889    async def get_dashboard_stations(self) -> list[tuple[str, str, str | None]]:
1890        """
1891        Get personalized recommended stations for the current user.
1892
1893        Calls rotor_stations_dashboard() — returns user-specific stations based
1894        on listening history, unlike rotor_stations_list() which is non-personalized.
1895
1896        :return: List of (station_id, name, image_url) tuples,
1897                 e.g. ('genre:rock', 'Рок', 'https://...').
1898        """
1899        try:
1900            dashboard: Dashboard | None = await self._call_with_retry(
1901                lambda c: c.rotor_stations_dashboard(),
1902                kind="rotor",
1903            )
1904        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1905            LOGGER.warning("Error fetching dashboard stations: %s", err)
1906            return []
1907
1908        if not dashboard or not dashboard.stations:
1909            return []
1910
1911        stations: list[tuple[str, str, str | None]] = []
1912        for result in dashboard.stations:
1913            station = result.station
1914            if station is None or station.id is None:
1915                continue
1916            category = station.id.type
1917            tag = station.id.tag
1918            if not category or not tag:
1919                continue
1920            if category == "user":
1921                continue
1922            station_id = f"{category}:{tag}"
1923            name = station.name or result.rup_title or tag
1924            image_url: str | None = None
1925            raw_url = station.full_image_url or (station.icon.image_url if station.icon else None)
1926            if raw_url:
1927                if not raw_url.startswith("http"):
1928                    raw_url = f"https://{raw_url}"
1929                if "%%" in raw_url:
1930                    image_url = raw_url.replace("%%", "400x400")
1931                else:
1932                    image_url = f"{raw_url}/400x400"
1933            stations.append((station_id, name, image_url))
1934        return stations
1935
1936    # Library modifications
1937
1938    async def like_track(self, track_id: str) -> bool:
1939        """
1940        Add a track to liked tracks.
1941
1942        :param track_id: Track ID to like.
1943        :return: True if successful.
1944        """
1945        try:
1946            result = await self._call_with_retry(lambda c: c.users_likes_tracks_add(track_id))
1947            return result is not None
1948        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1949            LOGGER.error("Error liking track %s: %s", track_id, err)
1950            return False
1951
1952    async def unlike_track(self, track_id: str) -> bool:
1953        """
1954        Remove a track from liked tracks.
1955
1956        :param track_id: Track ID to unlike.
1957        :return: True if successful.
1958        """
1959        try:
1960            result = await self._call_with_retry(lambda c: c.users_likes_tracks_remove(track_id))
1961            return result is not None
1962        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1963            LOGGER.error("Error unliking track %s: %s", track_id, err)
1964            return False
1965
1966    async def like_album(self, album_id: str) -> bool:
1967        """
1968        Add an album to liked albums.
1969
1970        :param album_id: Album ID to like.
1971        :return: True if successful.
1972        """
1973        try:
1974            result = await self._call_with_retry(lambda c: c.users_likes_albums_add(album_id))
1975            return result is not None
1976        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1977            LOGGER.error("Error liking album %s: %s", album_id, err)
1978            return False
1979
1980    async def unlike_album(self, album_id: str) -> bool:
1981        """
1982        Remove an album from liked albums.
1983
1984        :param album_id: Album ID to unlike.
1985        :return: True if successful.
1986        """
1987        try:
1988            result = await self._call_with_retry(lambda c: c.users_likes_albums_remove(album_id))
1989            return result is not None
1990        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
1991            LOGGER.error("Error unliking album %s: %s", album_id, err)
1992            return False
1993
1994    async def like_artist(self, artist_id: str) -> bool:
1995        """
1996        Add an artist to liked artists.
1997
1998        :param artist_id: Artist ID to like.
1999        :return: True if successful.
2000        """
2001        try:
2002            result = await self._call_with_retry(lambda c: c.users_likes_artists_add(artist_id))
2003            return result is not None
2004        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
2005            LOGGER.error("Error liking artist %s: %s", artist_id, err)
2006            return False
2007
2008    async def unlike_artist(self, artist_id: str) -> bool:
2009        """
2010        Remove an artist from liked artists.
2011
2012        :param artist_id: Artist ID to unlike.
2013        :return: True if successful.
2014        """
2015        try:
2016            result = await self._call_with_retry(lambda c: c.users_likes_artists_remove(artist_id))
2017            return result is not None
2018        except (BadRequestError, NetworkError, ProviderUnavailableError) as err:
2019            LOGGER.error("Error unliking artist %s: %s", artist_id, err)
2020            return False
2021