music-assistant-server

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