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