music-assistant-server

1"""Yandex Music provider implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import json
7import logging
8import random
9import uuid
10import zlib
11from collections.abc import AsyncGenerator, Sequence
12from io import BytesIO
13from typing import TYPE_CHECKING, Any
14
15from music_assistant_models.config_entries import (
16    ConfigActionResult,
17    ConfigEntry,
18    ConfigValueOption,
19)
20from music_assistant_models.enums import (
21    ConfigEntryType,
22    ImageType,
23    MediaType,
24    ProviderFeature,
25    StreamType,
26)
27from music_assistant_models.errors import (
28    InvalidDataError,
29    LoginFailed,
30    MediaNotFoundError,
31    ProviderUnavailableError,
32    ResourceTemporarilyUnavailable,
33)
34from music_assistant_models.media_items import (
35    Album,
36    Artist,
37    Audiobook,
38    BrowseFolder,
39    ItemMapping,
40    MediaItemChapter,
41    MediaItemImage,
42    MediaItemType,
43    Playlist,
44    Podcast,
45    PodcastEpisode,
46    ProviderMapping,
47    RecommendationFolder,
48    SearchResults,
49    Track,
50    UniqueList,
51)
52from music_assistant_models.streamdetails import StreamDetails
53from PIL import Image as PilImage
54from ya_passport_auth import SecretStr
55
56from music_assistant.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER
57from music_assistant.controllers.cache import use_cache
58from music_assistant.helpers.datetime import utc
59from music_assistant.models.music_provider import MusicProvider
60
61from .api_client import YandexMusicClient
62from .auth import refresh_credentials_via_passport, refresh_music_token
63from .constants import (
64    BROWSE_INITIAL_TRACKS,
65    COLLECTION_FOLDER_ID,
66    CONF_ACTION_DELETE_WAVE_PRESET,
67    CONF_ACTION_SAVE_WAVE_PRESET,
68    CONF_BASE_URL,
69    CONF_LIKED_TRACKS_MAX_TRACKS,
70    CONF_MY_WAVE_MAX_TRACKS,
71    CONF_QUALITY,
72    CONF_REFRESH_TOKEN,
73    CONF_RESTRICTIVE_RATE_LIMITS,
74    CONF_TOKEN,
75    CONF_WAVE_PRESET_DRAFT_DIVERSITY,
76    CONF_WAVE_PRESET_DRAFT_LANGUAGE,
77    CONF_WAVE_PRESET_DRAFT_MOOD,
78    CONF_WAVE_PRESET_DRAFT_NAME,
79    CONF_WAVE_PRESET_TO_DELETE,
80    CONF_WAVE_PRESETS_DATA,
81    CONF_X_TOKEN,
82    DEFAULT_BASE_URL,
83    DISCOVERY_INITIAL_TRACKS,
84    FOR_YOU_FOLDER_ID,
85    IMAGE_SIZE_MEDIUM,
86    LIKED_BATCH_JITTER_MIN_S,
87    LIKED_BATCH_JITTER_SPAN_S,
88    LIKED_TRACKS_PLAYLIST_ID,
89    LISTENING_HISTORY_FOLDER_ID,
90    MY_WAVE_BATCH_SIZE,
91    MY_WAVE_MODES_FOLDER_ID,
92    MY_WAVE_PLAYLIST_ID,
93    MY_WAVE_PRESETS_FOLDER_ID,
94    MY_WAVES_FOLDER_ID,
95    MY_WAVES_SET_FOLDER_ID,
96    PINNED_ITEMS_FOLDER_ID,
97    PLAYLIST_ID_SPLITTER,
98    QUALITY_BALANCED,
99    QUALITY_EFFICIENT,
100    QUALITY_HIGH,
101    QUALITY_SUPERB,
102    RADIO_FOLDER_ID,
103    RADIO_TRACK_ID_SEP,
104    ROTOR_STATION_MY_WAVE,
105    TAG_CATEGORY_ACTIVITY,
106    TAG_CATEGORY_ERA,
107    TAG_CATEGORY_GENRES,
108    TAG_CATEGORY_MOOD,
109    TAG_CATEGORY_ORDER,
110    TAG_MIXES,
111    TAG_SEASONAL_MAP,
112    TAG_SLUG_CATEGORY,
113    TRACK_BATCH_SIZE,
114    WAVE_CATEGORY_DISPLAY_ORDER,
115    WAVE_MODE_ORDER,
116    WAVE_MODE_PRESETS,
117    WAVE_MODE_SEP,
118    WAVE_PRESET_DIVERSITY_VALUES,
119    WAVE_PRESET_LANGUAGE_VALUES,
120    WAVE_PRESET_MOOD_VALUES,
121    WAVES_FOLDER_ID,
122    WAVES_LANDING_FOLDER_ID,
123)
124from .parsers import (
125    _get_image_url as get_image_url,
126)
127from .parsers import (
128    classify_album,
129    get_canonical_provider_name,
130    parse_album,
131    parse_artist,
132    parse_audiobook,
133    parse_playlist,
134    parse_podcast,
135    parse_podcast_episode,
136    parse_track,
137)
138from .presets import parse_stored_presets
139from .streaming import YandexMusicStreamingManager
140
141if TYPE_CHECKING:
142    from yandex_music import Album as YandexAlbum
143    from yandex_music import Track as YandexTrack
144
145
146# MediaType sub-paths that MA's default MusicProvider.browse() understands.
147# Used by the Collection dispatcher to delegate nested paths back to core.
148_COLLECTION_SUB_FOLDERS: frozenset[str] = frozenset(
149    {"tracks", "artists", "albums", "playlists", "audiobooks", "podcasts"}
150)
151
152# Collection sub-folder rows: (ProviderFeature, browse sub_id, strings.json label key,
153# is_playable). The sub_id ("tracks") and label key ("my_favorites") differ on purpose so the
154# Collection labels stay distinct from the core "media.folder.*" library labels.
155_COLLECTION_SUBFOLDERS: tuple[tuple[ProviderFeature, str, str, bool], ...] = (
156    (ProviderFeature.LIBRARY_TRACKS, "tracks", "my_favorites", True),
157    (ProviderFeature.LIBRARY_ARTISTS, "artists", "my_artists", True),
158    (ProviderFeature.LIBRARY_ALBUMS, "albums", "my_albums", True),
159    (ProviderFeature.LIBRARY_PLAYLISTS, "playlists", "my_playlists", True),
160    (ProviderFeature.LIBRARY_PODCASTS, "podcasts", "my_podcasts", False),
161    (ProviderFeature.LIBRARY_AUDIOBOOKS, "audiobooks", "my_audiobooks", False),
162)
163
164
165def _media_label_key(slug: str) -> str:
166    """Normalize a tag/category slug into its strings.json authoring key (spaces → underscores)."""
167    return slug.replace(" ", "_")
168
169
170def _split_wave_mode(station_id: str) -> tuple[str, dict[str, str]]:
171    """
172    Split a wave-mode station key into its base station ID and preset settings.
173
174    Keys like ``user:onyourwave#discover`` encode a specific preset on top of
175    the base rotor station. The part before ``#`` is the station ID that goes
176    to Yandex; the part after is a key into WAVE_MODE_PRESETS.
177
178    :param station_id: Station key, with or without a ``#preset`` suffix.
179    :return: Tuple of (base_station_id, settings_dict). The suffix, if
180        present, is always stripped — only the base station goes to
181        Yandex. ``settings_dict`` is the preset's settings when the suffix
182        matches a known WAVE_MODE_PRESETS key, or an empty dict otherwise
183        (unknown suffix → base station fired with no extra seeds).
184    """
185    if WAVE_MODE_SEP not in station_id:
186        return (station_id, {})
187    base, preset = station_id.split(WAVE_MODE_SEP, 1)
188    return (base, dict(WAVE_MODE_PRESETS.get(preset, {})))
189
190
191def _parse_radio_item_id(item_id: str) -> tuple[str, str | None]:
192    """
193    Extract track_id and optional station_id from provider item_id.
194
195    My Wave tracks use item_id format 'track_id@station_id'. Other tracks use
196    plain track_id.
197
198    :param item_id: Provider item_id (may contain RADIO_TRACK_ID_SEP).
199    :return: (track_id, station_id or None).
200    """
201    if RADIO_TRACK_ID_SEP in item_id:
202        parts = item_id.split(RADIO_TRACK_ID_SEP, 1)
203        return (parts[0], parts[1] if len(parts) > 1 else None)
204    return (item_id, None)
205
206
207def _extract_chapter_map_from_album(album: YandexAlbum) -> tuple[list[str], list[int]]:
208    """
209    Flatten an audiobook album's volumes into (chapter_track_ids, chapter_durations_ms).
210
211    Shared by ``_get_audiobook_stream_details`` and ``_resolve_audiobook_chapter_map``
212    so the two code paths can't drift (e.g. when we later filter bad tracks).
213    """
214    chapter_ids: list[str] = []
215    chapter_durations_ms: list[int] = []
216    for disc in album.volumes or []:
217        for track_obj in disc:
218            chapter_ids.append(str(track_obj.id))
219            chapter_durations_ms.append(int(track_obj.duration_ms or 0))
220    return chapter_ids, chapter_durations_ms
221
222
223def _merge_wave_preset(
224    name: str | None,
225    diversity: str | None,
226    mood: str | None,
227    language: str | None,
228    presets_data: str | None,
229) -> str:
230    """
231    Merge the given draft fields into the stored preset list and return the new JSON.
232
233    Overwrites an existing preset with the same name instead of creating a duplicate.
234    Raises ``InvalidDataError`` when the name is blank.
235
236    :param name: Draft preset name; blank/whitespace-only raises.
237    :param diversity: Draft diversity seed ("" / None → omitted).
238    :param mood: Draft mood/energy seed ("" / None → omitted).
239    :param language: Draft language seed ("" / None → omitted).
240    :param presets_data: The current stored presets JSON.
241    """
242    clean_name = name.strip() if isinstance(name, str) else ""
243    if not clean_name:
244        raise InvalidDataError("Please fill the preset name before saving.")
245    presets = parse_stored_presets(presets_data)
246    presets = [p for p in presets if p["name"] != clean_name]
247    new_preset: dict[str, str] = {
248        "name": clean_name,
249        **{
250            api_key: val
251            for val, api_key in (
252                (diversity, "diversity"),
253                (mood, "moodEnergy"),
254                (language, "language"),
255            )
256            if isinstance(val, str) and val
257        },
258    }
259    presets.append(new_preset)
260    return json.dumps(presets, ensure_ascii=False)
261
262
263def _remove_wave_preset(target: str | None, presets_data: str | None) -> str:
264    """
265    Remove the named preset from the stored list and return the new JSON.
266
267    Raises ``InvalidDataError`` when no name is selected. Idempotent — an absent
268    name simply rewrites an unchanged list.
269
270    :param target: Name of the preset to remove; blank/whitespace-only raises.
271    :param presets_data: The current stored presets JSON.
272    """
273    clean_target = target.strip() if isinstance(target, str) else ""
274    if not clean_target:
275        raise InvalidDataError("Please select a preset to delete.")
276    presets = parse_stored_presets(presets_data)
277    presets = [p for p in presets if p["name"] != clean_target]
278    return json.dumps(presets, ensure_ascii=False)
279
280
281def _wave_preset_config_entries(presets_data: str | None) -> list[ConfigEntry]:
282    """
283    Return the wave-preset builder UI (all advanced settings).
284
285    Layout:
286      - Section label showing how many presets are saved.
287      - Four "draft" fields (name + three dropdowns) the user fills in.
288      - "Save preset" action → copies draft into the JSON store.
289      - "Delete preset" dropdown + action (hidden when no presets exist).
290      - Hidden STRING carrying the JSON store itself.
291
292    Number of presets is unbounded; the user never edits JSON directly.
293
294    :param presets_data: The stored wave-presets JSON (``CONF_WAVE_PRESETS_DATA``).
295    """
296    empty_title = "— Default —"
297    diversity_options = [
298        ConfigValueOption(v, title=empty_title if not v else v.title())
299        for v in WAVE_PRESET_DIVERSITY_VALUES
300    ]
301    mood_options = [
302        ConfigValueOption(v, title=empty_title if not v else v.title())
303        for v in WAVE_PRESET_MOOD_VALUES
304    ]
305    language_options = [
306        ConfigValueOption(v, title=empty_title if not v else v.replace("-", " ").title())
307        for v in WAVE_PRESET_LANGUAGE_VALUES
308    ]
309
310    presets = parse_stored_presets(presets_data)
311    has_presets = bool(presets)
312    delete_options = [ConfigValueOption(p["name"], title=p["name"]) for p in presets]
313    if not delete_options:
314        # Empty options can break some frontends; supply a no-op placeholder.
315        delete_options = [ConfigValueOption("")]
316
317    return [
318        ConfigEntry(
319            key="wave_preset_section_label",
320            type=ConfigEntryType.LABEL,
321            translation_key="wave_preset_section_saved" if has_presets else None,
322            translation_params=[str(len(presets))] if has_presets else None,
323            advanced=True,
324        ),
325        ConfigEntry(
326            key=CONF_WAVE_PRESET_DRAFT_NAME,
327            type=ConfigEntryType.STRING,
328            default_value=None,
329            required=False,
330            advanced=True,
331        ),
332        ConfigEntry(
333            key=CONF_WAVE_PRESET_DRAFT_DIVERSITY,
334            type=ConfigEntryType.STRING,
335            options=diversity_options,
336            default_value="",
337            required=False,
338            advanced=True,
339        ),
340        ConfigEntry(
341            key=CONF_WAVE_PRESET_DRAFT_MOOD,
342            type=ConfigEntryType.STRING,
343            options=mood_options,
344            default_value="",
345            required=False,
346            advanced=True,
347        ),
348        ConfigEntry(
349            key=CONF_WAVE_PRESET_DRAFT_LANGUAGE,
350            type=ConfigEntryType.STRING,
351            options=language_options,
352            default_value="",
353            required=False,
354            advanced=True,
355        ),
356        ConfigEntry(
357            key=CONF_ACTION_SAVE_WAVE_PRESET,
358            type=ConfigEntryType.ACTION,
359            action=CONF_ACTION_SAVE_WAVE_PRESET,
360            advanced=True,
361        ),
362        ConfigEntry(
363            key=CONF_WAVE_PRESET_TO_DELETE,
364            type=ConfigEntryType.STRING,
365            options=delete_options,
366            default_value="",
367            required=False,
368            advanced=True,
369            hidden=not has_presets,
370        ),
371        ConfigEntry(
372            key=CONF_ACTION_DELETE_WAVE_PRESET,
373            type=ConfigEntryType.ACTION,
374            action=CONF_ACTION_DELETE_WAVE_PRESET,
375            advanced=True,
376            hidden=not has_presets,
377        ),
378        ConfigEntry(
379            key=CONF_WAVE_PRESETS_DATA,
380            type=ConfigEntryType.STRING,
381            default_value="",
382            required=False,
383            advanced=True,
384            hidden=True,
385        ),
386    ]
387
388
389class _WaveState:
390    """
391    Per-station mutable state for rotor wave playback.
392
393    Holds both the new session-based rotor identifiers (`session_id`) and the
394    legacy stations-based ones (`batch_id`). Call sites prefer `session_id`
395    when present; `batch_id` is still carried because feedback events anchor
396    to a specific batch within the session.
397    """
398
399    def __init__(self) -> None:
400        self.session_id: str | None = None
401        self.batch_id: str | None = None
402        self.last_track_id: str | None = None
403        self.playlist_next_cursor: str | None = None
404        self.seen_track_ids: set[str] = set()
405        self.radio_started_sent: bool = False
406        self.prefetched: list[Any] = []
407        self.settings: dict[str, str] = {}
408        self.lock: asyncio.Lock = asyncio.Lock()
409
410
411class YandexMusicProvider(MusicProvider):
412    """Implementation of a Yandex Music MusicProvider."""
413
414    _client: YandexMusicClient | None = None
415    _streaming: YandexMusicStreamingManager | None = None
416    _wave_states: dict[str, _WaveState]  # Per-station state (incl. My Wave)
417    _wave_bg_colors: dict[str, str]  # image_url -> hex bg color for transparent covers
418    # Short-lived cache to dedupe the three library syncs (albums/podcasts/audiobooks)
419    # that all derive from the same liked-albums endpoint.
420    _liked_albums_cache: tuple[float, list[YandexAlbum]] | None = None
421    _liked_albums_lock: asyncio.Lock
422    # Per-audiobook cache of (chapter_track_ids, chapter_durations_ms) used to
423    # report playback progress per chapter via play_audio.
424    _audiobook_chapter_cache: dict[str, tuple[list[str], list[int]]]
425    # Stable play_id per audiobook session, cleared in on_streamed.
426    _audiobook_play_ids: dict[str, str]
427
428    @property
429    def client(self) -> YandexMusicClient:
430        """Return the Yandex Music client."""
431        if self._client is None:
432            raise ProviderUnavailableError("Provider not initialized")
433        return self._client
434
435    @property
436    def streaming(self) -> YandexMusicStreamingManager:
437        """Return the streaming manager."""
438        if self._streaming is None:
439            raise ProviderUnavailableError("Provider not initialized")
440        return self._streaming
441
442    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
443        """
444        Return Config entries to configure this provider.
445
446        Authentication runs in the interactive setup flow (see setup_flow.py); this
447        surface only exposes the genuine playback options and the My Wave preset builder
448        (whose save/delete actions are handled in ``handle_config_action``).
449        """
450        return (
451            CONF_ENTRY_UNOFFICIAL_PROVIDER,
452            # Quality
453            ConfigEntry(
454                key=CONF_QUALITY,
455                type=ConfigEntryType.STRING,
456                options=[
457                    ConfigValueOption(QUALITY_EFFICIENT),
458                    ConfigValueOption(QUALITY_BALANCED),
459                    ConfigValueOption(QUALITY_HIGH),
460                    ConfigValueOption(QUALITY_SUPERB),
461                ],
462                default_value=QUALITY_BALANCED,
463            ),
464            # My Wave maximum tracks (advanced)
465            ConfigEntry(
466                key=CONF_MY_WAVE_MAX_TRACKS,
467                type=ConfigEntryType.INTEGER,
468                range=(10, 1000),
469                default_value=150,
470                required=False,
471                advanced=True,
472            ),
473            # User-defined wave presets: builder + save/delete actions (dynamic list)
474            *_wave_preset_config_entries(
475                self.get_config_value(CONF_WAVE_PRESETS_DATA, return_type=str)
476            ),
477            # Liked Tracks maximum tracks (advanced)
478            ConfigEntry(
479                key=CONF_LIKED_TRACKS_MAX_TRACKS,
480                type=ConfigEntryType.INTEGER,
481                range=(50, 2000),
482                default_value=200,
483                required=False,
484                advanced=True,
485            ),
486            # API Base URL (advanced)
487            ConfigEntry(
488                key=CONF_BASE_URL,
489                type=ConfigEntryType.STRING,
490                translation_params=[DEFAULT_BASE_URL],
491                default_value=DEFAULT_BASE_URL,
492                required=False,
493                advanced=True,
494            ),
495            # Restrictive rate limits (advanced)
496            ConfigEntry(
497                key=CONF_RESTRICTIVE_RATE_LIMITS,
498                type=ConfigEntryType.BOOLEAN,
499                default_value=False,
500                required=False,
501                advanced=True,
502            ),
503        )
504
505    async def handle_config_action(
506        self, action: str
507    ) -> tuple[ConfigEntry, ...] | ConfigActionResult | None:
508        """
509        Handle a wave-preset save/delete button press and re-render the entries.
510
511        Both actions mutate the hidden JSON store and clear the draft / selection
512        fields so the UI re-renders in a clean state. Draft values are read from
513        stored config (no form values are passed) and persisted immediately.
514
515        :param action: The action id of the pressed button.
516        """
517        if action == CONF_ACTION_SAVE_WAVE_PRESET:
518            new_presets = _merge_wave_preset(
519                self.get_config_value(CONF_WAVE_PRESET_DRAFT_NAME, return_type=str),
520                self.get_config_value(CONF_WAVE_PRESET_DRAFT_DIVERSITY, return_type=str),
521                self.get_config_value(CONF_WAVE_PRESET_DRAFT_MOOD, return_type=str),
522                self.get_config_value(CONF_WAVE_PRESET_DRAFT_LANGUAGE, return_type=str),
523                self.get_config_value(CONF_WAVE_PRESETS_DATA, return_type=str),
524            )
525            self._update_config_value(CONF_WAVE_PRESETS_DATA, new_presets, immediate=True)
526            # Clear draft so the UI is ready for the next preset
527            self._update_config_value(CONF_WAVE_PRESET_DRAFT_NAME, None, immediate=True)
528            self._update_config_value(CONF_WAVE_PRESET_DRAFT_DIVERSITY, "", immediate=True)
529            self._update_config_value(CONF_WAVE_PRESET_DRAFT_MOOD, "", immediate=True)
530            self._update_config_value(CONF_WAVE_PRESET_DRAFT_LANGUAGE, "", immediate=True)
531            return await self.get_config_entries()
532        if action == CONF_ACTION_DELETE_WAVE_PRESET:
533            new_presets = _remove_wave_preset(
534                self.get_config_value(CONF_WAVE_PRESET_TO_DELETE, return_type=str),
535                self.get_config_value(CONF_WAVE_PRESETS_DATA, return_type=str),
536            )
537            self._update_config_value(CONF_WAVE_PRESETS_DATA, new_presets, immediate=True)
538            self._update_config_value(CONF_WAVE_PRESET_TO_DELETE, "", immediate=True)
539            return await self.get_config_entries()
540        return await super().handle_config_action(action)
541
542    def _media_label(self, group: str, key: str, fallback: str) -> tuple[str, str | None]:
543        """
544        Resolve a media label to its English ``name`` and ``translation_key``.
545
546        The English source string lives in the provider's ``strings.json`` (the single source
547        of truth) and is localized for the connection locale at serialization via the returned
548        key. An unauthored key — e.g. a tag discovered from Yandex's landing API — returns
549        ``(fallback, None)`` so its already-localized name is kept verbatim.
550
551        :param group: Media translation group (``folder``, ``recommendations`` or ``playlist``).
552        :param key: Authoring key within the group; also the item's ``translation_key``.
553        :param fallback: English name to use when no string is authored for *key*.
554        """
555        authored = self.mass.translations.get_translation(
556            f"provider.{self.domain}.media.{group}.{key}.name"
557        )
558        if authored is None:
559            return fallback, None
560        return authored, key
561
562    async def _reauth_via_refresh_token(
563        self, x_token: str, refresh_token: str, base_url: str, original_err: Exception
564    ) -> None:
565        """
566        Silently re-issue full credentials when x_token refresh fails.
567
568        Device-flow accounts have a refresh_token that can mint a new
569        x_token + refresh_token + music_token without any user interaction.
570        Persists the rotated triple and connects the client. Any failure
571        here is terminal — clears all credentials and forces re-auth.
572        """
573        try:
574            new_creds = await refresh_credentials_via_passport(
575                SecretStr(x_token), SecretStr(refresh_token)
576            )
577        except ResourceTemporarilyUnavailable as err2:
578            # Transient Passport failure — keep creds, let MA retry later
579            self.logger.warning(
580                "Credential refresh temporarily unavailable: %s", type(err2).__name__
581            )
582            raise ProviderUnavailableError(
583                "Unable to refresh credentials right now. Please try again later."
584            ) from err2
585        except LoginFailed as err2:
586            self.logger.warning("Session and refresh tokens are both expired")
587            self._update_setup_data(CONF_TOKEN, None)
588            self._update_setup_data(CONF_X_TOKEN, None)
589            self._update_setup_data(CONF_REFRESH_TOKEN, None)
590            raise LoginFailed("Session expired. Please re-authenticate.") from err2
591
592        new_music_token = new_creds.music_token
593        new_refresh_token = new_creds.refresh_token
594        if new_music_token is None or new_refresh_token is None:
595            self._update_setup_data(CONF_TOKEN, None)
596            self._update_setup_data(CONF_X_TOKEN, None)
597            self._update_setup_data(CONF_REFRESH_TOKEN, None)
598            raise LoginFailed(
599                "Credential refresh returned an incomplete response."
600            ) from original_err
601
602        self._update_setup_data(CONF_TOKEN, new_music_token.get_secret())
603        self._update_setup_data(CONF_X_TOKEN, new_creds.x_token.get_secret())
604        self._update_setup_data(CONF_REFRESH_TOKEN, new_refresh_token.get_secret())
605        restrictive = bool(self.config.get_value(CONF_RESTRICTIVE_RATE_LIMITS, False))
606        self._client = YandexMusicClient(
607            new_music_token, base_url=base_url, restrictive_rate_limits=restrictive
608        )
609        await self._client.connect()
610        self.logger.info("Re-issued credentials silently from refresh token")
611
612    async def handle_async_init(self) -> None:
613        """Handle async initialization of the provider."""
614        token = self.get_setup_value(CONF_TOKEN)
615        x_token = self.get_setup_value(CONF_X_TOKEN)
616        refresh_token = self.get_setup_value(CONF_REFRESH_TOKEN)
617        base_url = self.config.get_value(CONF_BASE_URL, DEFAULT_BASE_URL)
618        restrictive = bool(self.config.get_value(CONF_RESTRICTIVE_RATE_LIMITS, False))
619
620        if not token and not x_token:
621            raise LoginFailed("No Yandex Music token provided. Please authenticate.")
622
623        # Try existing music token first (fast path)
624        if token:
625            try:
626                self._client = YandexMusicClient(
627                    SecretStr(str(token)),
628                    base_url=str(base_url),
629                    restrictive_rate_limits=restrictive,
630                )
631                await self._client.connect()
632            except LoginFailed:
633                self.logger.warning("Music token is invalid or expired")
634                # Clear the dead token so restarts go straight to refresh
635                self._update_setup_data(CONF_TOKEN, None)
636                if x_token:
637                    self.logger.info("Attempting to refresh from session token")
638                    token = None
639                    self._client = None
640                else:
641                    raise
642
643        # Refresh from x_token if music token absent or failed
644        if not token and x_token:
645            try:
646                new_music_token = await refresh_music_token(SecretStr(str(x_token)))
647                self._update_setup_data(CONF_TOKEN, new_music_token.get_secret())
648                self._client = YandexMusicClient(
649                    new_music_token,
650                    base_url=str(base_url),
651                    restrictive_rate_limits=restrictive,
652                )
653                await self._client.connect()
654                self.logger.info("Refreshed music token from session token")
655            except LoginFailed as err:
656                # x_token refresh failed. If a refresh_token is available
657                # (device-flow accounts), try silent re-issue of the full
658                # credential triple before giving up.
659                if refresh_token:
660                    await self._reauth_via_refresh_token(
661                        str(x_token), str(refresh_token), str(base_url), err
662                    )
663                else:
664                    # Definitive auth failure — clear dead credentials
665                    self.logger.warning("Session token is invalid or expired")
666                    self._update_setup_data(CONF_TOKEN, None)
667                    self._update_setup_data(CONF_X_TOKEN, None)
668                    raise LoginFailed("Session token expired. Please re-authenticate.") from err
669            except asyncio.CancelledError:
670                raise
671            except Exception as err:
672                # Transient/network failure — keep credentials for retry
673                self.logger.warning(
674                    "Session token refresh failed (network): %s",
675                    type(err).__name__,
676                )
677                raise ProviderUnavailableError(
678                    "Unable to refresh music token right now. Please try again later."
679                ) from err
680
681        # Suppress yandex_music library DEBUG dumps (full API request/response JSON)
682        logging.getLogger("yandex_music").setLevel(self.logger.level + 10)
683        # Propagate the MA instance log level to our per-module loggers
684        # (api_client, streaming, parsers, auth) so DEBUG hooks there actually
685        # print when MA is set to DEBUG for this provider.
686        logging.getLogger("music_assistant.providers.yandex_music").setLevel(self.logger.level)
687        self._streaming = YandexMusicStreamingManager(self)
688        # Per-station wave state (incl. My Wave under ROTOR_STATION_MY_WAVE).
689        # Entries are created lazily by _get_wave_state() on first access.
690        self._wave_states = {}
691        self._wave_bg_colors = {}
692        self._liked_albums_lock, self._liked_albums_cache = asyncio.Lock(), None
693        self._audiobook_chapter_cache, self._audiobook_play_ids = {}, {}
694        self.logger.info("Successfully connected to Yandex Music")
695
696    async def unload(self, is_removed: bool = False) -> None:
697        """
698        Handle unload/close of the provider.
699
700        :param is_removed: Whether the provider is being removed.
701        """
702        if self._client:
703            await self._client.disconnect()
704        self._client = None
705        self._streaming = None
706        self._wave_states.clear()
707        self._wave_bg_colors.clear()
708        self._liked_albums_cache = None
709        self._audiobook_chapter_cache.clear()
710        self._audiobook_play_ids.clear()
711        await super().unload(is_removed)
712
713    def get_item_mapping(self, media_type: MediaType | str, key: str, name: str) -> ItemMapping:
714        """
715        Create a generic item mapping.
716
717        :param media_type: The media type.
718        :param key: The item ID.
719        :param name: The item name.
720        :return: An ItemMapping instance.
721        """
722        if isinstance(media_type, str):
723            media_type = MediaType(media_type)
724        return ItemMapping(
725            media_type=media_type,
726            item_id=key,
727            provider=self.instance_id,
728            name=name,
729        )
730
731    async def browse(  # noqa: PLR0911, PLR0915
732        self, path: str
733    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
734        """
735        Browse provider items with locale-based folder names and My Wave.
736
737        Root level shows My Wave, artists, albums, liked tracks, playlists. Names
738        are in Russian when MA locale is ru_*, otherwise in English. My Wave
739        tracks use item_id format track_id@station_id for rotor feedback.
740
741        :param path: The path to browse (e.g. provider_id:// or provider_id://artists).
742        """
743        if ProviderFeature.BROWSE not in self.supported_features:
744            raise NotImplementedError
745
746        path_parts = path.split("://")[1].split("/") if "://" in path else []
747        subpath = path_parts[0] if len(path_parts) > 0 else None
748        sub_subpath = path_parts[1] if len(path_parts) > 1 else None
749
750        if subpath == MY_WAVE_PLAYLIST_ID:
751            async with self._get_wave_state(ROTOR_STATION_MY_WAVE).lock:
752                return await self._browse_my_wave(path, sub_subpath)
753
754        # Wave modes — accept two equivalent URL forms so both browse
755        # navigation (slash form "my_wave_modes/<preset>", emitted by our
756        # listing) and MA's play-time reconstruction (underscore form
757        # "my_wave_modes_<preset>", built as "<instance>://<item_id>") work.
758        mode_preset: str | None = None
759        if subpath == MY_WAVE_MODES_FOLDER_ID and sub_subpath is None:
760            return self._browse_my_wave_modes_list(path)
761        if subpath == MY_WAVE_MODES_FOLDER_ID and sub_subpath is not None:
762            mode_preset = sub_subpath if sub_subpath != "next" else None
763            if mode_preset is None:
764                return []
765            load_more_modes = len(path_parts) > 2 and path_parts[2] == "next"
766        elif subpath and subpath.startswith(f"{MY_WAVE_MODES_FOLDER_ID}_"):
767            mode_preset = subpath[len(MY_WAVE_MODES_FOLDER_ID) + 1 :]
768            load_more_modes = sub_subpath == "next"
769        if mode_preset is not None:
770            if mode_preset not in WAVE_MODE_PRESETS:
771                return []
772            station_key = f"{ROTOR_STATION_MY_WAVE}{WAVE_MODE_SEP}{mode_preset}"
773            async with self._get_wave_state(station_key).lock:
774                return await self._browse_my_wave_mode(path, station_key, load_more_modes)
775
776        # User-saved wave presets — same dual-form handling.
777        preset_idx: int | None = None
778        load_more_presets = False
779        if subpath == MY_WAVE_PRESETS_FOLDER_ID and sub_subpath is None:
780            return self._browse_user_presets_list(path, self._get_user_wave_presets())
781        if subpath == MY_WAVE_PRESETS_FOLDER_ID and sub_subpath is not None:
782            try:
783                preset_idx = int(sub_subpath)
784            except ValueError:
785                return []
786            load_more_presets = len(path_parts) > 2 and path_parts[2] == "next"
787        elif subpath and subpath.startswith(f"{MY_WAVE_PRESETS_FOLDER_ID}_"):
788            try:
789                preset_idx = int(subpath[len(MY_WAVE_PRESETS_FOLDER_ID) + 1 :])
790            except ValueError:
791                return []
792            load_more_presets = sub_subpath == "next"
793        if preset_idx is not None:
794            user_presets = self._get_user_wave_presets()
795            if not 0 <= preset_idx < len(user_presets):
796                return []
797            preset_data = user_presets[preset_idx]
798            station_key = f"{ROTOR_STATION_MY_WAVE}{WAVE_MODE_SEP}preset_{preset_idx}"
799            wave = self._get_wave_state(station_key)
800            # Stash user-chosen settings so _fetch_rotor_session_batch sends them
801            wave.settings = {
802                k: v
803                for k, v in preset_data.items()
804                if k in ("diversity", "moodEnergy", "language") and v
805            }
806            async with wave.lock:
807                return await self._browse_my_wave_mode(path, station_key, load_more_presets)
808
809        # For You folder (picks + mixes)
810        if subpath == FOR_YOU_FOLDER_ID:
811            return await self._browse_for_you(path, path_parts)
812
813        # Collection folder (library items). Two shapes:
814        #   <prov>://collection              → listing of library sub-folders
815        #   <prov>://collection/<sub>        → delegate to MA's library handler
816        # The nested form is what lets MA's "back" button return here (strip
817        # last /-segment) instead of dumping the user at the provider root.
818        if subpath == COLLECTION_FOLDER_ID:
819            if sub_subpath in _COLLECTION_SUB_FOLDERS:
820                return await super().browse(f"{self.instance_id}://{sub_subpath}")
821            return await self._browse_collection(path)
822
823        # Handle picks/ path (mood, activity, era, genres)
824        if subpath == "picks":
825            return await self._browse_picks(path, path_parts)
826
827        # Handle mixes/ path (seasonal collections)
828        if subpath == "mixes":
829            return await self._browse_mixes(path, path_parts)
830
831        # Handle waves/ and radio/ paths (rotor stations by genre/mood/activity)
832        if subpath in (WAVES_FOLDER_ID, RADIO_FOLDER_ID):
833            return await self._browse_waves(path, path_parts)
834
835        # Handle my_waves_set/ path (AI Wave Sets from /landing-blocks/mixes-waves)
836        if subpath == MY_WAVES_SET_FOLDER_ID:
837            return await self._browse_vibe_sets(path, path_parts)
838
839        # Pinned items folder
840        if subpath == PINNED_ITEMS_FOLDER_ID:
841            return await self._browse_pins()
842
843        # Listening history folder
844        if subpath == LISTENING_HISTORY_FOLDER_ID:
845            return await self._browse_history()
846
847        # Handle waves_landing/ path (Featured Waves from /landing-blocks/waves)
848        if subpath == WAVES_LANDING_FOLDER_ID:
849            return await self._browse_waves_landing(path, path_parts)
850
851        # Handle direct tag subpath (when folder is played by URI, the full path
852        # "picks/category/tag" is lost and only the tag slug arrives as subpath).
853        # Skip the API call for standard top-level folders that are never tag slugs.
854        _known_folders = {
855            "artists",
856            "albums",
857            "tracks",
858            "playlists",
859            "audiobooks",
860            "podcasts",
861            LIKED_TRACKS_PLAYLIST_ID,
862            WAVES_FOLDER_ID,
863            RADIO_FOLDER_ID,
864            MY_WAVES_FOLDER_ID,
865            MY_WAVES_SET_FOLDER_ID,
866            WAVES_LANDING_FOLDER_ID,
867            FOR_YOU_FOLDER_ID,
868            COLLECTION_FOLDER_ID,
869            PINNED_ITEMS_FOLDER_ID,
870            LISTENING_HISTORY_FOLDER_ID,
871        }
872        if subpath and subpath not in _known_folders:
873            # Handle direct wave station_id (e.g. "activity:workout") passed when
874            # MA plays a wave station folder using its item_id as the path subpath.
875            # Station IDs have format "category:tag" where category is non-numeric.
876            if ":" in subpath:
877                cat_part = subpath.split(":", 1)[0]
878                if not cat_part.isdigit():
879                    return await self._browse_wave_station(subpath)
880
881            discovered_tags = await self._get_discovered_tag_slugs()
882            if subpath in discovered_tags:
883                return await self._get_tag_playlists_as_browse(subpath)
884
885        if subpath:
886            return await super().browse(path)
887
888        # The English name on each folder doubles as the fallback; translation_key localizes
889        # it for the connection locale at serialization (the server is the single source).
890        items: list[MediaItemType | ItemMapping | BrowseFolder] = []
891        base = path if path.endswith("//") else path.rstrip("/") + "/"
892        # Expose My Wave as its dynamic virtual playlist so queue refills stay enabled.
893        items.append(await self.get_playlist(MY_WAVE_PLAYLIST_ID))
894        # Wave modes folder (P4): discover / calm / active / language presets
895        items.append(
896            BrowseFolder(
897                item_id=MY_WAVE_MODES_FOLDER_ID,
898                provider=self.instance_id,
899                path=f"{base}{MY_WAVE_MODES_FOLDER_ID}",
900                name="Wave Modes",
901                translation_key=MY_WAVE_MODES_FOLDER_ID,
902                is_playable=False,
903            )
904        )
905        # User-defined wave presets (P8) — shown only when any configured.
906        if self._get_user_wave_presets():
907            items.append(
908                BrowseFolder(
909                    item_id=MY_WAVE_PRESETS_FOLDER_ID,
910                    provider=self.instance_id,
911                    path=f"{base}{MY_WAVE_PRESETS_FOLDER_ID}",
912                    name="My Presets",
913                    translation_key=MY_WAVE_PRESETS_FOLDER_ID,
914                    is_playable=False,
915                )
916            )
917        # For You folder — Picks + Mixes (Яндекс «Для вас»)
918        items.append(
919            BrowseFolder(
920                item_id=FOR_YOU_FOLDER_ID,
921                provider=self.instance_id,
922                path=f"{base}{FOR_YOU_FOLDER_ID}",
923                name="For You",
924                translation_key=FOR_YOU_FOLDER_ID,
925                is_playable=False,
926            )
927        )
928        # Collection folder — library items (Яндекс «Коллекция»)
929        has_library = any(
930            f in self.supported_features
931            for f in (
932                ProviderFeature.LIBRARY_ARTISTS,
933                ProviderFeature.LIBRARY_ALBUMS,
934                ProviderFeature.LIBRARY_TRACKS,
935                ProviderFeature.LIBRARY_PLAYLISTS,
936            )
937        )
938        if has_library:
939            items.append(
940                BrowseFolder(
941                    item_id=COLLECTION_FOLDER_ID,
942                    provider=self.instance_id,
943                    path=f"{base}{COLLECTION_FOLDER_ID}",
944                    name="Collection",
945                    translation_key=COLLECTION_FOLDER_ID,
946                    is_playable=False,
947                )
948            )
949        # Radio folder — rotor stations (Яндекс волны, shown as Radio)
950        items.append(
951            BrowseFolder(
952                item_id=RADIO_FOLDER_ID,
953                provider=self.instance_id,
954                path=f"{base}{RADIO_FOLDER_ID}",
955                name="Radio",
956                translation_key=RADIO_FOLDER_ID,
957                is_playable=False,
958            )
959        )
960        # AI Wave Sets — parametric stations from /landing-blocks/mixes-waves
961        items.append(
962            BrowseFolder(
963                item_id=MY_WAVES_SET_FOLDER_ID,
964                provider=self.instance_id,
965                path=f"{base}{MY_WAVES_SET_FOLDER_ID}",
966                name="AI Wave Sets",
967                translation_key=MY_WAVES_SET_FOLDER_ID,
968                is_playable=False,
969            )
970        )
971        # Pinned items — user-pinned artists/albums/playlists/waves
972        items.append(
973            BrowseFolder(
974                item_id=PINNED_ITEMS_FOLDER_ID,
975                provider=self.instance_id,
976                path=f"{base}{PINNED_ITEMS_FOLDER_ID}",
977                name="Pinned",
978                translation_key=PINNED_ITEMS_FOLDER_ID,
979                is_playable=False,
980            )
981        )
982        # Listening history — recently played tracks/albums
983        items.append(
984            BrowseFolder(
985                item_id=LISTENING_HISTORY_FOLDER_ID,
986                provider=self.instance_id,
987                path=f"{base}{LISTENING_HISTORY_FOLDER_ID}",
988                name="Listening History",
989                translation_key=LISTENING_HISTORY_FOLDER_ID,
990                is_playable=False,
991            )
992        )
993        if len(items) == 1 and isinstance(items[0], BrowseFolder):
994            return await self.browse(items[0].path)
995        return items
996
997    async def _browse_my_wave(
998        self, path: str, sub_subpath: str | None
999    ) -> list[Track | BrowseFolder]:
1000        """
1001        Browse My Wave tracks (must be called under the My Wave state lock).
1002
1003        :param path: Full browse path.
1004        :param sub_subpath: Sub-path part ('next' for load more, or track_id cursor).
1005        :return: List of Track and optional BrowseFolder for "Load more".
1006        """
1007        wave = self._get_wave_state(ROTOR_STATION_MY_WAVE)
1008        max_tracks_config = int(
1009            self.config.get_value(CONF_MY_WAVE_MAX_TRACKS) or 150  # type: ignore[arg-type]
1010        )
1011        batch_size_config = MY_WAVE_BATCH_SIZE
1012
1013        # Effective limit on tracks to collect for this call:
1014        # initial browse is capped to BROWSE_INITIAL_TRACKS to avoid marking
1015        # extra tracks as "seen" that are never shown to the user.
1016        effective_limit = min(
1017            BROWSE_INITIAL_TRACKS if sub_subpath != "next" else max_tracks_config,
1018            max_tracks_config,
1019        )
1020
1021        # Root my_wave: fetch up to batch_size_config batches so Play adds more tracks.
1022        # "Load more" always uses single next batch.
1023        max_batches = batch_size_config if sub_subpath != "next" else 1
1024
1025        # Reset seen tracks on fresh browse (not "load more")
1026        if sub_subpath != "next":
1027            wave.seen_track_ids = set()
1028
1029        queue: str | int | None = None
1030        if sub_subpath == "next":
1031            queue = wave.last_track_id
1032        elif sub_subpath:
1033            queue = sub_subpath
1034
1035        all_tracks: list[Track | BrowseFolder] = []
1036        last_batch_id: str | None = None
1037        first_track_id_this_batch: str | None = None
1038        total_track_count = 0
1039
1040        for _ in range(max_batches):
1041            if total_track_count >= effective_limit:
1042                break
1043
1044            # On a fresh browse (non-"next"), honour any sub_subpath cursor override
1045            # by seeding wave.last_track_id so the helper picks it up.
1046            if queue is not None:
1047                wave.last_track_id = str(queue)
1048            yandex_tracks, batch_id = await self._fetch_rotor_session_batch(
1049                wave, ROTOR_STATION_MY_WAVE
1050            )
1051            if batch_id:
1052                last_batch_id = batch_id
1053            if not wave.radio_started_sent and yandex_tracks:
1054                sent = await self._send_wave_feedback(wave, ROTOR_STATION_MY_WAVE, "radioStarted")
1055                if sent:
1056                    wave.radio_started_sent = True
1057            first_track_id_this_batch = None
1058            for yt in yandex_tracks:
1059                if total_track_count >= effective_limit:
1060                    break
1061
1062                track = self._parse_my_wave_track(yt, wave.seen_track_ids)
1063                if track is None:
1064                    continue
1065                all_tracks.append(track)
1066                total_track_count += 1
1067
1068                track_id = track.item_id.split(RADIO_TRACK_ID_SEP, 1)[0]
1069                if first_track_id_this_batch is None:
1070                    first_track_id_this_batch = track_id
1071
1072            if first_track_id_this_batch is not None:
1073                wave.last_track_id = first_track_id_this_batch
1074            if (
1075                first_track_id_this_batch is None
1076                or not batch_id
1077                or not yandex_tracks
1078                or total_track_count >= effective_limit
1079            ):
1080                break
1081            queue = first_track_id_this_batch
1082
1083        # Only show "Load more" if we haven't reached the limit and there's more data
1084        if last_batch_id and total_track_count < max_tracks_config:
1085            all_tracks.append(
1086                BrowseFolder(
1087                    item_id="next",
1088                    provider=self.instance_id,
1089                    path=f"{path.rstrip('/')}/next",
1090                    name="Load more",
1091                    translation_key="load_more",
1092                    is_playable=False,
1093                )
1094            )
1095        return all_tracks
1096
1097    def _get_user_wave_presets(self) -> list[dict[str, str]]:
1098        """
1099        Decode user-defined wave presets from the hidden JSON config key.
1100
1101        Thin wrapper around :func:`presets.parse_stored_presets` so browse
1102        code and settings actions use the exact same parsing — avoids schema
1103        drift when preset fields are added or renamed.
1104        """
1105        return parse_stored_presets(self.config.get_value(CONF_WAVE_PRESETS_DATA))
1106
1107    def _browse_user_presets_list(
1108        self, path: str, presets: list[dict[str, str]]
1109    ) -> list[BrowseFolder]:
1110        """
1111        Return one playable BrowseFolder per configured user preset.
1112
1113        ``path`` is nested (``my_wave_presets/<idx>``) so MA's back-nav —
1114        which strips the last ``/``-segment — returns the user to the
1115        listing instead of the provider root. ``item_id`` uses the
1116        underscore form (``my_wave_presets_<idx>``) because MA rebuilds a
1117        playable folder's path from its item_id at play time. The browse
1118        dispatcher accepts both forms.
1119
1120        :param path: Current browse path.
1121        :param presets: Sanitized presets from ``_get_user_wave_presets``.
1122        :return: List of playable BrowseFolder entries.
1123        """
1124        base = path if path.endswith("/") else f"{path}/"
1125        folders: list[BrowseFolder] = []
1126        for idx, preset in enumerate(presets):
1127            folders.append(
1128                BrowseFolder(
1129                    item_id=f"{MY_WAVE_PRESETS_FOLDER_ID}_{idx}",
1130                    provider=self.instance_id,
1131                    path=f"{base}{idx}",
1132                    name=preset.get("name", f"Preset {idx + 1}"),
1133                    is_playable=True,
1134                )
1135            )
1136        return folders
1137
1138    def _browse_my_wave_modes_list(self, path: str) -> list[BrowseFolder]:
1139        """
1140        Return the 11 wave-mode entries as playable browse folders.
1141
1142        Same dual-form contract as user presets: nested ``path`` keeps
1143        back-navigation intact, underscore ``item_id`` survives MA's
1144        play-time reconstruction.
1145
1146        :param path: Browse path the user navigated into.
1147        :return: Ordered list of BrowseFolder entries, one per preset.
1148        """
1149        base = path if path.endswith("/") else f"{path}/"
1150        folders: list[BrowseFolder] = []
1151        for preset in WAVE_MODE_ORDER:
1152            name, translation_key = self._media_label(
1153                "folder", f"wave_mode_{preset}", preset.replace("_", " ").title()
1154            )
1155            folders.append(
1156                BrowseFolder(
1157                    item_id=f"{MY_WAVE_MODES_FOLDER_ID}_{preset}",
1158                    provider=self.instance_id,
1159                    path=f"{base}{preset}",
1160                    name=name,
1161                    translation_key=translation_key,
1162                    is_playable=True,
1163                )
1164            )
1165        return folders
1166
1167    async def _browse_my_wave_mode(
1168        self, path: str, station_key: str, load_more: bool
1169    ) -> list[Track | BrowseFolder]:
1170        """
1171        Fetch a batch of tracks for a specific wave-mode preset.
1172
1173        Reuses the session-API machinery: tracks live in
1174        ``_wave_states[station_key]`` where station_key is
1175        ``user:onyourwave#{preset}``. Tracks carry composite item_ids that
1176        route feedback back to this state.
1177
1178        :param path: Full browse path to this preset.
1179        :param station_key: Station key with a ``#preset`` suffix.
1180        :param load_more: True when called for ``.../next`` pagination.
1181        :return: Tracks + optional "Load more" folder.
1182        """
1183        wave = self._get_wave_state(station_key)
1184        max_tracks_config = int(
1185            self.config.get_value(CONF_MY_WAVE_MAX_TRACKS) or 150  # type: ignore[arg-type]
1186        )
1187        batch_size_config = MY_WAVE_BATCH_SIZE
1188        effective_limit = min(
1189            BROWSE_INITIAL_TRACKS if not load_more else max_tracks_config,
1190            max_tracks_config,
1191        )
1192        max_batches = batch_size_config if not load_more else 1
1193
1194        if not load_more:
1195            wave.seen_track_ids = set()
1196
1197        all_tracks: list[Track | BrowseFolder] = []
1198        last_batch_id: str | None = None
1199        total_track_count = 0
1200
1201        for _ in range(max_batches):
1202            if total_track_count >= effective_limit:
1203                break
1204            yandex_tracks, batch_id = await self._fetch_rotor_session_batch(wave, station_key)
1205            if batch_id:
1206                last_batch_id = batch_id
1207            if not wave.radio_started_sent and yandex_tracks:
1208                sent = await self._send_wave_feedback(wave, station_key, "radioStarted")
1209                if sent:
1210                    wave.radio_started_sent = True
1211            first_track_id_this_batch: str | None = None
1212            for yt in yandex_tracks:
1213                if total_track_count >= effective_limit:
1214                    break
1215                track = self._parse_my_wave_track(yt, wave.seen_track_ids, station_key=station_key)
1216                if track is None:
1217                    continue
1218                all_tracks.append(track)
1219                total_track_count += 1
1220                track_id = track.item_id.split(RADIO_TRACK_ID_SEP, 1)[0]
1221                if first_track_id_this_batch is None:
1222                    first_track_id_this_batch = track_id
1223            if first_track_id_this_batch is not None:
1224                wave.last_track_id = first_track_id_this_batch
1225            if (
1226                first_track_id_this_batch is None
1227                or not batch_id
1228                or not yandex_tracks
1229                or total_track_count >= effective_limit
1230            ):
1231                break
1232
1233        if last_batch_id and total_track_count < max_tracks_config:
1234            all_tracks.append(
1235                BrowseFolder(
1236                    item_id="next",
1237                    provider=self.instance_id,
1238                    path=f"{path.rstrip('/')}/next",
1239                    name="Load more",
1240                    translation_key="load_more",
1241                    is_playable=False,
1242                )
1243            )
1244        return all_tracks
1245
1246    def _parse_my_wave_track(
1247        self,
1248        yt: Any,
1249        seen_ids: set[str],
1250        *,
1251        station_key: str = ROTOR_STATION_MY_WAVE,
1252    ) -> Track | None:
1253        """
1254        Parse a Yandex track into a My Wave Track with composite item_id.
1255
1256        Extracts the track_id, checks for duplicates in the seen_ids set,
1257        sets composite item_id (track_id@station_key) and updates
1258        provider_mappings. `station_key` is the key in `_wave_states` under
1259        which the matching session lives; for preset modes it carries a
1260        `#preset` suffix so `on_played`/`on_streamed` find the right session.
1261
1262        Callers using shared state must hold the My Wave state lock.
1263
1264        :param yt: Yandex track object from rotor station response.
1265        :param seen_ids: Set of already-seen track IDs to check and update.
1266        :param station_key: Station key to embed in the composite item_id.
1267            Defaults to the plain My Wave station.
1268        :return: Parsed Track with composite item_id, or None if duplicate/invalid.
1269        """
1270        try:
1271            t = parse_track(self, yt)
1272        except InvalidDataError as err:
1273            self.logger.debug("Error parsing My Wave track: %s", err)
1274            return None
1275
1276        track_id = str(yt.id) if hasattr(yt, "id") and yt.id else getattr(yt, "track_id", None)
1277        if not track_id:
1278            return t
1279
1280        if track_id in seen_ids:
1281            self.logger.debug("Skipping duplicate My Wave track: %s", track_id)
1282            return None
1283
1284        seen_ids.add(track_id)
1285        t.item_id = f"{track_id}{RADIO_TRACK_ID_SEP}{station_key}"
1286        for pm in t.provider_mappings:
1287            if pm.provider_instance == self.instance_id:
1288                pm.item_id = t.item_id
1289                break
1290        return t
1291
1292    @use_cache(3600, allow_expired_cache=True)
1293    async def _get_valid_tags_for_category(self, category: str) -> list[str]:
1294        """
1295        Return tags for a category by combining hardcoded + landing-discovered.
1296
1297        Trusts the hardcoded ``TAG_CATEGORY_*`` lists (evergreen Yandex
1298        categories) and the landing API output (Yandex returns landing tags
1299        only when they have playlists). No per-tag runtime validation: that
1300        machinery was a parallel ``asyncio.gather`` over
1301        ``get_tag_playlists`` for every tag and tripped Yandex's edge
1302        per-endpoint concurrency limit on first browse — captcha within
1303        ~460ms of the burst. If a tag turns out to be empty at click time,
1304        ``_get_tag_playlists_as_browse`` already renders an empty folder.
1305
1306        :param category: Category name ('mood', 'activity', 'era', 'genres').
1307        :return: List of tag slugs (hardcoded order preserved, landing tags appended).
1308        """
1309        category_lists: dict[str, list[str]] = {
1310            "mood": list(TAG_CATEGORY_MOOD),
1311            "activity": list(TAG_CATEGORY_ACTIVITY),
1312            "era": list(TAG_CATEGORY_ERA),
1313            "genres": list(TAG_CATEGORY_GENRES),
1314        }
1315        tags = category_lists.get(category, [])
1316        try:
1317            landing_tags = await self.client.get_landing_tags()
1318            for slug, _title in landing_tags:
1319                cat = TAG_SLUG_CATEGORY.get(slug, "mood")
1320                if cat == category and slug not in tags:
1321                    tags.append(slug)
1322        except Exception as err:
1323            self.logger.debug("Landing tag discovery failed: %s", err)
1324        return tags
1325
1326    @use_cache(3600, allow_expired_cache=True)
1327    async def _get_discovered_tags(self, locale: str) -> list[tuple[str, str, str | None]]:
1328        """
1329        Return all browse-able tags: hardcoded (non-seasonal) + landing-discovered.
1330
1331        Same rationale as :meth:`_get_valid_tags_for_category` — runtime
1332        validation removed to avoid the per-endpoint concurrency burst that
1333        triggered Yandex captcha. The locale parameter is part of the cache
1334        key so locale changes invalidate the cached landing titles.
1335
1336        :param locale: Current metadata locale (used as part of cache key).
1337        :return: List of (slug, English name, translation_key) tuples in
1338            hardcoded-then-discovered order. Landing-discovered tags carry their
1339            (already localized) API title and no translation_key.
1340        """
1341        all_tags: dict[str, tuple[str, str | None]] = {}
1342        for slug, cat in TAG_SLUG_CATEGORY.items():
1343            if cat != "seasonal":
1344                all_tags[slug] = self._media_label("folder", _media_label_key(slug), slug.title())
1345        try:
1346            landing_tags = await self.client.get_landing_tags()
1347            for slug, title in landing_tags:
1348                if slug not in all_tags:
1349                    all_tags[slug] = (title, None)
1350        except Exception as err:
1351            self.logger.debug("Failed to discover tags from landing API: %s", err)
1352        return [(slug, name, translation_key) for slug, (name, translation_key) in all_tags.items()]
1353
1354    async def _get_discovered_tag_slugs(self) -> set[str]:
1355        """
1356        Get set of all valid tag slugs (cached).
1357
1358        :return: Set of tag slug strings that have playlists.
1359        """
1360        discovered = await self._get_discovered_tags(self.mass.metadata.locale or "en_US")
1361        return {slug for slug, _name, _key in discovered}
1362
1363    async def _browse_for_you(
1364        self, path: str, path_parts: list[str]
1365    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1366        """
1367        Browse «For You» folder — shows Picks and Mixes sub-folders.
1368
1369        :param path: Full browse path.
1370        :param path_parts: Split path parts after ://.
1371        :return: List of sub-folders (Picks, Mixes).
1372        """
1373        # Strip the for_you segment to build child paths that route to picks/mixes
1374        # Path format: ...//for_you  → child paths should be ...//picks, ...//mixes
1375        # We build base from the root (before for_you) by dropping the last segment.
1376        base_parts = path.split("//", 1)
1377        root_base = (base_parts[0] + "//") if len(base_parts) > 1 else path.rstrip("/") + "/"
1378
1379        if len(path_parts) == 1:
1380            return [
1381                BrowseFolder(
1382                    item_id="picks",
1383                    provider=self.instance_id,
1384                    path=f"{root_base}picks",
1385                    name="Picks",
1386                    translation_key="picks",
1387                    is_playable=False,
1388                ),
1389                BrowseFolder(
1390                    item_id="mixes",
1391                    provider=self.instance_id,
1392                    path=f"{root_base}mixes",
1393                    name="Mixes",
1394                    translation_key="mixes",
1395                    is_playable=False,
1396                ),
1397            ]
1398        # Deeper path: delegate to picks or mixes handler via canonical paths
1399        return await super().browse(path)
1400
1401    async def _browse_collection(
1402        self, path: str
1403    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1404        """
1405        Browse «Collection» folder — shows library sub-folders (tracks/artists/albums/playlists).
1406
1407        Child ``path`` is nested (``…/collection/tracks``) so MA's "back"
1408        button lands on this listing instead of the provider root. The
1409        dispatcher then strips the ``collection/`` prefix and hands off to
1410        core's default library handler.
1411
1412        :param path: Full browse path.
1413        :return: List of library sub-folders.
1414        """
1415        base = path if path.endswith("/") else f"{path}/"
1416
1417        folders: list[BrowseFolder] = []
1418        for feature, sub_id, label_key, is_playable in _COLLECTION_SUBFOLDERS:
1419            if feature not in self.supported_features:
1420                continue
1421            name, translation_key = self._media_label(
1422                "folder", label_key, label_key.replace("_", " ").title()
1423            )
1424            folders.append(
1425                BrowseFolder(
1426                    item_id=sub_id,
1427                    provider=self.instance_id,
1428                    path=f"{base}{sub_id}",
1429                    name=name,
1430                    translation_key=translation_key,
1431                    is_playable=is_playable,
1432                )
1433            )
1434        return folders
1435
1436    async def _browse_pins(self) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1437        """
1438        Browse user's pinned items (artists/albums/playlists from Yandex Pins).
1439
1440        Resolves each pin to its full media item via existing single-item lookups.
1441        Wave pins are skipped — MA has no native concept for them.
1442
1443        :return: List of resolved media items.
1444        """
1445        pins_list = await self.client.get_pins()
1446        pins = getattr(pins_list, "pins", None) if pins_list else None
1447        if not pins:
1448            return []
1449
1450        items: list[MediaItemType] = []
1451        for pin in pins:
1452            pin_type = getattr(pin, "type", None)
1453            data = getattr(pin, "data", None)
1454            if data is None:
1455                continue
1456            try:
1457                if pin_type == "artist_item" and getattr(data, "id", None) is not None:
1458                    items.append(await self.get_artist(str(data.id)))
1459                elif pin_type == "album_item" and getattr(data, "id", None) is not None:
1460                    items.append(await self.get_album(str(data.id)))
1461                elif pin_type == "playlist_item":
1462                    uid = getattr(data, "uid", None)
1463                    kind = getattr(data, "kind", None)
1464                    if uid is not None and kind is not None:
1465                        items.append(await self.get_playlist(f"{uid}:{kind}"))
1466            except (MediaNotFoundError, InvalidDataError) as err:
1467                self.logger.debug("Skipping pin %s: %s", pin_type, err)
1468        return items
1469
1470    async def _browse_history(self) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1471        """
1472        Browse user's recent listening history (flattened across days).
1473
1474        Collects ``track_id`` values from each history entry's ``item_id``
1475        sub-object (``full_model`` is not populated by the current API
1476        response — MarshalX exposes the IDs separately), dedupes, and
1477        batch-resolves them via ``get_tracks`` so the returned Track objects
1478        carry full artist/album/cover metadata.
1479
1480        Entries without a resolvable ``track_id`` (e.g. album-only context
1481        rows) are skipped silently. Order is preserved — most recent first —
1482        by collecting unique IDs in response order into ``ordered_ids``,
1483        then rebuilding the final list by iterating ``ordered_ids`` and
1484        looking up each batch-fetched track in an id→track map.
1485
1486        :return: List of recently played Track items.
1487        """
1488        history = await self.client.get_music_history()
1489        tabs = getattr(history, "history_tabs", None) if history else None
1490        if not tabs:
1491            return []
1492
1493        seen_track_ids: set[str] = set()
1494        ordered_ids: list[str] = []
1495        for tab in tabs:
1496            for group in getattr(tab, "items", None) or []:
1497                for hist_item in getattr(group, "tracks", None) or []:
1498                    if getattr(hist_item, "type", None) != "track":
1499                        continue
1500                    item_id_obj = getattr(getattr(hist_item, "data", None), "item_id", None)
1501                    track_key: str | None = None
1502                    if isinstance(item_id_obj, dict):
1503                        track_key = item_id_obj.get("track_id") or item_id_obj.get("id")
1504                    else:
1505                        track_key = getattr(item_id_obj, "track_id", None) or getattr(
1506                            item_id_obj, "id", None
1507                        )
1508                    if not track_key:
1509                        continue
1510                    track_key = str(track_key)
1511                    if track_key in seen_track_ids:
1512                        continue
1513                    seen_track_ids.add(track_key)
1514                    ordered_ids.append(track_key)
1515
1516        if not ordered_ids:
1517            return []
1518
1519        try:
1520            fetched = await self.client.get_tracks(ordered_ids)
1521        except ResourceTemporarilyUnavailable as err:
1522            self.logger.warning("Failed to hydrate history tracks: %s", err)
1523            return []
1524
1525        by_id = {str(t.id): t for t in fetched if getattr(t, "id", None) is not None}
1526        tracks: list[Track] = []
1527        for tid in ordered_ids:
1528            yt = by_id.get(tid)
1529            if yt is None:
1530                continue
1531            try:
1532                tracks.append(parse_track(self, yt))
1533            except InvalidDataError as err:
1534                self.logger.debug("Skipping history track %s: %s", tid, err)
1535        return tracks
1536
1537    async def _browse_picks(
1538        self, path: str, path_parts: list[str]
1539    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1540        """
1541        Browse picks folder using hardcoded tags validated against the API.
1542
1543        Tags are sourced from hardcoded category lists and landing API discovery,
1544        then validated via client.tags() to ensure they have playlists.
1545        Only categories with at least one valid tag are shown.
1546
1547        :param path: Full browse path.
1548        :param path_parts: Split path parts after ://.
1549        :return: List of folders or playlists.
1550        """
1551        base = path.rstrip("/") + "/"
1552
1553        # Get validated tags
1554        discovered = await self._get_discovered_tags(self.mass.metadata.locale or "en_US")
1555
1556        # Categorize valid tags, carrying each tag's (slug, English name, translation_key)
1557        categorized: dict[str, list[tuple[str, str, str | None]]] = {}
1558        for slug, name, translation_key in discovered:
1559            cat = TAG_SLUG_CATEGORY.get(slug, "mood")
1560            # Skip seasonal tags — they belong in mixes, not picks
1561            if cat == "seasonal":
1562                continue
1563            categorized.setdefault(cat, []).append((slug, name, translation_key))
1564
1565        # Sort tags within each category by preferred order
1566        for cat, cat_tags in categorized.items():
1567            order = TAG_CATEGORY_ORDER.get(cat, [])
1568            order_map = {s: i for i, s in enumerate(order)}
1569            cat_tags.sort(key=lambda t: order_map.get(t[0], len(order)))
1570
1571        # picks/ - show category folders (only those with valid tags)
1572        if len(path_parts) == 1:
1573            category_display_order = ["mood", "activity", "era", "genres"]
1574            folders: list[BrowseFolder] = []
1575            for cat in category_display_order:
1576                if cat in categorized:
1577                    name, translation_key = self._media_label("folder", cat, cat.title())
1578                    folders.append(
1579                        BrowseFolder(
1580                            item_id=cat,
1581                            provider=self.instance_id,
1582                            path=f"{base}{cat}",
1583                            name=name,
1584                            translation_key=translation_key,
1585                            is_playable=False,
1586                        )
1587                    )
1588            # Show any extra categories not in the standard order
1589            for cat in categorized:
1590                if cat not in category_display_order:
1591                    name, translation_key = self._media_label("folder", cat, cat.title())
1592                    folders.append(
1593                        BrowseFolder(
1594                            item_id=cat,
1595                            provider=self.instance_id,
1596                            path=f"{base}{cat}",
1597                            name=name,
1598                            translation_key=translation_key,
1599                            is_playable=False,
1600                        )
1601                    )
1602            return folders
1603
1604        category: str | None = path_parts[1] if len(path_parts) > 1 else None
1605        tag: str | None = path_parts[2] if len(path_parts) > 2 else None
1606
1607        self.logger.debug(
1608            "Browse picks: path=%s, category=%s, tag=%s",
1609            path,
1610            category,
1611            tag,
1612        )
1613
1614        # picks/category/ - show valid tag folders for this category
1615        if category and not tag:
1616            category_tags = categorized.get(category, [])
1617            folders = []
1618            for slug, name, translation_key in category_tags:
1619                folders.append(
1620                    BrowseFolder(
1621                        item_id=slug,
1622                        provider=self.instance_id,
1623                        path=f"{base}{slug}",
1624                        name=name,
1625                        translation_key=translation_key,
1626                        is_playable=False,
1627                    )
1628                )
1629            self.logger.debug("Returning %d tag folders for category %s", len(folders), category)
1630            return folders
1631
1632        # picks/category/tag - show playlists for the tag
1633        if tag:
1634            discovered_slugs = {slug for slug, _name, _key in discovered}
1635            if tag in discovered_slugs:
1636                self.logger.debug("Fetching playlists for tag: %s", tag)
1637                return await self._get_tag_playlists_as_browse(tag)
1638
1639        self.logger.debug("No match found, returning empty list")
1640        return []
1641
1642    async def _browse_mixes(
1643        self, path: str, path_parts: list[str]
1644    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1645        """
1646        Browse mixes folder (seasonal collections) using hardcoded tags.
1647
1648        Renders every seasonal tag from ``TAG_MIXES`` unconditionally. The
1649        old per-tag validation fired a ``Semaphore(5)+gather`` of
1650        ``get_tag_playlists`` calls and tripped Yandex's per-endpoint
1651        concurrency limit on first browse. If a season ends up empty at
1652        click time, ``_get_tag_playlists_as_browse`` already returns an
1653        empty folder.
1654
1655        :param path: Full browse path.
1656        :param path_parts: Split path parts after ://.
1657        :return: List of folders or playlists.
1658        """
1659        base = path.rstrip("/") + "/"
1660
1661        # mixes/ - show seasonal folders
1662        if len(path_parts) == 1:
1663            folders: list[BrowseFolder] = []
1664            for t in TAG_MIXES:
1665                name, translation_key = self._media_label("folder", t, t.title())
1666                folders.append(
1667                    BrowseFolder(
1668                        item_id=t,
1669                        provider=self.instance_id,
1670                        path=f"{base}{t}",
1671                        name=name,
1672                        translation_key=translation_key,
1673                        is_playable=False,
1674                    )
1675                )
1676            return folders
1677
1678        # mixes/tag - show playlists for the tag
1679        tag = path_parts[1] if len(path_parts) > 1 else None
1680        if tag and tag in TAG_MIXES:
1681            return await self._get_tag_playlists_as_browse(tag)
1682
1683        return []
1684
1685    def _get_wave_state(self, station_id: str) -> _WaveState:
1686        """
1687        Get or create per-station wave state.
1688
1689        :param station_id: Rotor station ID (e.g. 'genre:rock', 'mood:chill').
1690        :return: _WaveState instance for this station.
1691        """
1692        return self._wave_states.setdefault(station_id, _WaveState())
1693
1694    async def _send_wave_feedback(
1695        self,
1696        wave: _WaveState,
1697        station_id: str,
1698        event_type: str,
1699        *,
1700        track_id: str | None = None,
1701        total_played_seconds: int | None = None,
1702    ) -> bool:
1703        """
1704        Route rotor feedback to the session endpoint.
1705
1706        Requires an active ``wave.session_id`` — rotor feedback is only
1707        meaningful inside the session it originated from. The legacy
1708        stations-based endpoint (``/rotor/station/{id}/feedback``) is no
1709        longer reachable (returns 404 "not-found"), so when there's no
1710        session we skip silently rather than spamming the log.
1711
1712        This happens when the track's composite item_id was parsed in a
1713        previous provider run (e.g. loaded from MA's library cache) and
1714        the corresponding session_id is not in memory any more. History
1715        reporting via ``play_audio`` still works in that case — only the
1716        rotor recommendation signal is lost.
1717
1718        :param wave: Station state carrying session_id + batch_id.
1719        :param station_id: Rotor station ID (used only for logging here).
1720        :param event_type: Rotor event type (radioStarted, trackStarted, …).
1721        :param track_id: Yandex track ID the event refers to.
1722        :param total_played_seconds: Seconds played (trackFinished / skip only).
1723        :return: True if the feedback POST succeeded, False when skipped.
1724        """
1725        if not wave.session_id:
1726            self.logger.debug(
1727                "Skipping rotor feedback %s for %s: no active session",
1728                event_type,
1729                station_id,
1730            )
1731            return False
1732        return await self.client.rotor_session_feedback(
1733            wave.session_id,
1734            event_type,
1735            track_id=track_id,
1736            total_played_seconds=total_played_seconds,
1737            batch_id=wave.batch_id,
1738        )
1739
1740    async def _prefetch_rotor_session(self, station_key: str) -> None:
1741        """
1742        Fire-and-forget: fetch the next batch for an active wave session.
1743
1744        Called from ``on_played`` while a wave track starts playing, so by the
1745        time Music Assistant's DSTM asks for more via ``get_similar_tracks``,
1746        we already have Yandex-curated wave tracks sitting in
1747        ``wave.prefetched`` ready to serve (no extra round-trip).
1748
1749        No-op when the station has no active session yet (prefetch cannot
1750        safely create one — that requires holding the lock across the
1751        network call and would stall readers), or when the buffer already
1752        has items (avoids burning rate limit).
1753
1754        Three-phase lock discipline so the network round-trip does not
1755        block browse / drain paths that share the lock:
1756
1757          1. Acquire, verify session + empty buffer, snapshot
1758             ``session_id`` and ``last_track_id``, release.
1759          2. Call ``client.rotor_session_tracks`` **directly** (no
1760             ``_fetch_rotor_session_batch``) — that helper mutates shared
1761             state (session creation, batch_id write) and would race with
1762             other callers now that we hold no lock. The raw client call
1763             only reads the arguments we pass in.
1764          3. Re-acquire, verify the session hasn't been recycled and the
1765             buffer is still empty, then ``extend``.
1766
1767        :param station_key: Station key whose state to top up.
1768        """
1769        wave = self._wave_states.get(station_key)
1770        if wave is None:
1771            return
1772
1773        async with wave.lock:
1774            if wave.session_id is None or wave.prefetched:
1775                return
1776            session_id = wave.session_id
1777            cursor = wave.last_track_id
1778
1779        if not cursor:
1780            return  # No anchor for the next batch yet; try again later.
1781
1782        tracks, _ = await self.client.rotor_session_tracks(session_id, current_track_id=str(cursor))
1783        if not tracks:
1784            return
1785
1786        async with wave.lock:
1787            # Another task could have restarted the session or filled the
1788            # buffer while we were awaiting the network call; bail in both
1789            # cases to avoid stale extends.
1790            if wave.session_id != session_id or wave.prefetched:
1791                return
1792            wave.prefetched.extend(tracks)
1793
1794    async def _fetch_rotor_session_batch(
1795        self, wave: _WaveState, station_id: str
1796    ) -> tuple[list[YandexTrack], str | None]:
1797        """
1798        Fetch the next rotor-session batch for any station.
1799
1800        On first call (wave.session_id is None), starts a new rotor session
1801        and records session_id + batch_id on the wave state. On subsequent
1802        calls, paginates via rotor_session_tracks using wave.last_track_id.
1803
1804        If station_id carries a wave-mode suffix (e.g. "user:onyourwave#discover"),
1805        the suffix maps to a preset in WAVE_MODE_PRESETS and its settings are
1806        merged with wave.settings (wave.settings wins on key conflict). The
1807        base station ID (before "#") is what actually goes to Yandex.
1808
1809        :param wave: The _WaveState for this station (persists across calls).
1810        :param station_id: Rotor station key (may include a "#preset" suffix).
1811        :return: Tuple of (list of yandex tracks, batch_id or None).
1812        """
1813        # Session-creation path: no session yet, or we have a session but no
1814        # cursor yet (`tracks` with an empty queue returns a hard-to-debug
1815        # empty batch — starting a fresh session is the same latency but
1816        # actually yields tracks).
1817        if wave.session_id is None or not wave.last_track_id:
1818            base_station, preset_settings = _split_wave_mode(station_id)
1819            merged = {**preset_settings, **wave.settings}
1820            session_id, tracks, batch_id = await self.client.rotor_session_new(
1821                base_station, settings=merged or None
1822            )
1823            if session_id:
1824                wave.session_id = session_id
1825        else:
1826            tracks, batch_id = await self.client.rotor_session_tracks(
1827                wave.session_id, current_track_id=str(wave.last_track_id)
1828            )
1829        if batch_id:
1830            wave.batch_id = batch_id
1831        return (tracks, batch_id)
1832
1833    async def _browse_waves(
1834        self, path: str, path_parts: list[str]
1835    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1836        """
1837        Browse waves folder (rotor stations by genre/mood/activity/epoch/local).
1838
1839        Fetches available stations from the Yandex rotor API and groups them by category.
1840
1841        :param path: Full browse path.
1842        :param path_parts: Split path parts after ://.
1843        :return: List of folders or tracks.
1844        """
1845        base = path.rstrip("/") + "/"
1846
1847        locale = (self.mass.metadata.locale or "en_US").lower()
1848        language = "ru" if locale.startswith("ru") else "en"
1849
1850        all_stations = await self.client.get_wave_stations(language)
1851
1852        # Group stations by category, preserving image_url
1853        categorized: dict[str, list[tuple[str, str, str | None]]] = {}
1854        for station_id, cat_key, station_name, image_url in all_stations:
1855            categorized.setdefault(cat_key, []).append((station_id, station_name, image_url))
1856
1857        # waves/ — show category folders
1858        if len(path_parts) == 1:
1859            folders: list[BrowseFolder] = []
1860            # Personalized "My Waves" first — only show if dashboard returns stations
1861            dashboard_stations = await self._get_dashboard_stations_cached()
1862            if dashboard_stations:
1863                name, translation_key = self._media_label("folder", MY_WAVES_FOLDER_ID, "Personal")
1864                folders.append(
1865                    BrowseFolder(
1866                        item_id=MY_WAVES_FOLDER_ID,
1867                        provider=self.instance_id,
1868                        path=f"{base}{MY_WAVES_FOLDER_ID}",
1869                        name=name,
1870                        translation_key=translation_key,
1871                        is_playable=False,
1872                    )
1873                )
1874            # Featured Waves — only show if landing-blocks/waves returns data
1875            waves_landing = await self._get_waves_landing_cached()
1876            if waves_landing:
1877                name, translation_key = self._media_label(
1878                    "folder", WAVES_LANDING_FOLDER_ID, "Featured Waves"
1879                )
1880                folders.append(
1881                    BrowseFolder(
1882                        item_id=WAVES_LANDING_FOLDER_ID,
1883                        provider=self.instance_id,
1884                        path=f"{base}{WAVES_LANDING_FOLDER_ID}",
1885                        name=name,
1886                        translation_key=translation_key,
1887                        is_playable=False,
1888                    )
1889                )
1890            for cat in WAVE_CATEGORY_DISPLAY_ORDER:
1891                if cat in categorized:
1892                    name, translation_key = self._media_label("folder", cat, cat.title())
1893                    folders.append(
1894                        BrowseFolder(
1895                            item_id=cat,
1896                            provider=self.instance_id,
1897                            path=f"{base}{cat}",
1898                            name=name,
1899                            translation_key=translation_key,
1900                            is_playable=False,
1901                        )
1902                    )
1903            # Append any categories returned by API that aren't in the predefined order
1904            for cat in categorized:
1905                if cat not in WAVE_CATEGORY_DISPLAY_ORDER:
1906                    name, translation_key = self._media_label("folder", cat, cat.title())
1907                    folders.append(
1908                        BrowseFolder(
1909                            item_id=cat,
1910                            provider=self.instance_id,
1911                            path=f"{base}{cat}",
1912                            name=name,
1913                            translation_key=translation_key,
1914                            is_playable=False,
1915                        )
1916                    )
1917            return folders
1918
1919        category: str | None = path_parts[1] if len(path_parts) > 1 else None
1920        tag: str | None = path_parts[2] if len(path_parts) > 2 else None
1921
1922        # waves/my_waves/ — show personalized stations from dashboard
1923        if category == MY_WAVES_FOLDER_ID and not tag:
1924            return await self._browse_my_waves_stations(path)
1925
1926        # waves/waves_landing/... — redirect to Featured Waves browse
1927        if category == WAVES_LANDING_FOLDER_ID:
1928            return await self._browse_waves_landing(path, path_parts[1:])
1929
1930        # waves/my_waves/<tag>[/next] — play a specific personal station
1931        # The full station_id has format "genre:allrock", not "my_waves:allrock".
1932        # Resolve by matching against dashboard stations cache.
1933        if category == MY_WAVES_FOLDER_ID and tag:
1934            dashboard_stations = await self._get_dashboard_stations_cached()
1935            for sid, _, _ in dashboard_stations:
1936                sid_tag = sid.split(":", 1)[1] if ":" in sid else sid
1937                if sid_tag == tag:
1938                    return await self._browse_wave_station(sid, path=path)
1939            # Fallback: try tag as direct station_id (e.g. "genre:allrock" passed verbatim)
1940            if ":" in tag:
1941                return await self._browse_wave_station(tag, path=path)
1942            return []
1943
1944        # waves/<category>/ — show station folders with artwork
1945        if category and not tag:
1946            cat_stations = categorized.get(category, [])
1947            folders = []
1948            for station_id, station_name, image_url in cat_stations:
1949                tag_part = station_id.split(":", 1)[1] if ":" in station_id else station_id
1950                station_image: MediaItemImage | None = None
1951                if image_url:
1952                    station_image = MediaItemImage(
1953                        type=ImageType.THUMB,
1954                        path=image_url,
1955                        provider=self.instance_id,
1956                        remotely_accessible=True,
1957                    )
1958                folders.append(
1959                    BrowseFolder(
1960                        item_id=station_id,
1961                        provider=self.instance_id,
1962                        path=f"{base}{tag_part}",
1963                        name=station_name,
1964                        is_playable=True,
1965                        image=station_image,
1966                    )
1967                )
1968            return folders
1969
1970        # waves/<category>/<tag>[/next] — stream tracks from rotor station
1971        if category and tag:
1972            station_id = f"{category}:{tag}"
1973            return await self._browse_wave_station(station_id, path=path)
1974
1975        return []
1976
1977    @use_cache(600, allow_expired_cache=True)
1978    async def _get_dashboard_stations_cached(self) -> list[tuple[str, str, str | None]]:
1979        """
1980        Get personalized dashboard stations, cached for 10 minutes.
1981
1982        :return: List of (station_id, name, image_url) tuples.
1983        """
1984        return await self.client.get_dashboard_stations()
1985
1986    async def _browse_my_waves_stations(self, path: str) -> list[BrowseFolder]:
1987        """
1988        Browse personalized wave stations from rotor/stations/dashboard.
1989
1990        Names are resolved from the non-personalized station list so that
1991        stations show their actual genre/mood name (e.g. "Рок") rather than
1992        the generic "Моя волна" label that the dashboard API returns.
1993
1994        :param path: Full browse path (used to build sub-paths).
1995        :return: List of playable BrowseFolder items, one per station.
1996        """
1997        stations = await self._get_dashboard_stations_cached()
1998
1999        # Build a name map from the non-personalized list for proper localized names.
2000        locale = (self.mass.metadata.locale or "en_US").lower()
2001        language = "ru" if locale.startswith("ru") else "en"
2002        all_stations = await self.client.get_wave_stations(language)
2003        station_name_map: dict[str, str] = {sid: name for sid, _, name, _ in all_stations}
2004
2005        base = path.rstrip("/") + "/"
2006        folders: list[BrowseFolder] = []
2007        for station_id, fallback_name, image_url in stations:
2008            # Use full station_id (e.g. "genre:rock") in path to avoid collisions
2009            # when two stations share the same tag but differ by category.
2010            # The routing fallback (if ":" in tag) handles this correctly.
2011            name = station_name_map.get(station_id, fallback_name)
2012            station_image: MediaItemImage | None = None
2013            if image_url:
2014                station_image = MediaItemImage(
2015                    type=ImageType.THUMB,
2016                    path=image_url,
2017                    provider=self.instance_id,
2018                    remotely_accessible=True,
2019                )
2020            folders.append(
2021                BrowseFolder(
2022                    item_id=station_id,
2023                    provider=self.instance_id,
2024                    path=f"{base}{station_id}",
2025                    name=name,
2026                    is_playable=True,
2027                    image=station_image,
2028                )
2029            )
2030        return folders
2031
2032    async def _browse_wave_station(
2033        self, station_id: str, path: str = ""
2034    ) -> list[Track | BrowseFolder]:
2035        """
2036        Browse a rotor wave station and return tracks.
2037
2038        Fetches tracks from the rotor station, deduplicates within the current session,
2039        and sends radioStarted feedback on first call. Appends a "Load more" BrowseFolder
2040        at the end so MA can continue fetching the next batch automatically (radio mode).
2041
2042        :param station_id: Rotor station ID (e.g. 'genre:rock', 'mood:chill').
2043        :param path: Current browse path, used to construct the "Load more" next path.
2044        :return: List of Track objects with composite item_id (track_id@station_id),
2045                 followed by a "Load more" BrowseFolder if more tracks are available.
2046        """
2047        state = self._get_wave_state(station_id)
2048        async with state.lock:
2049            max_tracks = int(
2050                self.config.get_value(CONF_MY_WAVE_MAX_TRACKS) or 150  # type: ignore[arg-type]
2051            )
2052
2053            self.logger.debug(
2054                "Browse wave station: station_id=%s path=%s last_track_id=%s session=%s",
2055                station_id,
2056                path,
2057                state.last_track_id,
2058                state.session_id,
2059            )
2060            # Tagged stations (genre:*, mood:*, activity:*, epoch:*) accept the
2061            # same /rotor/session/* endpoint as user:onyourwave / track:{id},
2062            # verified against the live Yandex API. Reuse the session helper so
2063            # batch_id + session_id stay anchored across browse/play/feedback.
2064            yandex_tracks, _ = await self._fetch_rotor_session_batch(state, station_id)
2065
2066            if not state.radio_started_sent and yandex_tracks:
2067                sent = await self._send_wave_feedback(state, station_id, "radioStarted")
2068                if sent:
2069                    state.radio_started_sent = True
2070
2071            tracks: list[Track] = []
2072            first_track_id: str | None = None
2073            for yt in yandex_tracks:
2074                if len(state.seen_track_ids) >= max_tracks:
2075                    break
2076                track = self._parse_my_wave_track(yt, state.seen_track_ids)
2077                if track is None:
2078                    continue
2079                # Override station_id in composite item_id to reflect this specific station
2080                old_item_id = track.item_id
2081                track_id = old_item_id.split(RADIO_TRACK_ID_SEP, 1)[0]
2082                track.item_id = f"{track_id}{RADIO_TRACK_ID_SEP}{station_id}"
2083                # Keep provider mappings in sync with the new item_id
2084                for pm in getattr(track, "provider_mappings", []):
2085                    if (
2086                        getattr(pm, "item_id", None) == old_item_id
2087                        and getattr(pm, "provider_instance", None) == self.instance_id
2088                    ):
2089                        pm.item_id = track.item_id
2090                if first_track_id is None:
2091                    first_track_id = track_id
2092                tracks.append(track)
2093
2094            if first_track_id is not None:
2095                state.last_track_id = first_track_id
2096
2097            self.logger.debug(
2098                "Wave station %s returned %d tracks: %s",
2099                station_id,
2100                len(tracks),
2101                [t.item_id.split(RADIO_TRACK_ID_SEP, 1)[0] for t in tracks[:5]],
2102            )
2103            result: list[Track | BrowseFolder] = list(tracks)
2104
2105            # Append "Load more" sentinel so MA knows to call browse again for next batch.
2106            # This mirrors the My Wave mechanism and enables continuous radio playback.
2107            if tracks and len(state.seen_track_ids) < max_tracks and path:
2108                # Append /next to the current path (same pattern as _browse_my_wave).
2109                # This makes each "Load more" path unique (e.g. /next/next/next...)
2110                # so MA never serves a cached result for subsequent presses.
2111                result.append(
2112                    BrowseFolder(
2113                        item_id="next",
2114                        provider=self.instance_id,
2115                        path=f"{path.rstrip('/')}/next",
2116                        name="Load more",
2117                        translation_key="load_more",
2118                        is_playable=False,
2119                    )
2120                )
2121
2122            return result
2123
2124    @staticmethod
2125    def _extract_wave_item_cover(item: dict[str, Any]) -> tuple[str | None, str | None]:
2126        """
2127        Extract cover URI and background color from a wave/mix item.
2128
2129        Accepts both camelCase (``compactImageUrl`` — what /landing-blocks/
2130        actually returns) and snake_case (``compact_image_url`` — retained
2131        for safety if MarshalX ever normalises the payload).
2132
2133        :param item: Wave or mix item dict from the API.
2134        :return: (cover_uri, bg_color) tuple where bg_color is a hex string or None.
2135        """
2136        agent_uri = item.get("agent", {}).get("cover", {}).get("uri", "")
2137        cover_uri = agent_uri or item.get("compactImageUrl") or item.get("compact_image_url")
2138        bg_color = item.get("colors", {}).get("average")
2139        return cover_uri, bg_color
2140
2141    @use_cache(3600, allow_expired_cache=True)
2142    async def _get_mixes_waves_cached(self) -> list[dict[str, Any]] | None:
2143        """
2144        Get AI Wave Set data from /landing-blocks/mixes-waves, cached for 1 hour.
2145
2146        :return: List of mix category dicts from the API, or None on error.
2147        """
2148        return await self.client.get_mixes_waves()
2149
2150    @use_cache(3600, allow_expired_cache=True)
2151    async def _get_waves_landing_cached(self) -> list[dict[str, Any]] | None:
2152        """
2153        Get Featured Waves data from /landing-blocks/waves, cached for 1 hour.
2154
2155        :return: List of wave category dicts from the API, or None on error.
2156        """
2157        return await self.client.get_waves_landing()
2158
2159    async def _browse_waves_landing(
2160        self, path: str, path_parts: list[str]
2161    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
2162        """
2163        Browse Featured Waves (from /landing-blocks/waves).
2164
2165        :param path: Full browse path.
2166        :param path_parts: Split path parts after ://.
2167        :return: List of folders or tracks.
2168        """
2169        waves_data = await self._get_waves_landing_cached()
2170        return await self._browse_wave_categories(
2171            path, path_parts, waves_data or [], WAVES_LANDING_FOLDER_ID
2172        )
2173
2174    async def _browse_wave_categories(
2175        self,
2176        path: str,
2177        path_parts: list[str],
2178        categories_data: list[dict[str, Any]],
2179        id_prefix: str,
2180    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
2181        """
2182        Browse wave-like category folders and their station items.
2183
2184        Shared logic for both 'my_waves_set' browse trees:
2185        - Level 1 (e.g. my_waves_set/): category folders
2186        - Level 2 (e.g. my_waves_set/ai-sets/): playable station folders with artwork
2187        - Level 3+ (e.g. my_waves_set/ai-sets/genre:rock[/next]): track listing
2188
2189        :param path: Full browse path.
2190        :param path_parts: Split path parts after ://.
2191        :param categories_data: List of category dicts from the API.
2192        :param id_prefix: Prefix for BrowseFolder item_id (e.g. 'my_waves_set').
2193        :return: List of folders or tracks.
2194        """
2195        base = path.rstrip("/") + "/"
2196
2197        if not categories_data:
2198            return []
2199
2200        # Level 1 → category folders
2201        if len(path_parts) == 1:
2202            folders: list[BrowseFolder] = []
2203            for wave_category in categories_data:
2204                cat_id = wave_category.get("id", "")
2205                cat_title = wave_category.get("title", "")
2206                items = wave_category.get("items", [])
2207                if not items or not cat_id:
2208                    continue
2209                display_name = cat_title.capitalize() if cat_title else cat_id.capitalize()
2210                folders.append(
2211                    BrowseFolder(
2212                        item_id=f"{id_prefix}_{cat_id}",
2213                        provider=self.instance_id,
2214                        path=f"{base}{cat_id}",
2215                        name=display_name,
2216                        is_playable=False,
2217                    )
2218                )
2219            return folders
2220
2221        category_id = path_parts[1] if len(path_parts) > 1 else None
2222        if not category_id:
2223            return []
2224
2225        # Level 3+ → stream tracks from rotor station
2226        if len(path_parts) > 2:
2227            station_id = path_parts[2]
2228            return await self._browse_wave_station(station_id, path=path)
2229
2230        # Level 2 → playable station folders with artwork
2231        for wave_category in categories_data:
2232            if wave_category.get("id") == category_id:
2233                items = wave_category.get("items", [])
2234                result: list[BrowseFolder] = []
2235                for item in items:
2236                    # API returns camelCase (`stationId`); keep snake_case as a
2237                    # safety net if the payload is ever normalised upstream.
2238                    station_id = item.get("stationId") or item.get("station_id") or ""
2239                    title = item.get("title", "")
2240                    if not station_id or not title:
2241                        continue
2242                    cover_uri, bg_color = self._extract_wave_item_cover(item)
2243                    image: MediaItemImage | None = None
2244                    if cover_uri:
2245                        if cover_uri.startswith("http"):
2246                            img_url: str = cover_uri.replace("%%", IMAGE_SIZE_MEDIUM)
2247                        else:
2248                            raw = get_image_url(cover_uri)
2249                            img_url = "" if raw is None else raw
2250                        if img_url:
2251                            if bg_color:
2252                                # Append bg_color as URL fragment for cache-key uniqueness.
2253                                # MA will call resolve_image() to composite the transparent PNG.
2254                                if len(self._wave_bg_colors) > 200:
2255                                    self._wave_bg_colors.clear()
2256                                img_url = f"{img_url}#{bg_color.lstrip('#')}"
2257                                self._wave_bg_colors[img_url] = bg_color
2258                            image = MediaItemImage(
2259                                type=ImageType.THUMB,
2260                                path=img_url,
2261                                provider=self.instance_id,
2262                                remotely_accessible=bg_color is None,
2263                            )
2264                    result.append(
2265                        BrowseFolder(
2266                            item_id=station_id,
2267                            provider=self.instance_id,
2268                            path=f"{base}{station_id}",
2269                            name=title,
2270                            is_playable=True,
2271                            image=image,
2272                        )
2273                    )
2274                return result
2275
2276        return []
2277
2278    async def _browse_vibe_sets(
2279        self, path: str, path_parts: list[str]
2280    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
2281        """
2282        Browse AI Wave Sets (from /landing-blocks/mixes-waves).
2283
2284        :param path: Full browse path.
2285        :param path_parts: Split path parts after ://.
2286        :return: List of folders or tracks.
2287        """
2288        mixes_data = await self._get_mixes_waves_cached()
2289        return await self._browse_wave_categories(
2290            path, path_parts, mixes_data or [], MY_WAVES_SET_FOLDER_ID
2291        )
2292
2293    @use_cache(600, allow_expired_cache=True)
2294    async def _get_tag_playlists_as_browse(
2295        self, tag_id: str
2296    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
2297        """
2298        Get playlists for a tag and return as browse items.
2299
2300        :param tag_id: Tag identifier (e.g. 'chill', '80s').
2301        :return: List of Playlist objects.
2302        """
2303        self.logger.debug("Fetching playlists for tag: %s", tag_id)
2304        playlists = await self.client.get_tag_playlists(tag_id)
2305        self.logger.debug("Got %d playlists for tag %s", len(playlists), tag_id)
2306        result: list[Playlist] = []
2307        for playlist in playlists:
2308            try:
2309                result.append(parse_playlist(self, playlist))
2310            except InvalidDataError as err:
2311                self.logger.debug("Error parsing tag playlist: %s", err)
2312        self.logger.debug("Parsed %d playlists for tag %s", len(result), tag_id)
2313        return result
2314
2315    # Search
2316
2317    @use_cache(3600 * 24, allow_expired_cache=True)
2318    async def search(
2319        self, search_query: str, media_types: list[MediaType], limit: int = 5
2320    ) -> SearchResults:
2321        """
2322        Perform search on Yandex Music.
2323
2324        :param search_query: The search query.
2325        :param media_types: List of media types to search for.
2326        :param limit: Maximum number of results per type.
2327        :return: SearchResults with found items.
2328        """
2329        result = SearchResults()
2330
2331        # Determine search type based on requested media types
2332        # Map MediaType to Yandex API search type. AUDIOBOOK has no dedicated
2333        # Yandex type — it maps to "album" and is filtered by classify_album below.
2334        type_mapping = {
2335            MediaType.TRACK: "track",
2336            MediaType.ALBUM: "album",
2337            MediaType.AUDIOBOOK: "album",
2338            MediaType.ARTIST: "artist",
2339            MediaType.PLAYLIST: "playlist",
2340            MediaType.PODCAST: "podcast",
2341        }
2342        requested_types = list(
2343            dict.fromkeys(type_mapping[mt] for mt in media_types if mt in type_mapping)
2344        )
2345
2346        # Use specific type if only one requested, otherwise search all
2347        search_type = requested_types[0] if len(requested_types) == 1 else "all"
2348
2349        search_result = await self.client.search(search_query, search_type=search_type)
2350        if not search_result:
2351            return result
2352
2353        # Parse tracks
2354        if MediaType.TRACK in media_types and search_result.tracks:
2355            for track in search_result.tracks.results[:limit]:
2356                try:
2357                    result.tracks = [*result.tracks, parse_track(self, track)]
2358                except InvalidDataError as err:
2359                    self.logger.debug("Error parsing track: %s", err)
2360
2361        # Parse albums — audiobooks are split into the audiobooks bucket via
2362        # classify_album. Yandex-returned podcast albums are handled separately
2363        # through the dedicated `.podcasts` node below. ``limit`` is applied per
2364        # bucket AFTER classification — slicing first would drop audiobooks when
2365        # the first ``limit`` results happen to be music albums (or vice versa).
2366        want_album = MediaType.ALBUM in media_types
2367        want_audiobook = MediaType.AUDIOBOOK in media_types
2368        if (want_album or want_audiobook) and search_result.albums:
2369            album_count = 0
2370            audiobook_count = 0
2371            for album in search_result.albums.results:
2372                album_full = not want_album or album_count >= limit
2373                audiobook_full = not want_audiobook or audiobook_count >= limit
2374                if album_full and audiobook_full:
2375                    break
2376                kind = classify_album(album)
2377                try:
2378                    if kind == "audiobook" and want_audiobook and not audiobook_full:
2379                        result.audiobooks = [
2380                            *result.audiobooks,
2381                            parse_audiobook(self, album),
2382                        ]
2383                        audiobook_count += 1
2384                    elif kind == "music" and want_album and not album_full:
2385                        result.albums = [*result.albums, parse_album(self, album)]
2386                        album_count += 1
2387                except InvalidDataError as err:
2388                    self.logger.debug("Error parsing %s album: %s", kind, err)
2389
2390        # Parse artists
2391        if MediaType.ARTIST in media_types and search_result.artists:
2392            for artist in search_result.artists.results[:limit]:
2393                try:
2394                    result.artists = [*result.artists, parse_artist(self, artist)]
2395                except InvalidDataError as err:
2396                    self.logger.debug("Error parsing artist: %s", err)
2397
2398        # Parse playlists
2399        if MediaType.PLAYLIST in media_types and search_result.playlists:
2400            for playlist in search_result.playlists.results[:limit]:
2401                try:
2402                    result.playlists = [*result.playlists, parse_playlist(self, playlist)]
2403                except InvalidDataError as err:
2404                    self.logger.debug("Error parsing playlist: %s", err)
2405
2406        # Parse podcasts (Yandex returns them as albums under .podcasts)
2407        podcasts_node = getattr(search_result, "podcasts", None)
2408        if MediaType.PODCAST in media_types and podcasts_node:
2409            for album in podcasts_node.results[:limit]:
2410                try:
2411                    result.podcasts = [*result.podcasts, parse_podcast(self, album)]
2412                except InvalidDataError as err:
2413                    self.logger.debug("Error parsing podcast: %s", err)
2414
2415        return result
2416
2417    # Get single items
2418
2419    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
2420    async def get_artist(self, prov_artist_id: str) -> Artist:
2421        """
2422        Get artist details by ID, enriched with description and listener stats.
2423
2424        :param prov_artist_id: The provider artist ID.
2425        :return: Artist object.
2426        :raises MediaNotFoundError: If artist not found.
2427        """
2428        artist, about = await asyncio.gather(
2429            self.client.get_artist(prov_artist_id),
2430            self.client.get_artist_about(prov_artist_id),
2431        )
2432        if not artist:
2433            raise MediaNotFoundError(f"Artist {prov_artist_id} not found")
2434        return parse_artist(self, artist, about=about)
2435
2436    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
2437    async def get_album(self, prov_album_id: str) -> Album:
2438        """
2439        Get album details by ID.
2440
2441        :param prov_album_id: The provider album ID.
2442        :return: Album object.
2443        :raises MediaNotFoundError: If album not found.
2444        """
2445        album = await self.client.get_album(prov_album_id)
2446        if not album:
2447            raise MediaNotFoundError(f"Album {prov_album_id} not found")
2448        return parse_album(self, album)
2449
2450    @use_cache(3600 * 24, allow_expired_cache=True)
2451    async def get_podcast(self, prov_podcast_id: str) -> Podcast:
2452        """
2453        Get podcast details by ID (backed by a Yandex album).
2454
2455        :param prov_podcast_id: The provider podcast (album) ID.
2456        :return: Podcast object.
2457        :raises MediaNotFoundError: If not found.
2458        """
2459        album = await self.client.get_album(prov_podcast_id)
2460        if not album:
2461            raise MediaNotFoundError(f"Podcast {prov_podcast_id} not found")
2462        return parse_podcast(self, album)
2463
2464    async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
2465        """Iterate podcast episodes for a given podcast (album) ID."""
2466        album = await self.client.get_album_with_tracks(prov_podcast_id)
2467        if not album:
2468            raise MediaNotFoundError(f"Podcast {prov_podcast_id} not found")
2469        podcast = parse_podcast(self, album)
2470        position = 1
2471        for disc in album.volumes or []:
2472            for track_obj in disc:
2473                try:
2474                    yield parse_podcast_episode(self, track_obj, podcast, position=position)
2475                except InvalidDataError as err:
2476                    self.logger.debug("Error parsing podcast episode: %s", err)
2477                position += 1
2478
2479    async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
2480        """
2481        Get a single podcast episode by ID.
2482
2483        The parent Podcast is reconstructed from the track's parent album. If
2484        the album isn't present on the track, the episode cannot be converted
2485        into a valid MA model and InvalidDataError is raised.
2486        """
2487        tracks = await self.client.get_tracks([prov_episode_id])
2488        if not tracks:
2489            raise MediaNotFoundError(f"Podcast episode {prov_episode_id} not found")
2490        track_obj = tracks[0]
2491        if not track_obj.albums:
2492            raise InvalidDataError(
2493                f"Podcast episode {prov_episode_id} is missing parent podcast album data"
2494            )
2495        podcast = parse_podcast(self, track_obj.albums[0])
2496        return parse_podcast_episode(self, track_obj, podcast, position=0)
2497
2498    @use_cache(3600 * 24, allow_expired_cache=True)
2499    async def get_audiobook(self, prov_audiobook_id: str) -> Audiobook:
2500        """
2501        Get audiobook details by ID, including chapters built from tracks.
2502
2503        :param prov_audiobook_id: The provider audiobook (album) ID.
2504        :return: Audiobook object.
2505        :raises MediaNotFoundError: If not found.
2506        """
2507        album = await self.client.get_album_with_tracks(prov_audiobook_id)
2508        if not album:
2509            raise MediaNotFoundError(f"Audiobook {prov_audiobook_id} not found")
2510        audiobook = parse_audiobook(self, album)
2511
2512        chapters: list[MediaItemChapter] = []
2513        start = 0.0
2514        pos = 1
2515        for disc in album.volumes or []:
2516            for track_obj in disc:
2517                dur_s = (track_obj.duration_ms or 0) / 1000.0
2518                chapters.append(
2519                    MediaItemChapter(
2520                        position=pos,
2521                        name=track_obj.title or f"Chapter {pos}",
2522                        start=start,
2523                        end=start + dur_s,
2524                    )
2525                )
2526                start += dur_s
2527                pos += 1
2528        audiobook.metadata.chapters = chapters
2529        audiobook.duration = int(start)
2530        return audiobook
2531
2532    async def get_track(self, prov_track_id: str) -> Track:
2533        """
2534        Get track details by ID.
2535
2536        Supports composite item_id (track_id@station_id) for My Wave tracks;
2537        only the track_id part is used for the API. Normalizes the ID before
2538        caching to avoid duplicate cache entries.
2539
2540        :param prov_track_id: The provider track ID (or track_id@station_id).
2541        :return: Track object.
2542        :raises MediaNotFoundError: If track not found.
2543        """
2544        track_id, _ = _parse_radio_item_id(prov_track_id)
2545        return await self._get_track_cached(track_id)
2546
2547    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
2548    async def _get_track_cached(self, track_id: str) -> Track:
2549        """
2550        Get track details by normalized ID (cached).
2551
2552        :param track_id: Normalized track ID (without station suffix).
2553        :return: Track object.
2554        :raises MediaNotFoundError: If track not found.
2555        """
2556        yandex_track = await self.client.get_track(track_id)
2557        if not yandex_track:
2558            raise MediaNotFoundError(f"Track {track_id} not found")
2559
2560        # Use the already-fetched track object to avoid a duplicate API call
2561        lyrics, lyrics_synced = await self.client.get_track_lyrics_from_track(yandex_track)
2562
2563        return parse_track(self, yandex_track, lyrics=lyrics, lyrics_synced=lyrics_synced)
2564
2565    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
2566        """
2567        Get playlist details by ID.
2568
2569        Supports virtual playlists MY_WAVE_PLAYLIST_ID (My Wave) and
2570        LIKED_TRACKS_PLAYLIST_ID (Liked Tracks). Real playlists use format "owner_id:kind".
2571
2572        :param prov_playlist_id: The provider playlist ID (format: "owner_id:kind",
2573            my_wave, or liked_tracks).
2574        :return: Playlist object.
2575        :raises MediaNotFoundError: If playlist not found.
2576        """
2577        # Virtual playlists - constructed locally (no API call); translation_key localizes
2578        # the name for the connection locale at serialization.
2579        if prov_playlist_id == MY_WAVE_PLAYLIST_ID:
2580            return Playlist(
2581                item_id=MY_WAVE_PLAYLIST_ID,
2582                provider=self.instance_id,
2583                name="My Wave",
2584                translation_key=MY_WAVE_PLAYLIST_ID,
2585                owner=get_canonical_provider_name(self),
2586                provider_mappings={
2587                    ProviderMapping(
2588                        item_id=MY_WAVE_PLAYLIST_ID,
2589                        provider_domain=self.domain,
2590                        provider_instance=self.instance_id,
2591                        is_unique=True,
2592                    )
2593                },
2594                is_editable=False,
2595                is_dynamic=True,
2596            )
2597
2598        if prov_playlist_id == LIKED_TRACKS_PLAYLIST_ID:
2599            return Playlist(
2600                item_id=LIKED_TRACKS_PLAYLIST_ID,
2601                provider=self.instance_id,
2602                name="My Favorites",
2603                translation_key=LIKED_TRACKS_PLAYLIST_ID,
2604                owner=get_canonical_provider_name(self),
2605                provider_mappings={
2606                    ProviderMapping(
2607                        item_id=LIKED_TRACKS_PLAYLIST_ID,
2608                        provider_domain=self.domain,
2609                        provider_instance=self.instance_id,
2610                        is_unique=True,
2611                    )
2612                },
2613                is_editable=False,
2614            )
2615
2616        # Real playlists - use cached method
2617        return await self._get_real_playlist(prov_playlist_id)
2618
2619    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
2620    async def _get_real_playlist(self, prov_playlist_id: str) -> Playlist:
2621        """
2622        Get real playlist details by ID (cached).
2623
2624        :param prov_playlist_id: The provider playlist ID (format: "owner_id:kind").
2625        :return: Playlist object.
2626        :raises MediaNotFoundError: If playlist not found.
2627        """
2628        # Parse the playlist ID (format: owner_id:kind)
2629        if PLAYLIST_ID_SPLITTER in prov_playlist_id:
2630            owner_id, kind = prov_playlist_id.split(PLAYLIST_ID_SPLITTER, 1)
2631        else:
2632            owner_id = str(self.client.user_id)
2633            kind = prov_playlist_id
2634
2635        playlist = await self.client.get_playlist(owner_id, kind)
2636        if not playlist:
2637            raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found")
2638        return parse_playlist(self, playlist)
2639
2640    @use_cache(3600 * 3, allow_expired_cache=True)
2641    async def _get_my_wave_playlist_tracks(self, page: int) -> list[Track]:
2642        """
2643        Get My Wave tracks for virtual playlist (uses cursor for page > 0).
2644
2645        Fetches MY_WAVE_BATCH_SIZE Rotor API batches per page call to reduce
2646        the number of round-trips when the player controller paginates through pages.
2647
2648        :param page: Page number (0 = first batch, 1+ = next batches via queue cursor).
2649        :return: List of Track objects for this page.
2650        """
2651        wave = self._get_wave_state(ROTOR_STATION_MY_WAVE)
2652        async with wave.lock:
2653            max_tracks_config = int(
2654                self.config.get_value(CONF_MY_WAVE_MAX_TRACKS) or 150  # type: ignore[arg-type]
2655            )
2656
2657            # Reset seen tracks on first page
2658            if page == 0:
2659                wave.seen_track_ids = set()
2660
2661            queue: str | int | None = None
2662            if page > 0:
2663                queue = wave.playlist_next_cursor
2664                if not queue:
2665                    return []
2666
2667            # Check if we've already reached the limit
2668            if len(wave.seen_track_ids) >= max_tracks_config:
2669                return []
2670
2671            tracks: list[Track] = []
2672            next_cursor: str | None = None
2673
2674            # Fetch MY_WAVE_BATCH_SIZE Rotor API batches per page to reduce API round-trips
2675            for _ in range(MY_WAVE_BATCH_SIZE):
2676                if len(wave.seen_track_ids) >= max_tracks_config:
2677                    break
2678
2679                if queue is not None:
2680                    wave.last_track_id = str(queue)
2681                yandex_tracks, _ = await self._fetch_rotor_session_batch(
2682                    wave, ROTOR_STATION_MY_WAVE
2683                )
2684                if not wave.radio_started_sent and yandex_tracks:
2685                    sent = await self._send_wave_feedback(
2686                        wave, ROTOR_STATION_MY_WAVE, "radioStarted"
2687                    )
2688                    if sent:
2689                        wave.radio_started_sent = True
2690
2691                if not yandex_tracks:
2692                    break
2693
2694                first_track_id_this_batch = None
2695                for yt in yandex_tracks:
2696                    if len(wave.seen_track_ids) >= max_tracks_config:
2697                        break
2698
2699                    track = self._parse_my_wave_track(yt, wave.seen_track_ids)
2700                    if track is None:
2701                        continue
2702
2703                    tracks.append(track)
2704                    track_id = track.item_id.split(RADIO_TRACK_ID_SEP, 1)[0]
2705                    if first_track_id_this_batch is None:
2706                        first_track_id_this_batch = track_id
2707
2708                if first_track_id_this_batch is not None:
2709                    next_cursor = first_track_id_this_batch
2710                    queue = first_track_id_this_batch
2711                else:
2712                    # All tracks in this batch were duplicates or failed to parse
2713                    break
2714
2715            # Store cursor for next page call (None clears pagination so next call returns [])
2716            wave.playlist_next_cursor = next_cursor
2717            return tracks
2718
2719    @use_cache(3600 * 3, allow_expired_cache=True)
2720    async def _get_liked_tracks_playlist_tracks(self, page: int) -> list[Track]:
2721        """
2722        Get liked tracks for virtual playlist (sorted in reverse chronological order).
2723
2724        :param page: Page number (0 = all tracks limited by config, >0 = empty for pagination).
2725        :return: List of Track objects.
2726        """
2727        # Liked tracks API returns all tracks at once, so only return tracks on page 0
2728        if page > 0:
2729            return []
2730
2731        max_tracks_config = int(
2732            self.config.get_value(CONF_LIKED_TRACKS_MAX_TRACKS) or 200  # type: ignore[arg-type]
2733        )
2734
2735        # Fetch liked tracks (already sorted in reverse chronological order by api_client)
2736        track_shorts = await self.client.get_liked_tracks()
2737        if not track_shorts:
2738            self.logger.debug("No liked tracks found")
2739            return []
2740
2741        # Apply max tracks limit
2742        track_shorts = track_shorts[:max_tracks_config]
2743
2744        # Fetch full track details in batches
2745        track_ids = [str(ts.track_id) for ts in track_shorts if ts.track_id]
2746
2747        batch_size = TRACK_BATCH_SIZE
2748        full_tracks = []
2749        for i in range(0, len(track_ids), batch_size):
2750            batch_ids = track_ids[i : i + batch_size]
2751            batch_result = await self.client.get_tracks(batch_ids)
2752            full_tracks.extend(batch_result)
2753            # Spread bursts: insert a small jittered pause between batches so
2754            # a 500-track hydration doesn't look like a bot to Yandex's
2755            # smart-captcha. Skipped after the last batch.
2756            if i + batch_size < len(track_ids):
2757                await asyncio.sleep(
2758                    LIKED_BATCH_JITTER_MIN_S + random.random() * LIKED_BATCH_JITTER_SPAN_S
2759                )
2760
2761        # Create track ID to full track mapping by track ID directly
2762        track_map = {}
2763        for t in full_tracks:
2764            if hasattr(t, "id") and t.id:
2765                track_map[str(t.id)] = t
2766
2767        # Parse tracks in the original order (reverse chronological)
2768        tracks = []
2769        for track_id in track_ids:
2770            # track_id may be compound "trackId:albumId", extract base ID for lookup
2771            base_id = track_id.split(":")[0] if ":" in track_id else track_id
2772            found = track_map.get(track_id) or track_map.get(base_id)
2773            if found:
2774                try:
2775                    tracks.append(parse_track(self, found))
2776                except InvalidDataError as err:
2777                    self.logger.debug("Error parsing liked track %s: %s", track_id, err)
2778
2779        self.logger.debug("Liked tracks: fetched %s, parsed %s", len(track_shorts), len(tracks))
2780        return tracks
2781
2782    # Get related items
2783
2784    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
2785    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
2786        """
2787        Get album tracks.
2788
2789        :param prov_album_id: The provider album ID.
2790        :return: List of Track objects.
2791        """
2792        album = await self.client.get_album_with_tracks(prov_album_id)
2793        if not album or not album.volumes:
2794            return []
2795
2796        tracks = []
2797        for volume_index, volume in enumerate(album.volumes):
2798            for track_index, track in enumerate(volume):
2799                try:
2800                    parsed_track = parse_track(self, track)
2801                    parsed_track.disc_number = volume_index + 1
2802                    parsed_track.track_number = track_index + 1
2803                    tracks.append(parsed_track)
2804                except InvalidDataError as err:
2805                    self.logger.debug("Error parsing album track: %s", err)
2806        return tracks
2807
2808    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
2809        """
2810        Get similar tracks, preferring pre-fetched wave tracks when available.
2811
2812        Split in two paths with different caching policies:
2813
2814        - **Wave-drain path** (the seed carries a station suffix and
2815          ``wave.prefetched`` is non-empty). Uncached by design: it mutates
2816          state, a cache hit would replay the same drained tracks forever and
2817          the prefetch buffer would never advance.
2818        - **Fallback path** (plain track_id, no active wave, or empty buffer).
2819          Creates a per-seed rotor session under ``track:{id}`` and is cached
2820          for 3 hours — this is pure and safe to memoise.
2821
2822        :param prov_track_id: Provider track ID (plain or track_id@station_id).
2823        :param limit: Maximum number of tracks to return.
2824        :return: List of similar Track objects.
2825        """
2826        track_id, station_key = _parse_radio_item_id(prov_track_id)
2827
2828        if station_key:
2829            drained = await self._drain_prefetched_wave_tracks(station_key, limit)
2830            if drained:
2831                return drained
2832
2833        return await self._fetch_similar_tracks_for_seed(track_id, limit)
2834
2835    async def _drain_prefetched_wave_tracks(self, station_key: str, limit: int) -> list[Track]:
2836        """
2837        Pop up to ``limit`` prefetched tracks off the wave state.
2838
2839        Runs under ``wave.lock`` so it doesn't race with
2840        ``_prefetch_rotor_session`` which extends the same list under the
2841        same lock. Returns an empty list when there's no active session or
2842        nothing prefetched; callers then fall through to the cached fetch.
2843
2844        This method is intentionally not cached — it mutates wave state.
2845        """
2846        wave = self._wave_states.get(station_key)
2847        if not (wave and wave.session_id and wave.prefetched):
2848            return []
2849        async with wave.lock:
2850            if not wave.prefetched:
2851                return []
2852            drained_yt = wave.prefetched[:limit]
2853            wave.prefetched = wave.prefetched[limit:]
2854        tracks: list[Track] = []
2855        for yt in drained_yt:
2856            try:
2857                tracks.append(parse_track(self, yt))
2858            except InvalidDataError as err:
2859                self.logger.debug("Error parsing prefetched wave track: %s", err)
2860        return tracks
2861
2862    @use_cache(3600 * 3, allow_expired_cache=True)
2863    async def _fetch_similar_tracks_for_seed(self, track_id: str, limit: int) -> list[Track]:
2864        """
2865        Create a one-off rotor session for ``track:{id}`` and return up to ``limit`` tracks.
2866
2867        Stateless by design: similar-tracks results don't participate in
2868        playback feedback or prefetch, so there is no need to keep a
2869        ``_WaveState`` entry around. Going through ``_fetch_rotor_session_batch``
2870        would create one per unique seed and grow ``_wave_states`` without
2871        bound under normal DSTM usage; call ``rotor_session_new`` directly
2872        instead.
2873
2874        Pure function of ``track_id`` / ``limit``, hence safe to memoise
2875        via ``@use_cache``.
2876        """
2877        _, yandex_tracks, _ = await self.client.rotor_session_new(f"track:{track_id}")
2878        similar_tracks: list[Track] = []
2879        for yt in yandex_tracks[:limit]:
2880            try:
2881                similar_tracks.append(parse_track(self, yt))
2882            except InvalidDataError as err:
2883                self.logger.debug("Error parsing similar track: %s", err)
2884        return similar_tracks
2885
2886    @use_cache(3600 * 3, allow_expired_cache=True)
2887    async def get_similar_artists(self, prov_artist_id: str, limit: int = 25) -> list[Artist]:
2888        """
2889        Get artists similar to the given one via Yandex artists/similar endpoint.
2890
2891        :param prov_artist_id: Provider artist ID.
2892        :param limit: Maximum number of artists to return.
2893        :return: List of similar Artist objects.
2894        """
2895        yandex_artists = await self.client.get_similar_artists(prov_artist_id, limit=limit)
2896        artists: list[Artist] = []
2897        for ya in yandex_artists:
2898            try:
2899                artists.append(parse_artist(self, ya))
2900            except InvalidDataError as err:
2901                self.logger.debug("Error parsing similar artist: %s", err)
2902        return artists
2903
2904    async def get_recommendations(self) -> list[RecommendationFolder]:
2905        """
2906        Get the available recommendation rows, without items.
2907
2908        Returns My Wave, Made for You, Chart, New Releases, New Playlists,
2909        Top Picks, Mood Mix, Activity Mix and Seasonal Mix rows.
2910        """
2911        # The seasonal row title carries the current season, derived locally from the month.
2912        seasonal_tag = TAG_SEASONAL_MAP.get(utc().month, "autumn")
2913        seasonal_name, _ = self._media_label(
2914            "folder", _media_label_key(seasonal_tag), seasonal_tag.title()
2915        )
2916        return [
2917            RecommendationFolder(
2918                item_id=MY_WAVE_PLAYLIST_ID,
2919                provider=self.instance_id,
2920                name="My Wave",
2921                translation_key=MY_WAVE_PLAYLIST_ID,
2922                icon="mdi-waveform",
2923            ),
2924            RecommendationFolder(
2925                item_id="feed",
2926                provider=self.instance_id,
2927                name="Made for You",
2928                translation_key="feed",
2929                icon="mdi-account-music",
2930            ),
2931            RecommendationFolder(
2932                item_id="chart",
2933                provider=self.instance_id,
2934                name="Chart",
2935                translation_key="chart",
2936                icon="mdi-chart-line",
2937            ),
2938            RecommendationFolder(
2939                item_id="new_releases",
2940                provider=self.instance_id,
2941                name="New Releases",
2942                translation_key="new_releases",
2943                icon="mdi-new-box",
2944            ),
2945            RecommendationFolder(
2946                item_id="new_playlists",
2947                provider=self.instance_id,
2948                name="New Playlists",
2949                translation_key="new_playlists",
2950                icon="mdi-playlist-star",
2951            ),
2952            RecommendationFolder(
2953                item_id="top_picks",
2954                provider=self.instance_id,
2955                name="Top Picks",
2956                translation_key="top_picks",
2957                icon="mdi-star",
2958            ),
2959            # Mood/Activity rows have a static title; the hourly rotating tag - derived
2960            # deterministically, so the items call independently computes the same one -
2961            # shows as the row subtitle (cache-only tag-list read, no backend I/O).
2962            RecommendationFolder(
2963                item_id="mood_mix",
2964                provider=self.instance_id,
2965                name="Mood Mix",
2966                translation_key="mood_mix",
2967                subtitle=await self._rotating_row_tag_subtitle("mood"),
2968                icon="mdi-emoticon-outline",
2969            ),
2970            RecommendationFolder(
2971                item_id="activity_mix",
2972                provider=self.instance_id,
2973                name="Activity Mix",
2974                translation_key="activity_mix",
2975                subtitle=await self._rotating_row_tag_subtitle("activity"),
2976                icon="mdi-run",
2977            ),
2978            RecommendationFolder(
2979                item_id="seasonal_mix",
2980                provider=self.instance_id,
2981                name=f"Seasonal: {seasonal_name}",
2982                translation_key="seasonal_mix",
2983                translation_params=[seasonal_name],
2984                icon="mdi-weather-sunny",
2985            ),
2986        ]
2987
2988    async def get_recommendation_items(
2989        self, item_id: str
2990    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
2991        """
2992        Get the items for a single recommendation row.
2993
2994        :param item_id: The item_id of the row, as returned by get_recommendations.
2995        """
2996        folder: RecommendationFolder | None = None
2997        if item_id == MY_WAVE_PLAYLIST_ID:
2998            folder = await self._get_my_wave_recommendations()
2999        elif item_id == "feed":
3000            folder = await self._get_feed_recommendations()
3001        elif item_id == "chart":
3002            folder = await self._get_chart_recommendations()
3003        elif item_id == "new_releases":
3004            folder = await self._get_new_releases_recommendations()
3005        elif item_id == "new_playlists":
3006            folder = await self._get_new_playlists_recommendations()
3007        elif item_id == "top_picks":
3008            folder = await self._get_top_picks_recommendations()
3009        elif item_id == "mood_mix":
3010            # the deterministic hourly tag keeps the served items matching the row subtitle
3011            if mood_tags := await self._get_valid_tags_for_category("mood"):
3012                folder = await self._get_mood_mix_recommendations(
3013                    self._rotating_row_tag("mood", mood_tags)
3014                )
3015        elif item_id == "activity_mix":
3016            if activity_tags := await self._get_valid_tags_for_category("activity"):
3017                folder = await self._get_activity_mix_recommendations(
3018                    self._rotating_row_tag("activity", activity_tags)
3019                )
3020        elif item_id == "seasonal_mix":
3021            folder = await self._get_seasonal_mix_recommendations()
3022        if folder is None:
3023            return UniqueList()
3024        return folder.items
3025
3026    @use_cache(600, allow_expired_cache=True)
3027    async def _get_my_wave_recommendations(self) -> RecommendationFolder | None:
3028        """
3029        Get My Wave recommendation folder with personalized tracks.
3030
3031        Shares the same `_WaveState(ROTOR_STATION_MY_WAVE)` with browse and
3032        virtual-playlist flows, so session_id + batch_id established here
3033        carry into `on_played`/`on_streamed` feedback even when the user
3034        starts playback from this discovery card.
3035
3036        :return: RecommendationFolder with My Wave tracks, or None if empty.
3037        """
3038        max_tracks_config = int(
3039            self.config.get_value(CONF_MY_WAVE_MAX_TRACKS) or 150  # type: ignore[arg-type]
3040        )
3041        batch_size_config = MY_WAVE_BATCH_SIZE
3042
3043        wave = self._get_wave_state(ROTOR_STATION_MY_WAVE)
3044        # Local dedup so the recommendations card stays independent from the
3045        # browse/virtual-playlist dedup set (which may be larger and stale).
3046        # Only session_id + batch_id + last_track_id are shared with `wave`.
3047        seen_track_ids: set[str] = set()
3048        items: list[Track] = []
3049
3050        # Hold the wave lock across the whole fetch chain — we mutate shared
3051        # session_id/batch_id/last_track_id via _fetch_rotor_session_batch,
3052        # and other call sites (browse, virtual-playlist) guard the same
3053        # state with this lock. Concurrent calls without the lock would
3054        # interleave cursor updates and leave the session inconsistent.
3055        async with wave.lock:
3056            for _ in range(batch_size_config):
3057                if len(seen_track_ids) >= max_tracks_config:
3058                    break
3059
3060                yandex_tracks, _ = await self._fetch_rotor_session_batch(
3061                    wave, ROTOR_STATION_MY_WAVE
3062                )
3063                if not yandex_tracks:
3064                    break
3065
3066                first_track_id_this_batch: str | None = None
3067                for yt in yandex_tracks:
3068                    if len(seen_track_ids) >= max_tracks_config:
3069                        break
3070
3071                    track = self._parse_my_wave_track(yt, seen_ids=seen_track_ids)
3072                    if track is None:
3073                        continue
3074
3075                    items.append(track)
3076                    track_id = track.item_id.split(RADIO_TRACK_ID_SEP, 1)[0]
3077                    if first_track_id_this_batch is None:
3078                        first_track_id_this_batch = track_id
3079
3080                if first_track_id_this_batch is None:
3081                    break
3082                wave.last_track_id = first_track_id_this_batch
3083
3084        if not items:
3085            return None
3086
3087        initial_tracks_limit = DISCOVERY_INITIAL_TRACKS
3088        if len(items) > initial_tracks_limit:
3089            items = items[:initial_tracks_limit]
3090
3091        return RecommendationFolder(
3092            item_id=MY_WAVE_PLAYLIST_ID,
3093            provider=self.instance_id,
3094            name="My Wave",
3095            translation_key=MY_WAVE_PLAYLIST_ID,
3096            items=UniqueList(items),
3097            icon="mdi-waveform",
3098        )
3099
3100    @use_cache(1800, allow_expired_cache=True)
3101    async def _get_feed_recommendations(self) -> RecommendationFolder | None:
3102        """
3103        Get personalized feed playlists (Playlist of the Day, DejaVu, etc.).
3104
3105        :return: RecommendationFolder with generated playlists, or None if unavailable.
3106        """
3107        feed = await self.client.get_feed()
3108        if not feed or not feed.generated_playlists:
3109            return None
3110        items: list[Playlist] = []
3111        for gen_playlist in feed.generated_playlists:
3112            if gen_playlist.data and gen_playlist.ready:
3113                try:
3114                    # Mark feed-generated playlists (Playlist of the Day, DejaVu,
3115                    # Premiere, Missed Likes) as dynamic — Yandex regenerates them
3116                    # on a schedule so MA must not long-cache the track list.
3117                    items.append(parse_playlist(self, gen_playlist.data, is_dynamic=True))
3118                except InvalidDataError as err:
3119                    self.logger.debug("Error parsing feed playlist: %s", err)
3120        if not items:
3121            return None
3122        return RecommendationFolder(
3123            item_id="feed",
3124            provider=self.instance_id,
3125            name="Made for You",
3126            translation_key="feed",
3127            items=UniqueList(items),
3128            icon="mdi-account-music",
3129        )
3130
3131    @use_cache(3600, allow_expired_cache=True)
3132    async def _get_chart_recommendations(self) -> RecommendationFolder | None:
3133        """
3134        Get chart tracks (hot tracks of the month).
3135
3136        :return: RecommendationFolder with chart tracks, or None if unavailable.
3137        """
3138        chart_info = await self.client.get_chart()
3139        if not chart_info or not chart_info.chart:
3140            return None
3141        playlist = chart_info.chart
3142        if not playlist.tracks:
3143            return None
3144        # TrackShort objects in chart context have .track (full Track) and .chart (position)
3145        tracks: list[Track] = []
3146        for track_short in playlist.tracks[:20]:
3147            track_obj = getattr(track_short, "track", None)
3148            if not track_obj:
3149                continue
3150            try:
3151                tracks.append(parse_track(self, track_obj))
3152            except InvalidDataError as err:
3153                self.logger.debug("Error parsing chart track: %s", err)
3154        if not tracks:
3155            return None
3156        return RecommendationFolder(
3157            item_id="chart",
3158            provider=self.instance_id,
3159            name="Chart",
3160            translation_key="chart",
3161            items=UniqueList(tracks),
3162            icon="mdi-chart-line",
3163        )
3164
3165    @use_cache(3600, allow_expired_cache=True)
3166    async def _get_new_releases_recommendations(self) -> RecommendationFolder | None:
3167        """
3168        Get new album releases.
3169
3170        :return: RecommendationFolder with new albums, or None if unavailable.
3171        """
3172        releases = await self.client.get_new_releases()
3173        if not releases or not releases.new_releases:
3174            return None
3175        # new_releases is a list of album IDs (int) — need to batch-fetch full details
3176        album_ids = [str(aid) for aid in releases.new_releases[:20]]
3177        if not album_ids:
3178            return None
3179        full_albums = await self.client.get_albums(album_ids)
3180        if not full_albums:
3181            return None
3182        albums: list[Album] = []
3183        for album in full_albums:
3184            try:
3185                albums.append(parse_album(self, album))
3186            except InvalidDataError as err:
3187                self.logger.debug("Error parsing new release album: %s", err)
3188        if not albums:
3189            return None
3190        return RecommendationFolder(
3191            item_id="new_releases",
3192            provider=self.instance_id,
3193            name="New Releases",
3194            translation_key="new_releases",
3195            items=UniqueList(albums),
3196            icon="mdi-new-box",
3197        )
3198
3199    @use_cache(3600, allow_expired_cache=True)
3200    async def _get_new_playlists_recommendations(self) -> RecommendationFolder | None:
3201        """
3202        Get new editorial playlists.
3203
3204        :return: RecommendationFolder with new playlists, or None if unavailable.
3205        """
3206        result = await self.client.get_new_playlists()
3207        if not result or not result.new_playlists:
3208            return None
3209        # new_playlists is a list of PlaylistId objects (uid, kind) — fetch full details
3210        playlist_ids = [
3211            f"{pid.uid}:{pid.kind}"
3212            for pid in result.new_playlists[:20]
3213            if hasattr(pid, "uid") and hasattr(pid, "kind")
3214        ]
3215        if not playlist_ids:
3216            return None
3217        full_playlists = await self.client.get_playlists(playlist_ids)
3218        if not full_playlists:
3219            return None
3220        playlists: list[Playlist] = []
3221        for playlist in full_playlists:
3222            try:
3223                playlists.append(parse_playlist(self, playlist))
3224            except InvalidDataError as err:
3225                self.logger.debug("Error parsing new playlist: %s", err)
3226        if not playlists:
3227            return None
3228        return RecommendationFolder(
3229            item_id="new_playlists",
3230            provider=self.instance_id,
3231            name="New Playlists",
3232            translation_key="new_playlists",
3233            items=UniqueList(playlists),
3234            icon="mdi-playlist-star",
3235        )
3236
3237    @use_cache(3600, allow_expired_cache=True)
3238    async def _get_top_picks_recommendations(self) -> RecommendationFolder | None:
3239        """
3240        Get Top Picks recommendation folder (tag: top).
3241
3242        :return: RecommendationFolder with top playlists, or None if unavailable.
3243        """
3244        playlists = await self.client.get_tag_playlists("top")
3245        if not playlists:
3246            return None
3247        items: list[Playlist] = []
3248        for playlist in playlists[:10]:
3249            try:
3250                items.append(parse_playlist(self, playlist))
3251            except InvalidDataError as err:
3252                self.logger.debug("Error parsing top picks playlist: %s", err)
3253        if not items:
3254            return None
3255        return RecommendationFolder(
3256            item_id="top_picks",
3257            provider=self.instance_id,
3258            name="Top Picks",
3259            translation_key="top_picks",
3260            items=UniqueList(items),
3261            icon="mdi-star",
3262        )
3263
3264    @use_cache(1800, allow_expired_cache=True)
3265    async def _get_mood_mix_recommendations(self, mood_tag: str) -> RecommendationFolder | None:
3266        """
3267        Get Mood Mix recommendation folder for a specific tag.
3268
3269        :param mood_tag: Preselected mood tag slug.
3270        :return: RecommendationFolder with mood playlists, or None if unavailable.
3271        """
3272        playlists = await self.client.get_tag_playlists(mood_tag)
3273        if not playlists:
3274            self.logger.debug("No playlists for mood tag %s, skipping recommendation", mood_tag)
3275            return None
3276        items: list[Playlist] = []
3277        for playlist in playlists[:8]:
3278            try:
3279                items.append(parse_playlist(self, playlist))
3280            except InvalidDataError as err:
3281                self.logger.debug("Error parsing mood playlist: %s", err)
3282        if not items:
3283            return None
3284        tag_name, _ = self._media_label("folder", _media_label_key(mood_tag), mood_tag.title())
3285        return RecommendationFolder(
3286            item_id="mood_mix",
3287            provider=self.instance_id,
3288            name=f"Mood Mix: {tag_name}",
3289            translation_key="mood_mix",
3290            translation_params=[tag_name],
3291            items=UniqueList(items),
3292            icon="mdi-emoticon-outline",
3293        )
3294
3295    @use_cache(1800, allow_expired_cache=True)
3296    async def _get_activity_mix_recommendations(
3297        self, activity_tag: str
3298    ) -> RecommendationFolder | None:
3299        """
3300        Get Activity Mix recommendation folder for a specific tag.
3301
3302        :param activity_tag: Preselected activity tag slug.
3303        :return: RecommendationFolder with activity playlists, or None if unavailable.
3304        """
3305        playlists = await self.client.get_tag_playlists(activity_tag)
3306        if not playlists:
3307            self.logger.debug(
3308                "No playlists for activity tag %s, skipping recommendation", activity_tag
3309            )
3310            return None
3311        items: list[Playlist] = []
3312        for playlist in playlists[:8]:
3313            try:
3314                items.append(parse_playlist(self, playlist))
3315            except InvalidDataError as err:
3316                self.logger.debug("Error parsing activity playlist: %s", err)
3317        if not items:
3318            return None
3319        tag_name, _ = self._media_label(
3320            "folder", _media_label_key(activity_tag), activity_tag.title()
3321        )
3322        return RecommendationFolder(
3323            item_id="activity_mix",
3324            provider=self.instance_id,
3325            name=f"Activity Mix: {tag_name}",
3326            translation_key="activity_mix",
3327            translation_params=[tag_name],
3328            items=UniqueList(items),
3329            icon="mdi-run",
3330        )
3331
3332    @use_cache(3600 * 6, allow_expired_cache=True)
3333    async def _get_seasonal_mix_recommendations(self) -> RecommendationFolder | None:
3334        """
3335        Get Seasonal Mix recommendation folder (based on current month).
3336
3337        :return: RecommendationFolder with seasonal playlists, or None if unavailable.
3338        """
3339        # Determine current season tag; fall back to autumn if the seasonal
3340        # endpoint returns nothing (e.g. spring/autumn handover gap).
3341        current_month = utc().month
3342        seasonal_tag = TAG_SEASONAL_MAP.get(current_month, "autumn")
3343        playlists = await self.client.get_tag_playlists(seasonal_tag)
3344        if not playlists and seasonal_tag != "autumn":
3345            seasonal_tag = "autumn"
3346            playlists = await self.client.get_tag_playlists(seasonal_tag)
3347        if not playlists:
3348            return None
3349        items: list[Playlist] = []
3350        for playlist in playlists[:8]:
3351            try:
3352                items.append(parse_playlist(self, playlist))
3353            except InvalidDataError as err:
3354                self.logger.debug("Error parsing seasonal playlist: %s", err)
3355        if not items:
3356            return None
3357        tag_name, _ = self._media_label(
3358            "folder", _media_label_key(seasonal_tag), seasonal_tag.title()
3359        )
3360        return RecommendationFolder(
3361            item_id="seasonal_mix",
3362            provider=self.instance_id,
3363            name=f"Seasonal: {tag_name}",
3364            translation_key="seasonal_mix",
3365            translation_params=[tag_name],
3366            items=UniqueList(items),
3367            icon="mdi-weather-sunny",
3368        )
3369
3370    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
3371        """
3372        Get playlist tracks.
3373
3374        :param prov_playlist_id: The provider playlist ID (format: "owner_id:kind",
3375            my_wave, or liked_tracks).
3376        :param page: Page number for pagination.
3377        :return: List of Track objects.
3378        """
3379        self.logger.debug(
3380            "get_playlist_tracks called: prov_playlist_id=%s, page=%s", prov_playlist_id, page
3381        )
3382
3383        if prov_playlist_id == MY_WAVE_PLAYLIST_ID:
3384            self.logger.debug("Fetching My Wave tracks")
3385            return await self._get_my_wave_playlist_tracks(page)
3386
3387        if prov_playlist_id == LIKED_TRACKS_PLAYLIST_ID:
3388            self.logger.debug("Fetching Liked Tracks for virtual playlist")
3389            result = await self._get_liked_tracks_playlist_tracks(page)
3390            self.logger.debug("Liked Tracks playlist returned %s tracks", len(result))
3391            return result
3392
3393        return await self._get_regular_playlist_tracks(prov_playlist_id, page)
3394
3395    @use_cache(3600 * 3, allow_expired_cache=True)
3396    async def _get_regular_playlist_tracks(self, prov_playlist_id: str, page: int) -> list[Track]:
3397        """
3398        Get the tracks of a regular (non-virtual) playlist.
3399
3400        :param prov_playlist_id: The provider playlist ID (format: "owner_id:kind").
3401        :param page: Page number for pagination.
3402        :return: List of Track objects.
3403        """
3404        # Yandex Music API returns all playlist tracks in one call (no server-side pagination).
3405        # Return empty list for page > 0 so the controller pagination loop terminates.
3406        if page > 0:
3407            return []
3408
3409        # Parse the playlist ID (format: owner_id:kind)
3410        if PLAYLIST_ID_SPLITTER in prov_playlist_id:
3411            owner_id, kind = prov_playlist_id.split(PLAYLIST_ID_SPLITTER, 1)
3412        else:
3413            owner_id = str(self.client.user_id)
3414            kind = prov_playlist_id
3415
3416        playlist = await self.client.get_playlist(owner_id, kind)
3417        if not playlist:
3418            return []
3419
3420        # API sometimes returns playlist without tracks; fetch them explicitly if needed
3421        tracks_list = playlist.tracks or []
3422        track_count = getattr(playlist, "track_count", None) or 0
3423        if not tracks_list and track_count > 0:
3424            self.logger.debug(
3425                "Playlist %s/%s: track_count=%s but no tracks in response, "
3426                "calling fetch_tracks_async",
3427                owner_id,
3428                kind,
3429                track_count,
3430            )
3431            try:
3432                tracks_list = await playlist.fetch_tracks_async()
3433            except Exception as err:
3434                self.logger.warning("fetch_tracks_async failed for %s/%s: %s", owner_id, kind, err)
3435            if not tracks_list:
3436                raise ResourceTemporarilyUnavailable(
3437                    "Playlist tracks not available; try again later"
3438                )
3439
3440        if not tracks_list:
3441            return []
3442
3443        # Yandex returns TrackShort objects, we need to fetch full track info
3444        track_ids = [
3445            str(track.track_id) if hasattr(track, "track_id") else str(track.id)
3446            for track in tracks_list
3447            if track
3448        ]
3449        if not track_ids:
3450            return []
3451
3452        # Fetch full track details in batches to avoid timeouts
3453        batch_size = TRACK_BATCH_SIZE
3454        full_tracks = []
3455        for i in range(0, len(track_ids), batch_size):
3456            batch = track_ids[i : i + batch_size]
3457            batch_result = await self.client.get_tracks(batch)
3458            if not batch_result:
3459                # Skip this batch but keep going — the terminal guard below
3460                # raises if every batch comes back empty. Aborting on a single
3461                # empty batch threw away tracks already fetched from earlier
3462                # batches and forced a full retry hours later (under the
3463                # @use_cache TTL above).
3464                self.logger.warning(
3465                    "Empty batch %s-%s for playlist %s, skipping",
3466                    i,
3467                    i + len(batch) - 1,
3468                    prov_playlist_id,
3469                )
3470                continue
3471            full_tracks.extend(batch_result)
3472
3473        if track_ids and not full_tracks:
3474            raise ResourceTemporarilyUnavailable("Failed to load track details; try again later")
3475
3476        tracks = []
3477        for track in full_tracks:
3478            try:
3479                tracks.append(parse_track(self, track))
3480            except InvalidDataError as err:
3481                self.logger.debug("Error parsing playlist track: %s", err)
3482        return tracks
3483
3484    @use_cache(3600 * 24 * 7, allow_expired_cache=True)
3485    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
3486        """
3487        Get artist's albums.
3488
3489        :param prov_artist_id: The provider artist ID.
3490        :return: List of Album objects.
3491        """
3492        albums = await self.client.get_artist_albums(prov_artist_id)
3493        result = []
3494        for album in albums:
3495            try:
3496                result.append(parse_album(self, album))
3497            except InvalidDataError as err:
3498                self.logger.debug("Error parsing artist album: %s", err)
3499        return result
3500
3501    @use_cache(3600 * 24 * 7, allow_expired_cache=True)
3502    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
3503        """
3504        Get artist's top tracks.
3505
3506        :param prov_artist_id: The provider artist ID.
3507        :return: List of Track objects.
3508        """
3509        tracks = await self.client.get_artist_tracks(prov_artist_id)
3510        result = []
3511        for track in tracks:
3512            try:
3513                result.append(parse_track(self, track))
3514            except InvalidDataError as err:
3515                self.logger.debug("Error parsing artist track: %s", err)
3516        return result
3517
3518    # Library methods
3519
3520    async def get_library_artists(self) -> AsyncGenerator[Artist]:
3521        """Retrieve library artists from Yandex Music."""
3522        artists = await self.client.get_liked_artists()
3523        for artist in artists:
3524            try:
3525                yield parse_artist(self, artist)
3526            except InvalidDataError as err:
3527                self.logger.debug("Error parsing library artist: %s", err)
3528
3529    async def _get_liked_albums_cached(self, ttl: float = 30.0) -> list[YandexAlbum]:
3530        """
3531        Return liked albums with a short in-process TTL cache + lock.
3532
3533        Albums, podcasts and audiobooks are all derived from the same
3534        ``users/{uid}/likes/albums`` endpoint, so a full library sync would
3535        otherwise trigger three sequential (or concurrent) identical calls.
3536        The lock serializes refreshes so only one request hits the API when
3537        multiple library syncs start together.
3538        """
3539        async with self._liked_albums_lock:
3540            now = asyncio.get_running_loop().time()
3541            if self._liked_albums_cache is not None:
3542                cached_at, cached = self._liked_albums_cache
3543                if now - cached_at < ttl:
3544                    return cached
3545            albums = await self.client.get_liked_albums(batch_size=TRACK_BATCH_SIZE)
3546            self._liked_albums_cache = (now, albums)
3547            return albums
3548
3549    async def get_library_albums(self) -> AsyncGenerator[Album]:
3550        """
3551        Retrieve library albums from Yandex Music.
3552
3553        Excludes entries classified as podcasts or audiobooks so they don't
3554        duplicate into the Albums library view.
3555        """
3556        for album in await self._get_liked_albums_cached():
3557            if classify_album(album) != "music":
3558                continue
3559            try:
3560                yield parse_album(self, album)
3561            except InvalidDataError as err:
3562                self.logger.debug("Error parsing library album: %s", err)
3563
3564    async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
3565        """Retrieve library podcasts from Yandex Music (filtered liked albums)."""
3566        for album in await self._get_liked_albums_cached():
3567            if classify_album(album) != "podcast":
3568                continue
3569            try:
3570                yield parse_podcast(self, album)
3571            except InvalidDataError as err:
3572                self.logger.debug("Error parsing library podcast: %s", err)
3573
3574    async def get_library_audiobooks(self) -> AsyncGenerator[Audiobook]:
3575        """Retrieve library audiobooks from Yandex Music (filtered liked albums)."""
3576        for album in await self._get_liked_albums_cached():
3577            if classify_album(album) != "audiobook":
3578                continue
3579            try:
3580                yield parse_audiobook(self, album)
3581            except InvalidDataError as err:
3582                self.logger.debug("Error parsing library audiobook: %s", err)
3583
3584    async def get_library_tracks(self) -> AsyncGenerator[Track]:
3585        """Retrieve library tracks from Yandex Music."""
3586        track_shorts = await self.client.get_liked_tracks()
3587        if not track_shorts:
3588            return
3589
3590        # Fetch full track details in batches
3591        track_ids = [str(ts.track_id) for ts in track_shorts if ts.track_id]
3592        batch_size = TRACK_BATCH_SIZE
3593        for i in range(0, len(track_ids), batch_size):
3594            batch_ids = track_ids[i : i + batch_size]
3595            full_tracks = await self.client.get_tracks(batch_ids)
3596            for track in full_tracks:
3597                try:
3598                    yield parse_track(self, track)
3599                except InvalidDataError as err:
3600                    self.logger.debug("Error parsing library track: %s", err)
3601
3602    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
3603        """
3604        Retrieve library playlists from Yandex Music.
3605
3606        Includes virtual playlists (My Wave and Liked Tracks if enabled), user-created playlists,
3607        and user-liked editorial playlists (returned by a separate API endpoint).
3608        """
3609        yield await self.get_playlist(MY_WAVE_PLAYLIST_ID)
3610        yield await self.get_playlist(LIKED_TRACKS_PLAYLIST_ID)
3611        seen_ids: set[str] = set()
3612        # User-created playlists
3613        playlists = await self.client.get_user_playlists()
3614        for playlist in playlists:
3615            try:
3616                parsed = parse_playlist(self, playlist)
3617                seen_ids.add(parsed.item_id)
3618                yield parsed
3619            except InvalidDataError as err:
3620                self.logger.debug("Error parsing library playlist: %s", err)
3621        # User-liked editorial playlists (not in users_playlists_list)
3622        liked_playlists = await self.client.get_liked_playlists()
3623        for playlist in liked_playlists:
3624            try:
3625                parsed = parse_playlist(self, playlist)
3626                if parsed.item_id not in seen_ids:
3627                    yield parsed
3628            except InvalidDataError as err:
3629                self.logger.debug("Error parsing liked playlist: %s", err)
3630
3631    # Library edit methods
3632
3633    async def library_add(self, item: MediaItemType) -> bool:
3634        """
3635        Add item to library.
3636
3637        For tracks carrying a wave station context in the item_id (e.g. when
3638        the user adds a My Wave track to favourites during playback), also
3639        fires a rotor ``like`` feedback on the active session so the wave
3640        algorithm biases toward similar tracks immediately.
3641
3642        :param item: The media item to add.
3643        :return: True if successful.
3644        """
3645        prov_item_id = self._get_provider_item_id(item)
3646        if not prov_item_id:
3647            return False
3648        track_id, station_key = _parse_radio_item_id(prov_item_id)
3649
3650        if item.media_type == MediaType.TRACK:
3651            ok = await self.client.like_track(track_id)
3652            if ok and station_key:
3653                wave = self._wave_states.get(station_key)
3654                if wave and wave.session_id:
3655                    await self._send_wave_feedback(wave, station_key, "like", track_id=track_id)
3656            return ok
3657        if item.media_type in (MediaType.ALBUM, MediaType.PODCAST, MediaType.AUDIOBOOK):
3658            return await self.client.like_album(prov_item_id)
3659        if item.media_type == MediaType.ARTIST:
3660            return await self.client.like_artist(prov_item_id)
3661        return False
3662
3663    async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
3664        """
3665        Remove item from library.
3666
3667        :param prov_item_id: The provider item ID (may be track_id@station_id for tracks).
3668        :param media_type: The media type.
3669        :return: True if successful.
3670        """
3671        track_id, _ = _parse_radio_item_id(prov_item_id)
3672        if media_type == MediaType.TRACK:
3673            return await self.client.unlike_track(track_id)
3674        if media_type in (MediaType.ALBUM, MediaType.PODCAST, MediaType.AUDIOBOOK):
3675            return await self.client.unlike_album(prov_item_id)
3676        if media_type == MediaType.ARTIST:
3677            return await self.client.unlike_artist(prov_item_id)
3678        return False
3679
3680    def _get_provider_item_id(self, item: MediaItemType) -> str | None:
3681        """Get provider item ID from media item."""
3682        for mapping in item.provider_mappings:
3683            if mapping.provider_instance == self.instance_id:
3684                return mapping.item_id
3685        return item.item_id if item.provider == self.instance_id else None
3686
3687    # Streaming
3688
3689    async def get_stream_details(
3690        self, item_id: str, media_type: MediaType = MediaType.TRACK
3691    ) -> StreamDetails:
3692        """
3693        Get stream details for a track, podcast episode, or audiobook.
3694
3695        A podcast episode is a track underneath the Yandex API, so it flows
3696        through the same per-track streaming path. An audiobook is an album
3697        with multiple tracks (chapters) — returned as a CUSTOM stream whose
3698        generator concatenates each chapter's bytes in order.
3699
3700        :param item_id: The track / episode ID (or track_id@station_id for My Wave),
3701            or the audiobook (album) ID when ``media_type`` is AUDIOBOOK.
3702        :param media_type: The media type.
3703        :return: StreamDetails for the item.
3704        """
3705        if media_type == MediaType.AUDIOBOOK:
3706            return await self._get_audiobook_stream_details(item_id)
3707        return await self.streaming.get_stream_details(item_id)
3708
3709    async def _get_audiobook_stream_details(self, audiobook_id: str) -> StreamDetails:
3710        """
3711        Build StreamDetails for an audiobook as a chapter-concatenated CUSTOM stream.
3712
3713        Loads the album's tracks, uses the first chapter to establish the audio
3714        format, and stores the per-chapter track-IDs + durations in ``data`` so
3715        ``get_audio_stream`` can iterate them. ``can_seek=True`` so MA routes
3716        ``seek_position`` into ``get_audio_stream``, where the provider translates
3717        it into ``(start_chapter, in_chapter_offset)``. In-chapter precision
3718        requires a byte-seekable chapter codec (raw MP3); otherwise the chapter
3719        is restarted from its beginning.
3720        """
3721        album = await self.client.get_album_with_tracks(audiobook_id)
3722        if not album or not (album.volumes or []):
3723            raise MediaNotFoundError(f"Audiobook {audiobook_id} has no chapters")
3724
3725        chapter_ids, chapter_durations_ms = _extract_chapter_map_from_album(album)
3726        if not chapter_ids:
3727            raise MediaNotFoundError(f"Audiobook {audiobook_id} has no chapters")
3728
3729        self._audiobook_chapter_cache[audiobook_id] = (chapter_ids, chapter_durations_ms)
3730
3731        # Resolve first-chapter format so MA/ffmpeg know what it's decoding
3732        first = await self.streaming.get_stream_details(chapter_ids[0])
3733        total_duration = sum(chapter_durations_ms) // 1000
3734
3735        return StreamDetails(
3736            item_id=audiobook_id,
3737            provider=self.instance_id,
3738            media_type=MediaType.AUDIOBOOK,
3739            audio_format=first.audio_format,
3740            stream_type=StreamType.CUSTOM,
3741            duration=total_duration,
3742            data={
3743                "chapter_ids": chapter_ids,
3744                "chapter_durations_ms": chapter_durations_ms,
3745            },
3746            can_seek=True,
3747            allow_seek=True,
3748        )
3749
3750    async def get_audio_stream(
3751        self, streamdetails: StreamDetails, seek_position: int = 0
3752    ) -> AsyncGenerator[bytes]:
3753        """
3754        Return the audio stream for the provider item.
3755
3756        For tracks and podcast episodes, streams via windowed Range requests
3757        (raw or AES-CTR encrypted). For audiobooks, iterates chapters: each
3758        chapter's bytes are streamed through the per-track path and concatenated.
3759
3760        :param streamdetails: Stream details with URL and optional decryption key.
3761        :param seek_position: Seek position in seconds (handled by provider for raw transport).
3762        :return: Async generator yielding audio chunks.
3763        """
3764        data = streamdetails.data if isinstance(streamdetails.data, dict) else None
3765        if streamdetails.media_type == MediaType.AUDIOBOOK and data and "chapter_ids" in data:
3766            async for chunk in self._stream_audiobook_chapters(data, seek_position):
3767                yield chunk
3768            return
3769        async for chunk in self.streaming.get_audio_stream(streamdetails, seek_position):
3770            yield chunk
3771
3772    def _resolve_audiobook_seek(
3773        self, chapter_durations_ms: list[int], seek_position: int, n_chapters: int
3774    ) -> tuple[int, int]:
3775        """Map an audiobook ``seek_position`` (seconds) to (start_idx, chapter_seek)."""
3776        if seek_position <= 0 or not chapter_durations_ms:
3777            return 0, 0
3778        accumulated_ms = 0
3779        seek_ms = seek_position * 1000
3780        for idx, dur_ms in enumerate(chapter_durations_ms):
3781            if accumulated_ms + dur_ms > seek_ms:
3782                return idx, (seek_ms - accumulated_ms) // 1000
3783            accumulated_ms += dur_ms
3784        # Seek past end — start at last chapter from 0
3785        return max(n_chapters - 1, 0), 0
3786
3787    async def _resolve_audiobook_chapter_map(
3788        self, audiobook_id: str
3789    ) -> tuple[list[str], list[int]]:
3790        """
3791        Return (chapter_track_ids, chapter_durations_ms) for an audiobook.
3792
3793        Served from an in-memory cache populated by ``_get_audiobook_stream_details``.
3794        On a miss (e.g. ``on_played`` fires before streaming has started), falls back
3795        to a fresh ``get_album_with_tracks`` call and refills the cache.
3796        """
3797        cached = self._audiobook_chapter_cache.get(audiobook_id)
3798        if cached is not None:
3799            return cached
3800        album = await self.client.get_album_with_tracks(audiobook_id)
3801        if not album or not (album.volumes or []):
3802            return [], []
3803        chapter_ids, chapter_durations_ms = _extract_chapter_map_from_album(album)
3804        self._audiobook_chapter_cache[audiobook_id] = (chapter_ids, chapter_durations_ms)
3805        return chapter_ids, chapter_durations_ms
3806
3807    async def _stream_audiobook_chapters(
3808        self, data: dict[str, Any], seek_position: int
3809    ) -> AsyncGenerator[bytes]:
3810        """
3811        Concatenate per-chapter streams of an audiobook.
3812
3813        Translates ``seek_position`` into (start_chapter, in_chapter_offset) and
3814        delegates each chapter to the per-track streaming path. In-chapter offset
3815        is only applied when the chapter codec is byte-seekable (``can_seek``);
3816        otherwise the chapter is restarted from its beginning. Tracks consecutive
3817        chapter failures and raises ``MediaNotFoundError`` once the threshold is
3818        exceeded, so playback never silently truncates.
3819        """
3820        chapter_ids: list[str] = list(data.get("chapter_ids") or [])
3821        chapter_durations_ms: list[int] = list(data.get("chapter_durations_ms") or [])
3822        if not chapter_ids:
3823            return
3824
3825        start_idx, chapter_seek = self._resolve_audiobook_seek(
3826            chapter_durations_ms, seek_position, len(chapter_ids)
3827        )
3828
3829        max_consecutive_failures = 3
3830        consecutive_failures = 0
3831        has_yielded_audio = False
3832        last_error: Exception | None = None
3833
3834        for idx in range(start_idx, len(chapter_ids)):
3835            chapter_id = chapter_ids[idx]
3836            requested_offset = chapter_seek if idx == start_idx else 0
3837            chapter_details: StreamDetails | None = None
3838            try:
3839                chapter_details = await self.streaming.get_stream_details(chapter_id)
3840            except asyncio.CancelledError:
3841                raise
3842            except Exception as err:
3843                last_error = err
3844                self.logger.warning(
3845                    "Audiobook chapter %d (%s) stream-details failed: %s",
3846                    idx + 1,
3847                    chapter_id,
3848                    err,
3849                )
3850
3851            if chapter_details is None:
3852                consecutive_failures += 1
3853                if consecutive_failures >= max_consecutive_failures:
3854                    raise MediaNotFoundError(
3855                        "Unable to stream audiobook: too many consecutive chapter failures"
3856                    ) from last_error
3857                continue
3858
3859            # Apply the in-chapter offset only when the chapter codec supports
3860            # byte-offset seeking; otherwise restart the chapter from 0 to avoid
3861            # decoding garbled bytes from mid-file of a container format.
3862            offset = requested_offset if chapter_details.can_seek else 0
3863            chapter_had_audio = False
3864            try:
3865                async for chunk in self.streaming.get_audio_stream(chapter_details, offset):
3866                    chapter_had_audio = True
3867                    has_yielded_audio = True
3868                    yield chunk
3869            except asyncio.CancelledError:
3870                raise
3871            except Exception as err:
3872                last_error = err
3873                self.logger.warning(
3874                    "Audiobook chapter %d (%s) stream failed mid-play: %s",
3875                    idx + 1,
3876                    chapter_id,
3877                    err,
3878                )
3879
3880            if chapter_had_audio:
3881                consecutive_failures = 0
3882                last_error = None
3883            else:
3884                consecutive_failures += 1
3885                if consecutive_failures >= max_consecutive_failures:
3886                    raise MediaNotFoundError(
3887                        "Unable to stream audiobook: too many consecutive chapter failures"
3888                    ) from last_error
3889
3890        if not has_yielded_audio:
3891            raise MediaNotFoundError(
3892                "Unable to stream audiobook: no playable chapters found"
3893            ) from last_error
3894
3895    async def get_rotor_station_tracks(
3896        self, station_id: str, queue: str | int | None = None
3897    ) -> tuple[list[Any], str | None]:
3898        """
3899        Fetch tracks from a rotor station using the session API.
3900
3901        Public surface — pinned by the ynison plugin
3902        (`YandexMusicProviderLike.get_rotor_station_tracks`). The
3903        ``(tracks, batch_id)`` return contract is kept for that caller even
3904        though batch_id is now a session-scoped identifier.
3905
3906        Routes to ``_fetch_rotor_session_batch`` so the wave session state
3907        (`session_id`, seen tracks, prefetch) is shared with our own Browse /
3908        on_played / on_streamed flows. ``queue`` is the most recently played
3909        track ID the external caller observed — we record it as the
3910        pagination cursor before calling through.
3911
3912        :param station_id: Rotor station ID (e.g. "user:onyourwave",
3913            "genre:rock", "mood:calm", "track:1234").
3914        :param queue: Last-played track ID for pagination. Ignored on the
3915            very first call (no session yet) but still recorded.
3916        :return: Tuple of (list of yandex tracks, batch_id or None).
3917        """
3918        wave = self._get_wave_state(station_id)
3919        # Cursor update + batch fetch run under the station's lock, matching
3920        # the discipline in browse / recommendations / prefetch. Without it,
3921        # ynison replenish racing with a concurrent MA browse could interleave
3922        # last_track_id writes and leave session_id / batch_id out of sync.
3923        async with wave.lock:
3924            if queue is not None:
3925                wave.last_track_id = str(queue)
3926            return await self._fetch_rotor_session_batch(wave, station_id)
3927
3928    def get_quality(self) -> str:
3929        """Return the configured audio quality tier (e.g. 'balanced', 'superb')."""
3930        quality = str(self.config.get_value(CONF_QUALITY) or QUALITY_BALANCED).strip().lower()
3931        if quality == "lossless":
3932            quality = QUALITY_SUPERB
3933        return quality
3934
3935    async def resolve_image(self, path: str) -> str | bytes:
3936        """
3937        Resolve wave cover image with background color fill for transparent PNGs.
3938
3939        If the image URL has an associated background color (stored in _wave_bg_colors),
3940        downloads the PNG from Yandex CDN and composites it on a solid color background
3941        using Pillow, returning JPEG bytes. Falls back to the original URL on any error.
3942
3943        :param path: Image URL (may include #rrggbb fragment used as cache key).
3944        :return: Composited JPEG bytes, or original path string as fallback.
3945        """
3946        bg_color = self._wave_bg_colors.get(path)
3947        if not bg_color:
3948            return path
3949
3950        # Strip the #color fragment before fetching the actual image
3951        fetch_url = path.split("#", maxsplit=1)[0] if "#" in path else path
3952        try:
3953            async with self.mass.http_session.get(fetch_url) as resp:
3954                resp.raise_for_status()
3955                raw = await resp.read()
3956        except Exception as err:
3957            self.logger.debug("Failed to fetch wave cover %s: %s", fetch_url, err)
3958            return fetch_url
3959
3960        def _composite() -> bytes:
3961            bg_clean = bg_color.lstrip("#")
3962            try:
3963                r = int(bg_clean[0:2], 16)
3964                g = int(bg_clean[2:4], 16)
3965                b = int(bg_clean[4:6], 16)
3966            except ValueError, IndexError:
3967                return raw
3968            fg = PilImage.open(BytesIO(raw)).convert("RGBA")
3969            bg = PilImage.new("RGBA", fg.size, (r, g, b, 255))
3970            bg.paste(fg, mask=fg)
3971            out = BytesIO()
3972            bg.convert("RGB").save(out, "JPEG", quality=92)
3973            return out.getvalue()
3974
3975        try:
3976            return await asyncio.to_thread(_composite)
3977        except Exception as err:
3978            self.logger.debug("Wave cover composite failed for %s: %s", fetch_url, err)
3979            return fetch_url
3980
3981    async def on_played(
3982        self,
3983        media_type: MediaType,
3984        prov_item_id: str,
3985        fully_played: bool,
3986        position: int,
3987        media_item: MediaItemType,
3988        is_playing: bool = False,
3989    ) -> None:
3990        """
3991        Report periodic playback updates.
3992
3993        - Audiobooks: persist chapter progress via play_audio so Yandex's
3994          own clients resume at the right point.
3995        - Wave tracks: send rotor ``trackStarted`` while actively playing and
3996          kick off a background prefetch so DSTM refill serves wave-curated
3997          tracks with no extra round-trip. DSTM itself is the user's toggle —
3998          the provider does not flip it.
3999
4000        Generic track history reporting is not attempted here — the only
4001        known channel Yandex writes into ``/handlers/music-history`` is a
4002        long-lived Ynison WebSocket session, which lives in the sibling
4003        yandex_ynison plugin. Regular tracks played through MA are therefore
4004        invisible to Listening History unless that plugin is also active.
4005        """
4006        if media_type == MediaType.AUDIOBOOK:
4007            await self._report_audiobook_progress(prov_item_id, position)
4008            return
4009        if media_type != MediaType.TRACK:
4010            return
4011        _, station_id = _parse_radio_item_id(prov_item_id)
4012        if station_id and is_playing:
4013            track_id, _ = _parse_radio_item_id(prov_item_id)
4014            wave = self._wave_states.get(station_id) or self._get_wave_state(station_id)
4015            await self._send_wave_feedback(wave, station_id, "trackStarted", track_id=track_id)
4016            self.mass.create_task(self._prefetch_rotor_session(station_id))
4017
4018    async def on_streamed(self, streamdetails: StreamDetails) -> None:
4019        """
4020        Report stream completion to Yandex.
4021
4022        - Audiobooks: a final ``play_audio`` with the absolute stream
4023          position so the last listening point is preserved across Yandex
4024          clients. Cleans up session state even when ``data`` was stripped.
4025        - Wave tracks (composite item_id carries a station suffix): a rotor
4026          ``trackFinished`` or ``skip`` event with the actual seconds streamed
4027          so Yandex can improve recommendations.
4028        """
4029        data = streamdetails.data if isinstance(streamdetails.data, dict) else None
4030        if streamdetails.media_type == MediaType.AUDIOBOOK:
4031            await self._report_audiobook_final(streamdetails, data or {})
4032            return
4033        if streamdetails.media_type != MediaType.TRACK:
4034            return
4035        track_id, station_id = _parse_radio_item_id(streamdetails.item_id)
4036        if not station_id:
4037            return
4038        seconds = int(streamdetails.seconds_streamed or 0)
4039        duration = int(streamdetails.duration or 0)
4040        feedback_type = "trackFinished" if duration and seconds >= max(0, duration - 10) else "skip"
4041        wave = self._wave_states.get(station_id) or self._get_wave_state(station_id)
4042        await self._send_wave_feedback(
4043            wave, station_id, feedback_type, track_id=track_id, total_played_seconds=seconds
4044        )
4045
4046    def _audiobook_progress_point(
4047        self,
4048        chapter_durations_ms: list[int],
4049        n_chapters: int,
4050        absolute_sec: int,
4051    ) -> tuple[int, int, int]:
4052        """
4053        Resolve an absolute book position into a play_audio-ready tuple.
4054
4055        Returns ``(chapter_idx, track_length_seconds, offset_seconds)``, applying
4056        two invariants Yandex cares about and that ``_resolve_audiobook_seek``
4057        alone doesn't guarantee:
4058
4059        - At/beyond end-of-book, map to end of the last chapter (not start),
4060          so Yandex's resume point doesn't rewind to the start of the final
4061          chapter on natural completion.
4062        - ``track_length_seconds`` is clamped to at least 1 and ``offset`` to
4063          ``[0, track_length_seconds]`` — a chapter with ``duration_ms=None``
4064          (coerced to 0 by the chapter-map builder) would otherwise send
4065          ``track_length_seconds=0`` and block progress from syncing.
4066        """
4067        absolute_sec = max(0, absolute_sec)
4068        total_duration_sec = sum(chapter_durations_ms) // 1000
4069        last_idx = max(n_chapters - 1, 0)
4070        if absolute_sec >= total_duration_sec > 0:
4071            idx = last_idx
4072            track_length_sec = max(1, chapter_durations_ms[idx] // 1000)
4073            offset = track_length_sec
4074        else:
4075            idx, offset_raw = self._resolve_audiobook_seek(
4076                chapter_durations_ms, absolute_sec, n_chapters
4077            )
4078            track_length_sec = max(1, chapter_durations_ms[idx] // 1000)
4079            offset = max(0, min(int(offset_raw), track_length_sec))
4080        return idx, track_length_sec, offset
4081
4082    async def _report_audiobook_progress(self, audiobook_id: str, position_sec: int) -> None:
4083        """
4084        Push current listening position of an audiobook to Yandex.
4085
4086        Resolves the playing chapter + offset from the cached chapter map, then
4087        calls play_audio so Yandex persists the position for cross-client resume.
4088
4089        Best-effort: any non-cancellation failure while resolving the chapter
4090        map (rate-limit, network blip, auth edge case bubbling out of
4091        ``_call_with_retry``) must never break pause/stop, so it is swallowed
4092        here in addition to the errors already absorbed inside
4093        ``api_client.play_audio``.
4094        """
4095        try:
4096            chapter_ids, chapter_durations_ms = await self._resolve_audiobook_chapter_map(
4097                audiobook_id
4098            )
4099        except asyncio.CancelledError:
4100            raise
4101        except Exception as err:
4102            self.logger.debug(
4103                "Skipping audiobook progress report for %s (chapter map resolution failed): %s",
4104                audiobook_id,
4105                err,
4106            )
4107            return
4108        if not chapter_ids:
4109            self.logger.debug(
4110                "Audiobook %s has no chapter map; skipping progress report", audiobook_id
4111            )
4112            return
4113        idx, track_length_sec, offset = self._audiobook_progress_point(
4114            chapter_durations_ms, len(chapter_ids), int(position_sec)
4115        )
4116        play_id = self._audiobook_play_ids.setdefault(audiobook_id, uuid.uuid4().hex)
4117        await self.client.play_audio(
4118            track_id=chapter_ids[idx],
4119            album_id=audiobook_id,
4120            play_id=play_id,
4121            track_length_seconds=track_length_sec,
4122            total_played_seconds=offset,
4123            end_position_seconds=offset,
4124        )
4125
4126    async def _report_audiobook_final(
4127        self, streamdetails: StreamDetails, data: dict[str, Any]
4128    ) -> None:
4129        """
4130        Send a closing play_audio for an audiobook stream.
4131
4132        Uses the streamdetails' own ``chapter_ids`` / ``chapter_durations_ms``
4133        (populated when the StreamDetails was created) to stay consistent with
4134        what was actually played, then clears the session play_id and drops
4135        the chapter-map cache entry so long-running instances can't grow the
4136        cache without bound as users play more audiobooks.
4137        """
4138        audiobook_id = streamdetails.item_id
4139        chapter_ids = data.get("chapter_ids") or []
4140        chapter_durations_ms = data.get("chapter_durations_ms") or []
4141        play_id = self._audiobook_play_ids.pop(audiobook_id, None) or uuid.uuid4().hex
4142        self._audiobook_chapter_cache.pop(audiobook_id, None)
4143        if not chapter_ids or not chapter_durations_ms:
4144            return
4145        absolute_sec = int(streamdetails.seek_position + (streamdetails.seconds_streamed or 0))
4146        idx, track_length_sec, offset = self._audiobook_progress_point(
4147            chapter_durations_ms, len(chapter_ids), absolute_sec
4148        )
4149        await self.client.play_audio(
4150            track_id=chapter_ids[idx],
4151            album_id=audiobook_id,
4152            play_id=play_id,
4153            track_length_seconds=track_length_sec,
4154            total_played_seconds=offset,
4155            end_position_seconds=offset,
4156        )
4157
4158    async def _rotating_row_tag_subtitle(self, category: str) -> str | None:
4159        """
4160        Return the display label of the current rotating tag for a mood/activity row.
4161
4162        Cache-only read of the validated tag list (rows must stay free of backend I/O):
4163        returns None - no subtitle - until an items fetch has warmed that cache.
4164
4165        :param category: Tag category ('mood' or 'activity').
4166        """
4167        # key mirrors the @use_cache key construction on _get_valid_tags_for_category:
4168        # the wrapped function's __name__ (preserved by functools.wraps, so it survives
4169        # renames) plus its positional args, joined by dots
4170        tags, _, found = await self.mass.cache.get_with_freshness(
4171            f"{self._get_valid_tags_for_category.__name__}.{category}",
4172            provider=self.instance_id,
4173            include_expired=True,
4174        )
4175        if not found or not tags:
4176            return None
4177        tag = self._rotating_row_tag(category, tags)
4178        return self._media_label("folder", _media_label_key(tag), tag.title())[0]
4179
4180    def _rotating_row_tag(self, category: str, valid_tags: list[str]) -> str:
4181        """
4182        Deterministically pick the current hour's tag for a mood/activity row.
4183
4184        Rows and items derive the same tag independently - no shared state, so
4185        concurrent clients (or multiple users on one instance) can never make the
4186        served items mismatch the row subtitle. The pick rotates hourly and
4187        differs per provider instance.
4188
4189        :param category: Tag category the tags belong to.
4190        :param valid_tags: Non-empty list of valid tag slugs to pick from.
4191        """
4192        hour_bucket = int(utc().timestamp()) // 3600
4193        seed = f"{self.instance_id}.{category}.{hour_bucket}".encode()
4194        return sorted(valid_tags)[zlib.crc32(seed) % len(valid_tags)]
4195