music-assistant-server

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