music-assistant-server

1"""KION Music provider implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import zlib
8from collections.abc import AsyncGenerator, Coroutine, Sequence
9from io import BytesIO
10from typing import TYPE_CHECKING, Any
11
12from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
13from music_assistant_models.enums import ConfigEntryType, ImageType, MediaType, ProviderFeature
14from music_assistant_models.errors import (
15    InvalidDataError,
16    LoginFailed,
17    MediaNotFoundError,
18    ProviderUnavailableError,
19    ResourceTemporarilyUnavailable,
20)
21from music_assistant_models.media_items import (
22    Album,
23    Artist,
24    BrowseFolder,
25    ItemMapping,
26    MediaItemImage,
27    MediaItemType,
28    Playlist,
29    ProviderMapping,
30    RecommendationFolder,
31    SearchResults,
32    Track,
33    UniqueList,
34)
35from PIL import Image as PilImage
36
37from music_assistant.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER
38from music_assistant.controllers.cache import use_cache
39from music_assistant.helpers.datetime import utc
40from music_assistant.models.music_provider import MusicProvider
41
42from .api_client import KionMusicClient
43from .constants import (
44    BROWSE_INITIAL_TRACKS,
45    COLLECTION_FOLDER_ID,
46    CONF_BASE_URL,
47    CONF_CODECS,
48    CONF_LIKED_TRACKS_MAX_TRACKS,
49    CONF_MY_WAVE_MAX_TRACKS,
50    CONF_QUALITY,
51    CONF_TOKEN,
52    CONF_TRANSPORT,
53    DEFAULT_BASE_URL,
54    DISCOVERY_INITIAL_TRACKS,
55    FOR_YOU_FOLDER_ID,
56    IMAGE_SIZE_MEDIUM,
57    LIKED_TRACKS_PLAYLIST_ID,
58    LISTENING_HISTORY_FOLDER_ID,
59    MY_WAVE_BATCH_SIZE,
60    MY_WAVE_PLAYLIST_ID,
61    MY_WAVES_FOLDER_ID,
62    MY_WAVES_SET_FOLDER_ID,
63    PINNED_ITEMS_FOLDER_ID,
64    PLAYLIST_ID_SPLITTER,
65    QUALITY_BALANCED,
66    QUALITY_EFFICIENT,
67    QUALITY_HIGH,
68    QUALITY_LOSSLESS,
69    RADIO_FOLDER_ID,
70    RADIO_TRACK_ID_SEP,
71    ROTOR_STATION_MY_MIX,
72    TAG_CATEGORY_ACTIVITY,
73    TAG_CATEGORY_ERA,
74    TAG_CATEGORY_GENRES,
75    TAG_CATEGORY_MOOD,
76    TAG_CATEGORY_ORDER,
77    TAG_MIXES,
78    TAG_SEASONAL_MAP,
79    TAG_SLUG_CATEGORY,
80    TRACK_BATCH_SIZE,
81    TRANSPORT_ENCRAW,
82    TRANSPORT_RAW,
83    WAVE_CATEGORY_DISPLAY_ORDER,
84    WAVES_FOLDER_ID,
85    WAVES_LANDING_FOLDER_ID,
86)
87from .parsers import (
88    _get_image_url as get_image_url,
89)
90from .parsers import (
91    get_canonical_provider_name,
92    parse_album,
93    parse_artist,
94    parse_playlist,
95    parse_track,
96)
97from .streaming import KionMusicStreamingManager
98
99if TYPE_CHECKING:
100    from music_assistant_models.streamdetails import StreamDetails
101
102
103def _parse_radio_item_id(item_id: str) -> tuple[str, str | None]:
104    """
105    Extract track_id and optional station_id from provider item_id.
106
107    My Mix tracks use item_id format 'track_id@station_id'. Other tracks use
108    plain track_id.
109
110    :param item_id: Provider item_id (may contain RADIO_TRACK_ID_SEP).
111    :return: (track_id, station_id or None).
112    """
113    if RADIO_TRACK_ID_SEP in item_id:
114        parts = item_id.split(RADIO_TRACK_ID_SEP, 1)
115        return (parts[0], parts[1] if len(parts) > 1 else None)
116    return (item_id, None)
117
118
119# Collection sub-folder browse ids -> (ProviderFeature, library sub_id, label key, English name).
120# The library sub_id ("tracks") and label key ("my_favorites") differ on purpose so the Collection
121# labels stay distinct from the core "media.folder.*" library labels.
122_COLLECTION_SUBFOLDERS: tuple[tuple[ProviderFeature, str, str, str], ...] = (
123    (ProviderFeature.LIBRARY_TRACKS, "tracks", "my_favorites", "My Favorites"),
124    (ProviderFeature.LIBRARY_ARTISTS, "artists", "my_artists", "My Artists"),
125    (ProviderFeature.LIBRARY_ALBUMS, "albums", "my_albums", "My Albums"),
126    (ProviderFeature.LIBRARY_PLAYLISTS, "playlists", "my_playlists", "My Playlists"),
127)
128
129
130def _media_label_key(slug: str) -> str:
131    """Normalize a tag/category slug into its strings.json authoring key (spaces → underscores)."""
132    return slug.replace(" ", "_")
133
134
135class _WaveState:
136    """Per-station mutable state for rotor wave playback."""
137
138    def __init__(self) -> None:
139        self.batch_id: str | None = None
140        self.last_track_id: str | None = None
141        self.seen_track_ids: set[str] = set()
142        self.radio_started_sent: bool = False
143        self.lock: asyncio.Lock = asyncio.Lock()
144
145
146class KionMusicProvider(MusicProvider):
147    """Implementation of a KION Music MusicProvider."""
148
149    _client: KionMusicClient | None = None
150    _streaming: KionMusicStreamingManager | None = None
151    _my_wave_batch_id: str | None = None
152    _my_wave_last_track_id: str | None = None  # last track id for "Load more" (API queue param)
153    _my_wave_playlist_next_cursor: str | None = None  # first_track_id for next playlist page
154    _my_wave_radio_started_sent: bool = False
155    _my_wave_seen_track_ids: set[str]  # Track IDs seen in current My Mix session
156    _my_wave_lock: asyncio.Lock  # Protects My Mix mutable state
157    _wave_states: dict[str, _WaveState]  # Per-station state for tagged wave stations
158    _wave_bg_colors: dict[str, str]  # image_url -> hex bg color for transparent covers
159
160    @property
161    def client(self) -> KionMusicClient:
162        """Return the KION Music client."""
163        if self._client is None:
164            raise ProviderUnavailableError("Provider not initialized")
165        return self._client
166
167    @property
168    def streaming(self) -> KionMusicStreamingManager:
169        """Return the streaming manager."""
170        if self._streaming is None:
171            raise ProviderUnavailableError("Provider not initialized")
172        return self._streaming
173
174    async def get_recommendations(self) -> list[RecommendationFolder]:
175        """
176        Get the available recommendation rows, without items.
177
178        Returns My Mix, Made for you, Chart, New Releases, New Playlists,
179        Top Picks, Mood Mix, Activity Mix and Seasonal Mix rows.
180        """
181        # The seasonal row title carries the current season, derived locally from the month.
182        seasonal_tag = TAG_SEASONAL_MAP.get(utc().month, "autumn")
183        seasonal_name = (
184            self._media_source_name("folder", _media_label_key(seasonal_tag))
185            or seasonal_tag.title()
186        )
187        return [
188            RecommendationFolder(
189                item_id=MY_WAVE_PLAYLIST_ID,
190                provider=self.instance_id,
191                name="My Mix",
192                translation_key=MY_WAVE_PLAYLIST_ID,
193                icon="mdi-waveform",
194            ),
195            RecommendationFolder(
196                item_id="feed",
197                provider=self.instance_id,
198                name="Made for you",
199                translation_key="made_for_you",
200                icon="mdi-account-music",
201            ),
202            RecommendationFolder(
203                item_id="chart",
204                provider=self.instance_id,
205                name="Chart",
206                translation_key="chart",
207                icon="mdi-chart-line",
208            ),
209            RecommendationFolder(
210                item_id="new_releases",
211                provider=self.instance_id,
212                name="New Releases",
213                translation_key="new_releases",
214                icon="mdi-new-box",
215            ),
216            RecommendationFolder(
217                item_id="new_playlists",
218                provider=self.instance_id,
219                name="New Playlists",
220                translation_key="new_playlists",
221                icon="mdi-playlist-star",
222            ),
223            RecommendationFolder(
224                item_id="top_picks",
225                provider=self.instance_id,
226                name="Top Picks",
227                translation_key="top_picks",
228                icon="mdi-star",
229            ),
230            # Mood/Activity rows have a static title; the hourly rotating tag - derived
231            # deterministically, so the items call independently computes the same one -
232            # shows as the row subtitle (cache-only tag-list read, no backend I/O).
233            RecommendationFolder(
234                item_id="mood_mix",
235                provider=self.instance_id,
236                name="Mood Mix",
237                translation_key="mood_mix",
238                subtitle=await self._rotating_row_tag_subtitle("mood"),
239                icon="mdi-emoticon-outline",
240            ),
241            RecommendationFolder(
242                item_id="activity_mix",
243                provider=self.instance_id,
244                name="Activity Mix",
245                translation_key="activity_mix",
246                subtitle=await self._rotating_row_tag_subtitle("activity"),
247                icon="mdi-run",
248            ),
249            RecommendationFolder(
250                item_id="seasonal_mix",
251                provider=self.instance_id,
252                name=f"Seasonal: {seasonal_name}",
253                translation_key="seasonal_mix",
254                translation_params=[seasonal_name],
255                icon="mdi-weather-sunny",
256            ),
257        ]
258
259    async def get_recommendation_items(
260        self, item_id: str
261    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
262        """
263        Get the items for a single recommendation row.
264
265        :param item_id: The item_id of the row, as returned by get_recommendations.
266        """
267        folder: RecommendationFolder | None = None
268        if item_id == MY_WAVE_PLAYLIST_ID:
269            folder = await self._get_my_wave_recommendations()
270        elif item_id == "feed":
271            folder = await self._get_feed_recommendations()
272        elif item_id == "chart":
273            folder = await self._get_chart_recommendations()
274        elif item_id == "new_releases":
275            folder = await self._get_new_releases_recommendations()
276        elif item_id == "new_playlists":
277            folder = await self._get_new_playlists_recommendations()
278        elif item_id == "top_picks":
279            folder = await self._get_top_picks_recommendations()
280        elif item_id == "mood_mix":
281            # the deterministic hourly tag keeps the served items matching the row subtitle
282            if mood_tags := await self._get_valid_tags_for_category("mood"):
283                folder = await self._get_mood_mix_recommendations(
284                    self._rotating_row_tag("mood", mood_tags)
285                )
286        elif item_id == "activity_mix":
287            if activity_tags := await self._get_valid_tags_for_category("activity"):
288                folder = await self._get_activity_mix_recommendations(
289                    self._rotating_row_tag("activity", activity_tags)
290                )
291        elif item_id == "seasonal_mix":
292            folder = await self._get_seasonal_mix_recommendations()
293        if folder is None:
294            return UniqueList()
295        return folder.items
296
297    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
298        """
299        Return Config entries to configure this provider.
300
301        The token is collected by the interactive setup flow (see setup_flow.py); this
302        surface only exposes the genuine playback options.
303        """
304        return (
305            CONF_ENTRY_UNOFFICIAL_PROVIDER,
306            # Quality
307            ConfigEntry(
308                key=CONF_QUALITY,
309                type=ConfigEntryType.STRING,
310                options=[
311                    ConfigValueOption(QUALITY_EFFICIENT),
312                    ConfigValueOption(QUALITY_BALANCED),
313                    ConfigValueOption(QUALITY_HIGH),
314                    ConfigValueOption(QUALITY_LOSSLESS),
315                ],
316                default_value=QUALITY_BALANCED,
317            ),
318            # My Mix maximum tracks (advanced)
319            ConfigEntry(
320                key=CONF_MY_WAVE_MAX_TRACKS,
321                type=ConfigEntryType.INTEGER,
322                range=(10, 1000),
323                default_value=150,
324                required=False,
325                advanced=True,
326            ),
327            # Liked Tracks maximum tracks (advanced)
328            ConfigEntry(
329                key=CONF_LIKED_TRACKS_MAX_TRACKS,
330                type=ConfigEntryType.INTEGER,
331                range=(50, 2000),
332                default_value=500,
333                required=False,
334                advanced=True,
335            ),
336            # Transport mode (advanced)
337            ConfigEntry(
338                key=CONF_TRANSPORT,
339                type=ConfigEntryType.STRING,
340                options=[
341                    ConfigValueOption(TRANSPORT_RAW),
342                    ConfigValueOption(TRANSPORT_ENCRAW),
343                ],
344                default_value=TRANSPORT_RAW,
345                required=False,
346                advanced=True,
347            ),
348            # Custom codecs override (advanced)
349            ConfigEntry(
350                key=CONF_CODECS,
351                type=ConfigEntryType.STRING,
352                default_value="",
353                required=False,
354                advanced=True,
355            ),
356            # API Base URL (advanced)
357            ConfigEntry(
358                key=CONF_BASE_URL,
359                type=ConfigEntryType.STRING,
360                translation_params=[DEFAULT_BASE_URL],
361                default_value=DEFAULT_BASE_URL,
362                required=False,
363                advanced=True,
364            ),
365        )
366
367    async def handle_async_init(self) -> None:
368        """Handle async initialization of the provider."""
369        token = self.get_setup_value(CONF_TOKEN)
370        if not token:
371            raise LoginFailed("No KION Music token provided")
372
373        base_url = self.config.get_value(CONF_BASE_URL, DEFAULT_BASE_URL)
374        self._client = KionMusicClient(str(token), base_url=str(base_url))
375        await self._client.connect()
376        # Suppress kion_music library DEBUG dumps (full API request/response JSON)
377        logging.getLogger("yandex_music").setLevel(self.logger.level + 10)
378        self._streaming = KionMusicStreamingManager(self)
379        # Initialize My Mix duplicate tracking
380        self._my_wave_seen_track_ids = set()
381        self._my_wave_lock = asyncio.Lock()
382        # Initialize per-station wave state dict
383        self._wave_states = {}
384        self._wave_bg_colors = {}
385        self.logger.info("Successfully connected to KION Music")
386
387    async def unload(self, is_removed: bool = False) -> None:
388        """
389        Handle unload/close of the provider.
390
391        :param is_removed: Whether the provider is being removed.
392        """
393        if self._client:
394            await self._client.disconnect()
395        self._client = None
396        self._streaming = None
397        await super().unload(is_removed)
398
399    def get_item_mapping(self, media_type: MediaType | str, key: str, name: str) -> ItemMapping:
400        """
401        Create a generic item mapping.
402
403        :param media_type: The media type.
404        :param key: The item ID.
405        :param name: The item name.
406        :return: An ItemMapping instance.
407        """
408        if isinstance(media_type, str):
409            media_type = MediaType(media_type)
410        return ItemMapping(
411            media_type=media_type,
412            item_id=key,
413            provider=self.instance_id,
414            name=name,
415        )
416
417    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
418        """
419        Browse provider items.
420
421        Root level shows My Mix (personalised radio), For You (picks & mixes),
422        Collection (liked tracks/albums/artists/playlists), Radio (rotor stations
423        by genre/mood/activity/era/local) and AI Mix Sets. Folder labels carry a
424        translation_key so the server localizes them for the connection locale.
425        My Mix tracks use item_id format track_id@station_id for rotor feedback.
426
427        :param path: The path to browse (e.g. provider_id:// or provider_id://waves).
428        """
429        if ProviderFeature.BROWSE not in self.supported_features:
430            raise NotImplementedError
431
432        path_parts = path.split("://")[1].split("/") if "://" in path else []
433        subpath = path_parts[0] if len(path_parts) > 0 else None
434        sub_subpath = path_parts[1] if len(path_parts) > 1 else None
435
436        if subpath == MY_WAVE_PLAYLIST_ID:
437            async with self._my_wave_lock:
438                return await self._browse_my_wave(path, sub_subpath)
439
440        # For You folder (picks + mixes)
441        if subpath == FOR_YOU_FOLDER_ID:
442            return await self._browse_for_you(path, path_parts)
443
444        # Collection folder (library items)
445        if subpath == COLLECTION_FOLDER_ID:
446            return await self._browse_collection(path)
447
448        # Handle picks/ path (mood, activity, era, genres)
449        if subpath == "picks":
450            return await self._browse_picks(path, path_parts)
451
452        # Handle mixes/ path (seasonal collections)
453        if subpath == "mixes":
454            return await self._browse_mixes(path, path_parts)
455
456        # Handle waves/ and radio/ paths (rotor stations by genre/mood/activity)
457        if subpath in (WAVES_FOLDER_ID, RADIO_FOLDER_ID):
458            return await self._browse_waves(path, path_parts)
459
460        # Handle my_waves_set/ path (AI Mix Sets from /landing-blocks/mixes-waves)
461        if subpath == MY_WAVES_SET_FOLDER_ID:
462            return await self._browse_vibe_sets(path, path_parts)
463
464        # Handle waves_landing/ path (Featured Mixes from /landing-blocks/waves)
465        if subpath == WAVES_LANDING_FOLDER_ID:
466            return await self._browse_waves_landing(path, path_parts)
467
468        # Pinned items folder
469        if subpath == PINNED_ITEMS_FOLDER_ID:
470            return await self._browse_pins()
471
472        # Listening history folder
473        if subpath == LISTENING_HISTORY_FOLDER_ID:
474            return await self._browse_history()
475
476        # Handle direct tag subpath (when folder is played by URI, the full path
477        # "picks/category/tag" is lost and only the tag slug arrives as subpath).
478        # Skip the API call for standard top-level folders that are never tag slugs.
479        _known_folders = {
480            "artists",
481            "albums",
482            "tracks",
483            "playlists",
484            LIKED_TRACKS_PLAYLIST_ID,
485            WAVES_FOLDER_ID,
486            RADIO_FOLDER_ID,
487            MY_WAVES_FOLDER_ID,
488            MY_WAVES_SET_FOLDER_ID,
489            WAVES_LANDING_FOLDER_ID,
490            FOR_YOU_FOLDER_ID,
491            COLLECTION_FOLDER_ID,
492            PINNED_ITEMS_FOLDER_ID,
493            LISTENING_HISTORY_FOLDER_ID,
494        }
495        if subpath and subpath not in _known_folders:
496            # Handle direct wave station_id (e.g. "activity:workout") passed when
497            # MA plays a wave station folder using its item_id as the path subpath.
498            # Station IDs have format "category:tag" where category is non-numeric.
499            if ":" in subpath:
500                cat_part = subpath.split(":", 1)[0]
501                if not cat_part.isdigit():
502                    return await self._browse_wave_station(subpath)
503
504            discovered_tags = await self._get_discovered_tag_slugs()
505            if subpath in discovered_tags:
506                return await self._get_tag_playlists_as_browse(subpath)
507
508        if subpath:
509            return await super().browse(path)
510
511        # Each folder carries an English name plus a translation_key; the server localizes the
512        # name for the connection locale at serialization, falling back to the English name.
513        folders: list[BrowseFolder] = []
514        base = path if path.endswith("//") else path.rstrip("/") + "/"
515        # My Mix folder (always enabled — Яндекс «Мой микс»)
516        folders.append(
517            BrowseFolder(
518                item_id=MY_WAVE_PLAYLIST_ID,
519                provider=self.instance_id,
520                path=f"{base}{MY_WAVE_PLAYLIST_ID}",
521                name="My Mix",
522                translation_key=MY_WAVE_PLAYLIST_ID,
523                is_playable=True,
524            )
525        )
526        # For You folder — Picks + Mixes (Яндекс «Для вас»)
527        folders.append(
528            BrowseFolder(
529                item_id=FOR_YOU_FOLDER_ID,
530                provider=self.instance_id,
531                path=f"{base}{FOR_YOU_FOLDER_ID}",
532                name="For You",
533                translation_key=FOR_YOU_FOLDER_ID,
534                is_playable=False,
535            )
536        )
537        # Collection folder — library items (Яндекс «Коллекция»)
538        has_library = any(
539            f in self.supported_features
540            for f in (
541                ProviderFeature.LIBRARY_ARTISTS,
542                ProviderFeature.LIBRARY_ALBUMS,
543                ProviderFeature.LIBRARY_TRACKS,
544                ProviderFeature.LIBRARY_PLAYLISTS,
545            )
546        )
547        if has_library:
548            folders.append(
549                BrowseFolder(
550                    item_id=COLLECTION_FOLDER_ID,
551                    provider=self.instance_id,
552                    path=f"{base}{COLLECTION_FOLDER_ID}",
553                    name="Collection",
554                    translation_key=COLLECTION_FOLDER_ID,
555                    is_playable=False,
556                )
557            )
558        # Radio folder — rotor stations (reuses the shared "Radio" label)
559        folders.append(
560            BrowseFolder(
561                item_id=RADIO_FOLDER_ID,
562                provider=self.instance_id,
563                path=f"{base}{RADIO_FOLDER_ID}",
564                name="Radio",
565                translation_key="radios",
566                is_playable=False,
567            )
568        )
569        # AI Mix Sets — parametric stations from /landing-blocks/mixes-waves
570        folders.append(
571            BrowseFolder(
572                item_id=MY_WAVES_SET_FOLDER_ID,
573                provider=self.instance_id,
574                path=f"{base}{MY_WAVES_SET_FOLDER_ID}",
575                name="AI Mix Sets",
576                translation_key=MY_WAVES_SET_FOLDER_ID,
577                is_playable=False,
578            )
579        )
580        # Pinned items — user-pinned artists/albums/playlists/waves
581        folders.append(
582            BrowseFolder(
583                item_id=PINNED_ITEMS_FOLDER_ID,
584                provider=self.instance_id,
585                path=f"{base}{PINNED_ITEMS_FOLDER_ID}",
586                name="Pinned",
587                translation_key=PINNED_ITEMS_FOLDER_ID,
588                is_playable=False,
589            )
590        )
591        # Listening history — recently played tracks/albums
592        folders.append(
593            BrowseFolder(
594                item_id=LISTENING_HISTORY_FOLDER_ID,
595                provider=self.instance_id,
596                path=f"{base}{LISTENING_HISTORY_FOLDER_ID}",
597                name="Listening History",
598                translation_key=LISTENING_HISTORY_FOLDER_ID,
599                is_playable=False,
600            )
601        )
602        if len(folders) == 1:
603            return await self.browse(folders[0].path)
604        return folders
605
606    def _media_source_name(self, group: str, key: str) -> str | None:
607        """
608        Return the authored English ``name`` for *key* in *group*, or None when not authored.
609
610        :param group: Media translation group (``folder``, ``recommendations`` or ``playlist``).
611        :param key: Authoring key within the group.
612        """
613        return self.mass.translations.get_translation(
614            f"provider.{self.domain}.media.{group}.{key}.name"
615        )
616
617    def _media_label(self, group: str, key: str, fallback: str) -> tuple[str, str | None]:
618        """
619        Map a browse label to the in-code ``name`` and ``translation_key`` for its media item.
620
621        Authored keys return ``(English source name, key)`` — the English name from the
622        provider's ``strings.json`` plus a ``translation_key`` so the server localizes it for
623        the connection locale at serialization. An unauthored key — e.g. a tag discovered from
624        KION's landing API — returns ``(fallback, None)`` so its already-localized name is kept.
625
626        :param group: Media translation group (``folder``, ``recommendations`` or ``playlist``).
627        :param key: Authoring key within the group; also the item's ``translation_key``.
628        :param fallback: English name to use when no string is authored for *key*.
629        """
630        source = self._media_source_name(group, key)
631        if source is None:
632            return fallback, None
633        return source, key
634
635    async def _browse_my_wave(
636        self, path: str, sub_subpath: str | None
637    ) -> list[Track | BrowseFolder]:
638        """
639        Browse My Mix tracks (must be called under _my_wave_lock).
640
641        :param path: Full browse path.
642        :param sub_subpath: Sub-path part ('next' for load more, or track_id cursor).
643        :return: List of Track and optional BrowseFolder for "Load more".
644        """
645        max_tracks_config = int(
646            self.config.get_value(CONF_MY_WAVE_MAX_TRACKS) or 150  # type: ignore[arg-type]
647        )
648        batch_size_config = MY_WAVE_BATCH_SIZE
649
650        # Effective limit on tracks to collect for this call:
651        # initial browse is capped to BROWSE_INITIAL_TRACKS to avoid marking
652        # extra tracks as "seen" that are never shown to the user.
653        effective_limit = min(
654            BROWSE_INITIAL_TRACKS if sub_subpath != "next" else max_tracks_config,
655            max_tracks_config,
656        )
657
658        # Root my_wave: fetch up to batch_size_config batches so Play adds more tracks.
659        # "Load more" always uses single next batch.
660        max_batches = batch_size_config if sub_subpath != "next" else 1
661
662        # Reset seen tracks on fresh browse (not "load more")
663        if sub_subpath != "next":
664            self._my_wave_seen_track_ids = set()
665
666        queue: str | int | None = None
667        if sub_subpath == "next":
668            queue = self._my_wave_last_track_id
669        elif sub_subpath:
670            queue = sub_subpath
671
672        all_tracks: list[Track | BrowseFolder] = []
673        last_batch_id: str | None = None
674        first_track_id_this_batch: str | None = None
675        total_track_count = 0
676
677        for _ in range(max_batches):
678            if total_track_count >= effective_limit:
679                break
680
681            raw_tracks, batch_id = await self.client.get_my_wave_tracks(queue=queue)
682            if batch_id:
683                self._my_wave_batch_id = batch_id
684                last_batch_id = batch_id
685            if not self._my_wave_radio_started_sent and raw_tracks:
686                sent = await self.client.send_rotor_station_feedback(
687                    ROTOR_STATION_MY_MIX,
688                    "radioStarted",
689                    batch_id=batch_id,
690                )
691                if sent:
692                    self._my_wave_radio_started_sent = True
693            first_track_id_this_batch = None
694            for yt in raw_tracks:
695                if total_track_count >= effective_limit:
696                    break
697
698                track = self._parse_my_wave_track(yt, self._my_wave_seen_track_ids)
699                if track is None:
700                    continue
701                all_tracks.append(track)
702                total_track_count += 1
703
704                track_id = track.item_id.split(RADIO_TRACK_ID_SEP, 1)[0]
705                if first_track_id_this_batch is None:
706                    first_track_id_this_batch = track_id
707
708            if first_track_id_this_batch is not None:
709                self._my_wave_last_track_id = first_track_id_this_batch
710            if (
711                first_track_id_this_batch is None
712                or not batch_id
713                or not raw_tracks
714                or total_track_count >= effective_limit
715            ):
716                break
717            queue = first_track_id_this_batch
718
719        # Only show "Load more" if we haven't reached the limit and there's more data
720        if last_batch_id and total_track_count < max_tracks_config:
721            all_tracks.append(
722                BrowseFolder(
723                    item_id="next",
724                    provider=self.instance_id,
725                    path=f"{path.rstrip('/')}/next",
726                    name="Load more",
727                    translation_key="load_more",
728                    is_playable=False,
729                )
730            )
731        return all_tracks
732
733    def _parse_my_wave_track(self, yt: Any, seen_ids: set[str]) -> Track | None:
734        """
735        Parse a Kion track into a My Mix Track with composite item_id.
736
737        Extracts the track_id, checks for duplicates in the seen_ids set,
738        sets composite item_id (track_id@station_id), and updates provider_mappings.
739        Callers using shared state must hold _my_wave_lock.
740
741        :param yt: Kion track object from rotor station response.
742        :param seen_ids: Set of already-seen track IDs to check and update.
743        :return: Parsed Track with composite item_id, or None if duplicate/invalid.
744        """
745        try:
746            t = parse_track(self, yt)
747        except InvalidDataError as err:
748            self.logger.debug("Error parsing My Mix track: %s", err)
749            return None
750
751        track_id = str(yt.id) if hasattr(yt, "id") and yt.id else getattr(yt, "track_id", None)
752        if not track_id:
753            return t
754
755        if track_id in seen_ids:
756            self.logger.debug("Skipping duplicate My Mix track: %s", track_id)
757            return None
758
759        seen_ids.add(track_id)
760        t.item_id = f"{track_id}{RADIO_TRACK_ID_SEP}{ROTOR_STATION_MY_MIX}"
761        for pm in t.provider_mappings:
762            if pm.provider_instance == self.instance_id:
763                pm.item_id = t.item_id
764                break
765        return t
766
767    @use_cache(3600)
768    async def _validate_tag(self, tag_slug: str) -> bool:
769        """
770        Check if a tag has playlists by calling client.get_tag_playlists().
771
772        :param tag_slug: Tag identifier (e.g. 'chill', '80s').
773        :return: True if the tag has at least one playlist.
774        """
775        try:
776            playlists = await self.client.get_tag_playlists(tag_slug)
777            return len(playlists) > 0
778        except Exception as err:
779            self.logger.debug("Tag validation failed for %s: %s", tag_slug, err)
780            return False
781
782    # allow_expired_cache keeps the items call serving the same (possibly stale) tag
783    # list the rows subtitle was derived from, while a background refresh runs
784    @use_cache(3600, allow_expired_cache=True)
785    async def _get_valid_tags_for_category(self, category: str) -> list[str]:
786        """
787        Get validated tags for a category (only those with playlists).
788
789        Combines hardcoded tags from the category lists with any landing-discovered
790        tags, validates each by calling client.tags(), and returns only those with
791        playlists.
792
793        :param category: Category name ('mood', 'activity', 'era', 'genres').
794        :return: List of valid tag slugs.
795        """
796        category_lists: dict[str, list[str]] = {
797            "mood": list(TAG_CATEGORY_MOOD),
798            "activity": list(TAG_CATEGORY_ACTIVITY),
799            "era": list(TAG_CATEGORY_ERA),
800            "genres": list(TAG_CATEGORY_GENRES),
801        }
802        tags = category_lists.get(category, [])
803
804        # Add landing-discovered tags for this category
805        try:
806            landing_tags = await self.client.get_landing_tags()
807            for slug, _title in landing_tags:
808                cat = TAG_SLUG_CATEGORY.get(slug, "mood")
809                if cat == category and slug not in tags:
810                    tags.append(slug)
811        except Exception as err:
812            self.logger.debug("Landing tag discovery failed: %s", err)
813
814        # Validate tags in parallel with bounded concurrency
815        sem = asyncio.Semaphore(8)
816
817        async def _check(tag: str) -> str | None:
818            async with sem:
819                return tag if await self._validate_tag(tag) else None
820
821        results = await asyncio.gather(*[_check(tag) for tag in tags])
822        return [tag for tag in results if tag is not None]
823
824    @use_cache(3600)
825    async def _get_discovered_tags(self, locale: str) -> list[tuple[str, str]]:
826        """
827        Get all available tags by combining hardcoded tags with landing discovery.
828
829        Starts with all hardcoded tags from category lists, adds landing-discovered
830        tags, validates each via client.tags(), and returns only those with playlists.
831        Results are cached for 1 hour. The locale parameter is included in the cache
832        key so that a locale change invalidates the cached landing titles.
833
834        :param locale: Current metadata locale (used as part of cache key).
835        :return: List of (slug, fallback display name) tuples for tags that have playlists.
836            Hardcoded tags use a derived fallback; landing-discovered tags carry their
837            (already localized) API title. Folder sites attach the translation_key via
838            ``_media_label`` so authored tags localize and discovered ones keep their title.
839        """
840        # Collect all hardcoded tags (non-seasonal)
841        all_tags: dict[str, str] = {}
842        for slug, cat in TAG_SLUG_CATEGORY.items():
843            if cat != "seasonal":
844                all_tags[slug] = slug.title()
845
846        # Add landing-discovered tags
847        try:
848            landing_tags = await self.client.get_landing_tags()
849            for slug, title in landing_tags:
850                if slug not in all_tags:
851                    all_tags[slug] = title
852        except Exception as err:
853            self.logger.debug("Failed to discover tags from landing API: %s", err)
854
855        # Validate tags in parallel with bounded concurrency
856        sem = asyncio.Semaphore(8)
857
858        async def _check(slug: str) -> bool:
859            async with sem:
860                return await self._validate_tag(slug)
861
862        tag_items = list(all_tags.items())
863        results = await asyncio.gather(*[_check(slug) for slug, _ in tag_items])
864        return [
865            (slug, name) for (slug, name), valid in zip(tag_items, results, strict=True) if valid
866        ]
867
868    async def _get_discovered_tag_slugs(self) -> set[str]:
869        """
870        Get set of all valid tag slugs (cached).
871
872        :return: Set of tag slug strings that have playlists.
873        """
874        discovered = await self._get_discovered_tags(self.mass.metadata.locale or "en_US")
875        return {slug for slug, _name in discovered}
876
877    async def _browse_for_you(
878        self, path: str, path_parts: list[str]
879    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
880        """
881        Browse «For You» folder — shows Picks and Mixes sub-folders.
882
883        :param path: Full browse path.
884        :param path_parts: Split path parts after ://.
885        :return: List of sub-folders (Picks, Mixes).
886        """
887        # Strip the for_you segment to build child paths that route to picks/mixes
888        # Path format: ...//for_you  → child paths should be ...//picks, ...//mixes
889        # We build base from the root (before for_you) by dropping the last segment.
890        base_parts = path.split("//", 1)
891        root_base = (base_parts[0] + "//") if len(base_parts) > 1 else path.rstrip("/") + "/"
892
893        if len(path_parts) == 1:
894            return [
895                BrowseFolder(
896                    item_id="picks",
897                    provider=self.instance_id,
898                    path=f"{root_base}picks",
899                    name="Picks",
900                    translation_key="picks",
901                    is_playable=False,
902                ),
903                BrowseFolder(
904                    item_id="mixes",
905                    provider=self.instance_id,
906                    path=f"{root_base}mixes",
907                    name="Mixes",
908                    translation_key="mixes",
909                    is_playable=False,
910                ),
911            ]
912        # Deeper path: delegate to picks or mixes handler via canonical paths
913        return await super().browse(path)
914
915    async def _browse_collection(
916        self, path: str
917    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
918        """
919        Browse «Collection» folder — shows library sub-folders (tracks/artists/albums/playlists).
920
921        :param path: Full browse path.
922        :return: List of library sub-folders.
923        """
924        base_parts = path.split("//", 1)
925        root_base = (base_parts[0] + "//") if len(base_parts) > 1 else path.rstrip("/") + "/"
926
927        folders: list[BrowseFolder] = []
928        for feature, sub_id, label_key, label_name in _COLLECTION_SUBFOLDERS:
929            if feature not in self.supported_features:
930                continue
931            folders.append(
932                BrowseFolder(
933                    item_id=sub_id,
934                    provider=self.instance_id,
935                    path=f"{root_base}{sub_id}",
936                    name=label_name,
937                    translation_key=label_key,
938                    is_playable=True,
939                )
940            )
941        return folders
942
943    async def _browse_pins(self) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
944        """
945        Browse user's pinned items (artists/albums/playlists from Kion Pins).
946
947        Resolves each pin to its full media item via existing single-item lookups.
948        Wave pins are skipped — MA has no native concept for them.
949
950        Pins are resolved concurrently via ``asyncio.gather`` so latency is
951        dominated by the slowest lookup rather than their sum. Individual
952        failures (``MediaNotFoundError`` / ``InvalidDataError``) are skipped
953        without aborting the batch.
954
955        :return: List of resolved media items.
956        """
957        pins_list = await self.client.get_pins()
958        pins = getattr(pins_list, "pins", None) if pins_list else None
959        if not pins:
960            return []
961
962        tasks: list[Coroutine[Any, Any, MediaItemType]] = []
963        pin_descs: list[str] = []
964        for pin in pins:
965            pin_type = getattr(pin, "type", None)
966            data = getattr(pin, "data", None)
967            if data is None:
968                continue
969            if pin_type == "artist_item" and getattr(data, "id", None) is not None:
970                tasks.append(self.get_artist(str(data.id)))
971                pin_descs.append(f"artist:{data.id}")
972            elif pin_type == "album_item" and getattr(data, "id", None) is not None:
973                tasks.append(self.get_album(str(data.id)))
974                pin_descs.append(f"album:{data.id}")
975            elif pin_type == "playlist_item":
976                uid = getattr(data, "uid", None)
977                kind = getattr(data, "kind", None)
978                if uid is not None and kind is not None:
979                    tasks.append(self.get_playlist(f"{uid}:{kind}"))
980                    pin_descs.append(f"playlist:{uid}:{kind}")
981
982        if not tasks:
983            return []
984
985        results = await asyncio.gather(*tasks, return_exceptions=True)
986        items: list[MediaItemType] = []
987        for desc, result in zip(pin_descs, results, strict=True):
988            if isinstance(result, (MediaNotFoundError, InvalidDataError)):
989                self.logger.debug("Skipping pin %s: %s", desc, result)
990            elif isinstance(result, BaseException):
991                raise result
992            else:
993                items.append(result)
994        return items
995
996    async def _browse_history(self) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
997        """
998        Browse user's recent listening history (flattened across days).
999
1000        Filters to ``type == "track"`` entries only — album/playlist context
1001        items in the history feed are dropped. Tracks are de-duplicated by
1002        id and returned in most-recent-first order.
1003
1004        :return: List of recently played Track items.
1005        """
1006        history = await self.client.get_music_history()
1007        tabs = getattr(history, "history_tabs", None) if history else None
1008        if not tabs:
1009            return []
1010
1011        seen_track_ids: set[str] = set()
1012        tracks: list[Track] = []
1013        for tab in tabs:
1014            groups = getattr(tab, "items", None) or []
1015            for group in groups:
1016                history_items = getattr(group, "tracks", None) or []
1017                for hist_item in history_items:
1018                    if getattr(hist_item, "type", None) != "track":
1019                        continue
1020                    full = getattr(getattr(hist_item, "data", None), "full_model", None)
1021                    if full is None or getattr(full, "id", None) is None:
1022                        continue
1023                    track_key = str(full.id)
1024                    if track_key in seen_track_ids:
1025                        continue
1026                    seen_track_ids.add(track_key)
1027                    try:
1028                        tracks.append(parse_track(self, full))
1029                    except InvalidDataError as err:
1030                        self.logger.debug("Skipping history track: %s", err)
1031        return tracks
1032
1033    async def _browse_picks(
1034        self, path: str, path_parts: list[str]
1035    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1036        """
1037        Browse picks folder using hardcoded tags validated against the API.
1038
1039        Tags are sourced from hardcoded category lists and landing API discovery,
1040        then validated via client.tags() to ensure they have playlists.
1041        Only categories with at least one valid tag are shown.
1042
1043        :param path: Full browse path.
1044        :param path_parts: Split path parts after ://.
1045        :return: List of folders or playlists.
1046        """
1047        base = path.rstrip("/") + "/"
1048
1049        # Get validated tags
1050        discovered = await self._get_discovered_tags(self.mass.metadata.locale or "en_US")
1051
1052        # Categorize valid tags, carrying each tag's (slug, fallback display name)
1053        categorized: dict[str, list[tuple[str, str]]] = {}
1054        for slug, fallback_name in discovered:
1055            cat = TAG_SLUG_CATEGORY.get(slug, "mood")
1056            # Skip seasonal tags — they belong in mixes, not picks
1057            if cat == "seasonal":
1058                continue
1059            categorized.setdefault(cat, []).append((slug, fallback_name))
1060
1061        # Sort tags within each category by preferred order
1062        for cat, cat_tags in categorized.items():
1063            order = TAG_CATEGORY_ORDER.get(cat, [])
1064            order_map = {s: i for i, s in enumerate(order)}
1065            cat_tags.sort(key=lambda t: order_map.get(t[0], len(order)))
1066
1067        # picks/ - show category folders (only those with valid tags)
1068        if len(path_parts) == 1:
1069            category_display_order = ["mood", "activity", "era", "genres"]
1070            folders: list[BrowseFolder] = []
1071            for cat in category_display_order:
1072                if cat in categorized:
1073                    name, translation_key = self._media_label("folder", cat, cat.title())
1074                    folders.append(
1075                        BrowseFolder(
1076                            item_id=cat,
1077                            provider=self.instance_id,
1078                            path=f"{base}{cat}",
1079                            name=name,
1080                            translation_key=translation_key,
1081                            is_playable=False,
1082                        )
1083                    )
1084            # Show any extra categories not in the standard order
1085            for cat in categorized:
1086                if cat not in category_display_order:
1087                    name, translation_key = self._media_label("folder", cat, cat.title())
1088                    folders.append(
1089                        BrowseFolder(
1090                            item_id=cat,
1091                            provider=self.instance_id,
1092                            path=f"{base}{cat}",
1093                            name=name,
1094                            translation_key=translation_key,
1095                            is_playable=False,
1096                        )
1097                    )
1098            return folders
1099
1100        category: str | None = path_parts[1] if len(path_parts) > 1 else None
1101        tag: str | None = path_parts[2] if len(path_parts) > 2 else None
1102
1103        self.logger.debug(
1104            "Browse picks: path=%s, category=%s, tag=%s",
1105            path,
1106            category,
1107            tag,
1108        )
1109
1110        # picks/category/ - show valid tag folders for this category
1111        if category and not tag:
1112            category_tags = categorized.get(category, [])
1113            folders = []
1114            for slug, fallback_name in category_tags:
1115                name, translation_key = self._media_label(
1116                    "folder", _media_label_key(slug), fallback_name
1117                )
1118                folders.append(
1119                    BrowseFolder(
1120                        item_id=slug,
1121                        provider=self.instance_id,
1122                        path=f"{base}{slug}",
1123                        name=name,
1124                        translation_key=translation_key,
1125                        is_playable=False,
1126                    )
1127                )
1128            self.logger.debug("Returning %d tag folders for category %s", len(folders), category)
1129            return folders
1130
1131        # picks/category/tag - show playlists for the tag
1132        if tag:
1133            discovered_slugs = {slug for slug, _name in discovered}
1134            if tag in discovered_slugs:
1135                self.logger.debug("Fetching playlists for tag: %s", tag)
1136                return await self._get_tag_playlists_as_browse(tag)
1137
1138        self.logger.debug("No match found, returning empty list")
1139        return []
1140
1141    async def _browse_mixes(
1142        self, path: str, path_parts: list[str]
1143    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1144        """
1145        Browse mixes folder (seasonal collections) using hardcoded tags.
1146
1147        Uses TAG_MIXES directly and validates each tag via client.tags()
1148        to check if it has playlists. Does not depend on landing API discovery.
1149
1150        :param path: Full browse path.
1151        :param path_parts: Split path parts after ://.
1152        :return: List of folders or playlists.
1153        """
1154        base = path.rstrip("/") + "/"
1155
1156        # Validate seasonal tags in parallel (no landing dependency)
1157        sem = asyncio.Semaphore(5)
1158
1159        async def _check(tag: str) -> str | None:
1160            async with sem:
1161                return tag if await self._validate_tag(tag) else None
1162
1163        results = await asyncio.gather(*[_check(t) for t in TAG_MIXES])
1164        available_mixes = [t for t in results if t is not None]
1165
1166        # mixes/ - show seasonal folders (only valid ones)
1167        if len(path_parts) == 1:
1168            folders = []
1169            for t in available_mixes:
1170                name, translation_key = self._media_label("folder", t, t.title())
1171                folders.append(
1172                    BrowseFolder(
1173                        item_id=t,
1174                        provider=self.instance_id,
1175                        path=f"{base}{t}",
1176                        name=name,
1177                        translation_key=translation_key,
1178                        is_playable=False,
1179                    )
1180                )
1181            return folders
1182
1183        # mixes/tag - show playlists for the tag
1184        tag = path_parts[1] if len(path_parts) > 1 else None
1185        if tag and tag in TAG_MIXES:
1186            return await self._get_tag_playlists_as_browse(tag)
1187
1188        return []
1189
1190    def _get_wave_state(self, station_id: str) -> _WaveState:
1191        """
1192        Get or create per-station wave state.
1193
1194        :param station_id: Rotor station ID (e.g. 'genre:rock', 'mood:chill').
1195        :return: _WaveState instance for this station.
1196        """
1197        return self._wave_states.setdefault(station_id, _WaveState())
1198
1199    async def _browse_waves(
1200        self, path: str, path_parts: list[str]
1201    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1202        """
1203        Browse waves folder (rotor stations by genre/mood/activity/epoch/local).
1204
1205        Fetches available stations from the Kion rotor API and groups them by category.
1206
1207        :param path: Full browse path.
1208        :param path_parts: Split path parts after ://.
1209        :return: List of folders or tracks.
1210        """
1211        base = path.rstrip("/") + "/"
1212
1213        # Station names come back from the rotor API already localized; the metadata locale
1214        # selects the API content language (not the folder labels, which use translation_key).
1215        locale = (self.mass.metadata.locale or "en_US").lower()
1216        language = "ru" if locale.startswith("ru") else "en"
1217
1218        all_stations = await self.client.get_wave_stations(language)
1219
1220        # Group stations by category, preserving image_url
1221        categorized: dict[str, list[tuple[str, str, str | None]]] = {}
1222        for station_id, cat_key, station_name, image_url in all_stations:
1223            categorized.setdefault(cat_key, []).append((station_id, station_name, image_url))
1224
1225        # waves/ — show category folders
1226        if len(path_parts) == 1:
1227            folders: list[BrowseFolder] = []
1228            # Personalized stations first — only show if dashboard returns stations
1229            dashboard_stations = await self._get_dashboard_stations_cached()
1230            if dashboard_stations:
1231                name, translation_key = self._media_label("folder", MY_WAVES_FOLDER_ID, "Personal")
1232                folders.append(
1233                    BrowseFolder(
1234                        item_id=MY_WAVES_FOLDER_ID,
1235                        provider=self.instance_id,
1236                        path=f"{base}{MY_WAVES_FOLDER_ID}",
1237                        name=name,
1238                        translation_key=translation_key,
1239                        is_playable=False,
1240                    )
1241                )
1242            # Featured Mixes — only show if landing-blocks/waves returns data
1243            waves_landing = await self._get_waves_landing_cached()
1244            if waves_landing:
1245                name, translation_key = self._media_label(
1246                    "folder", WAVES_LANDING_FOLDER_ID, "Featured Mixes"
1247                )
1248                folders.append(
1249                    BrowseFolder(
1250                        item_id=WAVES_LANDING_FOLDER_ID,
1251                        provider=self.instance_id,
1252                        path=f"{base}{WAVES_LANDING_FOLDER_ID}",
1253                        name=name,
1254                        translation_key=translation_key,
1255                        is_playable=False,
1256                    )
1257                )
1258            for cat in WAVE_CATEGORY_DISPLAY_ORDER:
1259                if cat in categorized:
1260                    name, translation_key = self._media_label("folder", cat, cat.title())
1261                    folders.append(
1262                        BrowseFolder(
1263                            item_id=cat,
1264                            provider=self.instance_id,
1265                            path=f"{base}{cat}",
1266                            name=name,
1267                            translation_key=translation_key,
1268                            is_playable=False,
1269                        )
1270                    )
1271            # Append any categories returned by API that aren't in the predefined order
1272            for cat in categorized:
1273                if cat not in WAVE_CATEGORY_DISPLAY_ORDER:
1274                    name, translation_key = self._media_label("folder", cat, cat.title())
1275                    folders.append(
1276                        BrowseFolder(
1277                            item_id=cat,
1278                            provider=self.instance_id,
1279                            path=f"{base}{cat}",
1280                            name=name,
1281                            translation_key=translation_key,
1282                            is_playable=False,
1283                        )
1284                    )
1285            return folders
1286
1287        category: str | None = path_parts[1] if len(path_parts) > 1 else None
1288        tag: str | None = path_parts[2] if len(path_parts) > 2 else None
1289
1290        # waves/my_waves/ — show personalized stations from dashboard
1291        if category == MY_WAVES_FOLDER_ID and not tag:
1292            return await self._browse_my_waves_stations(path)
1293
1294        # waves/waves_landing/... — redirect to Featured Mixes browse
1295        if category == WAVES_LANDING_FOLDER_ID:
1296            return await self._browse_waves_landing(path, path_parts[1:])
1297
1298        # waves/my_waves/<tag>[/next] — play a specific personal station
1299        # The full station_id has format "genre:allrock", not "my_waves:allrock".
1300        # Resolve by matching against dashboard stations cache.
1301        if category == MY_WAVES_FOLDER_ID and tag:
1302            dashboard_stations = await self._get_dashboard_stations_cached()
1303            for sid, _, _ in dashboard_stations:
1304                sid_tag = sid.split(":", 1)[1] if ":" in sid else sid
1305                if sid_tag == tag:
1306                    return await self._browse_wave_station(sid, path=path)
1307            # Fallback: try tag as direct station_id (e.g. "genre:allrock" passed verbatim)
1308            if ":" in tag:
1309                return await self._browse_wave_station(tag, path=path)
1310            return []
1311
1312        # waves/<category>/ — show station folders with artwork
1313        if category and not tag:
1314            cat_stations = categorized.get(category, [])
1315            folders = []
1316            for station_id, station_name, image_url in cat_stations:
1317                tag_part = station_id.split(":", 1)[1] if ":" in station_id else station_id
1318                station_image: MediaItemImage | None = None
1319                if image_url:
1320                    station_image = MediaItemImage(
1321                        type=ImageType.THUMB,
1322                        path=image_url,
1323                        provider=self.instance_id,
1324                        remotely_accessible=True,
1325                    )
1326                folders.append(
1327                    BrowseFolder(
1328                        item_id=station_id,
1329                        provider=self.instance_id,
1330                        path=f"{base}{tag_part}",
1331                        name=station_name,
1332                        is_playable=True,
1333                        image=station_image,
1334                    )
1335                )
1336            return folders
1337
1338        # waves/<category>/<tag>[/next] — stream tracks from rotor station
1339        if category and tag:
1340            station_id = f"{category}:{tag}"
1341            return await self._browse_wave_station(station_id, path=path)
1342
1343        return []
1344
1345    @use_cache(600)
1346    async def _get_dashboard_stations_cached(self) -> list[tuple[str, str, str | None]]:
1347        """
1348        Get personalized dashboard stations, cached for 10 minutes.
1349
1350        :return: List of (station_id, name, image_url) tuples.
1351        """
1352        return await self.client.get_dashboard_stations()
1353
1354    async def _browse_my_waves_stations(self, path: str) -> list[BrowseFolder]:
1355        """
1356        Browse personalized wave stations from rotor/stations/dashboard.
1357
1358        Names are resolved from the non-personalized station list so that
1359        stations show their actual genre/mood name (e.g. "Рок") rather than
1360        the generic "Мой микс" label that the dashboard API returns.
1361
1362        :param path: Full browse path (used to build sub-paths).
1363        :return: List of playable BrowseFolder items, one per station.
1364        """
1365        stations = await self._get_dashboard_stations_cached()
1366
1367        # Build a name map from the non-personalized list for proper localized names.
1368        locale = (self.mass.metadata.locale or "en_US").lower()
1369        language = "ru" if locale.startswith("ru") else "en"
1370        all_stations = await self.client.get_wave_stations(language)
1371        station_name_map: dict[str, str] = {sid: name for sid, _, name, _ in all_stations}
1372
1373        base = path.rstrip("/") + "/"
1374        folders: list[BrowseFolder] = []
1375        for station_id, fallback_name, image_url in stations:
1376            # Use full station_id (e.g. "genre:rock") in path to avoid collisions
1377            # when two stations share the same tag but differ by category.
1378            # The routing fallback (if ":" in tag) handles this correctly.
1379            name = station_name_map.get(station_id, fallback_name)
1380            station_image: MediaItemImage | None = None
1381            if image_url:
1382                station_image = MediaItemImage(
1383                    type=ImageType.THUMB,
1384                    path=image_url,
1385                    provider=self.instance_id,
1386                    remotely_accessible=True,
1387                )
1388            folders.append(
1389                BrowseFolder(
1390                    item_id=station_id,
1391                    provider=self.instance_id,
1392                    path=f"{base}{station_id}",
1393                    name=name,
1394                    is_playable=True,
1395                    image=station_image,
1396                )
1397            )
1398        return folders
1399
1400    async def _browse_wave_station(
1401        self, station_id: str, path: str = ""
1402    ) -> list[Track | BrowseFolder]:
1403        """
1404        Browse a rotor wave station and return tracks.
1405
1406        Fetches tracks from the rotor station, deduplicates within the current session,
1407        and sends radioStarted feedback on first call. Appends a "Load more" BrowseFolder
1408        at the end so MA can continue fetching the next batch automatically (radio mode).
1409
1410        :param station_id: Rotor station ID (e.g. 'genre:rock', 'mood:chill').
1411        :param path: Current browse path, used to construct the "Load more" next path.
1412        :return: List of Track objects with composite item_id (track_id@station_id),
1413                 followed by a "Load more" BrowseFolder if more tracks are available.
1414        """
1415        state = self._get_wave_state(station_id)
1416        async with state.lock:
1417            max_tracks = int(
1418                self.config.get_value(CONF_MY_WAVE_MAX_TRACKS) or 150  # type: ignore[arg-type]
1419            )
1420
1421            self.logger.debug(
1422                "Browse wave station: station_id=%s path=%s last_track_id=%s",
1423                station_id,
1424                path,
1425                state.last_track_id,
1426            )
1427            raw_tracks, batch_id = await self.client.get_rotor_station_tracks(
1428                station_id, queue=state.last_track_id
1429            )
1430            if batch_id:
1431                state.batch_id = batch_id
1432
1433            if not state.radio_started_sent and raw_tracks:
1434                sent = await self.client.send_rotor_station_feedback(
1435                    station_id,
1436                    "radioStarted",
1437                    batch_id=batch_id,
1438                )
1439                if sent:
1440                    state.radio_started_sent = True
1441
1442            tracks: list[Track] = []
1443            first_track_id: str | None = None
1444            for yt in raw_tracks:
1445                if len(state.seen_track_ids) >= max_tracks:
1446                    break
1447                track = self._parse_my_wave_track(yt, state.seen_track_ids)
1448                if track is None:
1449                    continue
1450                # Override station_id in composite item_id to reflect this specific station
1451                old_item_id = track.item_id
1452                track_id = old_item_id.split(RADIO_TRACK_ID_SEP, 1)[0]
1453                track.item_id = f"{track_id}{RADIO_TRACK_ID_SEP}{station_id}"
1454                # Keep provider mappings in sync with the new item_id
1455                for pm in getattr(track, "provider_mappings", []):
1456                    if (
1457                        getattr(pm, "item_id", None) == old_item_id
1458                        and getattr(pm, "provider_instance", None) == self.instance_id
1459                    ):
1460                        pm.item_id = track.item_id
1461                if first_track_id is None:
1462                    first_track_id = track_id
1463                tracks.append(track)
1464
1465            if first_track_id is not None:
1466                state.last_track_id = first_track_id
1467
1468            self.logger.debug(
1469                "Wave station %s returned %d tracks: %s",
1470                station_id,
1471                len(tracks),
1472                [t.item_id.split(RADIO_TRACK_ID_SEP, 1)[0] for t in tracks[:5]],
1473            )
1474            result: list[Track | BrowseFolder] = list(tracks)
1475
1476            # Append "Load more" sentinel so MA knows to call browse again for next batch.
1477            # This mirrors the My Mix mechanism and enables continuous radio playback.
1478            if tracks and len(state.seen_track_ids) < max_tracks and path:
1479                # Append /next to the current path (same pattern as _browse_my_wave).
1480                # This makes each "Load more" path unique (e.g. /next/next/next...)
1481                # so MA never serves a cached result for subsequent presses.
1482                result.append(
1483                    BrowseFolder(
1484                        item_id="next",
1485                        provider=self.instance_id,
1486                        path=f"{path.rstrip('/')}/next",
1487                        name="Load more",
1488                        translation_key="load_more",
1489                        is_playable=False,
1490                    )
1491                )
1492
1493            return result
1494
1495    @staticmethod
1496    def _extract_wave_item_cover(item: dict[str, Any]) -> tuple[str | None, str | None]:
1497        """
1498        Extract cover URI and background color from a wave/mix item.
1499
1500        :param item: Wave or mix item dict from the API.
1501        :return: (cover_uri, bg_color) tuple where bg_color is a hex string or None.
1502        """
1503        agent_uri = item.get("agent", {}).get("cover", {}).get("uri", "")
1504        cover_uri = agent_uri or item.get("compact_image_url")
1505        bg_color = item.get("colors", {}).get("average")
1506        return cover_uri, bg_color
1507
1508    @use_cache(3600)
1509    async def _get_mixes_waves_cached(self) -> list[dict[str, Any]] | None:
1510        """
1511        Get AI Wave Set data from /landing-blocks/mixes-waves, cached for 1 hour.
1512
1513        :return: List of mix category dicts from the API, or None on error.
1514        """
1515        return await self.client.get_mixes_waves()
1516
1517    @use_cache(3600)
1518    async def _get_waves_landing_cached(self) -> list[dict[str, Any]] | None:
1519        """
1520        Get Featured Mixes data from /landing-blocks/waves, cached for 1 hour.
1521
1522        :return: List of wave category dicts from the API, or None on error.
1523        """
1524        return await self.client.get_waves_landing()
1525
1526    async def _browse_waves_landing(
1527        self, path: str, path_parts: list[str]
1528    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1529        """
1530        Browse Featured Mixes (from /landing-blocks/waves).
1531
1532        :param path: Full browse path.
1533        :param path_parts: Split path parts after ://.
1534        :return: List of folders or tracks.
1535        """
1536        waves_data = await self._get_waves_landing_cached()
1537        return await self._browse_wave_categories(
1538            path, path_parts, waves_data or [], WAVES_LANDING_FOLDER_ID
1539        )
1540
1541    async def _browse_wave_categories(
1542        self,
1543        path: str,
1544        path_parts: list[str],
1545        categories_data: list[dict[str, Any]],
1546        id_prefix: str,
1547    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1548        """
1549        Browse wave-like category folders and their station items.
1550
1551        Shared logic for both 'my_waves_set' browse trees:
1552        - Level 1 (e.g. my_waves_set/): category folders
1553        - Level 2 (e.g. my_waves_set/ai-sets/): playable station folders with artwork
1554        - Level 3+ (e.g. my_waves_set/ai-sets/genre:rock[/next]): track listing
1555
1556        :param path: Full browse path.
1557        :param path_parts: Split path parts after ://.
1558        :param categories_data: List of category dicts from the API.
1559        :param id_prefix: Prefix for BrowseFolder item_id (e.g. 'my_waves_set').
1560        :return: List of folders or tracks.
1561        """
1562        base = path.rstrip("/") + "/"
1563
1564        if not categories_data:
1565            return []
1566
1567        # Level 1 → category folders
1568        if len(path_parts) == 1:
1569            folders: list[BrowseFolder] = []
1570            for wave_category in categories_data:
1571                cat_id = wave_category.get("id", "")
1572                cat_title = wave_category.get("title", "")
1573                items = wave_category.get("items", [])
1574                if not items or not cat_id:
1575                    continue
1576                display_name = cat_title.capitalize() if cat_title else cat_id.capitalize()
1577                folders.append(
1578                    BrowseFolder(
1579                        item_id=f"{id_prefix}_{cat_id}",
1580                        provider=self.instance_id,
1581                        path=f"{base}{cat_id}",
1582                        name=display_name,
1583                        is_playable=False,
1584                    )
1585                )
1586            return folders
1587
1588        category_id = path_parts[1] if len(path_parts) > 1 else None
1589        if not category_id:
1590            return []
1591
1592        # Level 3+ → stream tracks from rotor station
1593        if len(path_parts) > 2:
1594            station_id = path_parts[2]
1595            return await self._browse_wave_station(station_id, path=path)
1596
1597        # Level 2 → playable station folders with artwork
1598        for wave_category in categories_data:
1599            if wave_category.get("id") == category_id:
1600                items = wave_category.get("items", [])
1601                result: list[BrowseFolder] = []
1602                for item in items:
1603                    station_id = item.get("station_id", "")
1604                    title = item.get("title", "")
1605                    if not station_id or not title:
1606                        continue
1607                    cover_uri, bg_color = self._extract_wave_item_cover(item)
1608                    image: MediaItemImage | None = None
1609                    if cover_uri:
1610                        if cover_uri.startswith("http"):
1611                            img_url: str = cover_uri.replace("%%", IMAGE_SIZE_MEDIUM)
1612                        else:
1613                            raw = get_image_url(cover_uri)
1614                            img_url = "" if raw is None else raw
1615                        if img_url:
1616                            if bg_color:
1617                                # Append bg_color as URL fragment for cache-key uniqueness.
1618                                # MA will call resolve_image() to composite the transparent PNG.
1619                                if len(self._wave_bg_colors) > 200:
1620                                    self._wave_bg_colors.clear()
1621                                img_url = f"{img_url}#{bg_color.lstrip('#')}"
1622                                self._wave_bg_colors[img_url] = bg_color
1623                            image = MediaItemImage(
1624                                type=ImageType.THUMB,
1625                                path=img_url,
1626                                provider=self.instance_id,
1627                                remotely_accessible=bg_color is None,
1628                            )
1629                    result.append(
1630                        BrowseFolder(
1631                            item_id=station_id,
1632                            provider=self.instance_id,
1633                            path=f"{base}{station_id}",
1634                            name=title,
1635                            is_playable=True,
1636                            image=image,
1637                        )
1638                    )
1639                return result
1640
1641        return []
1642
1643    async def _browse_vibe_sets(
1644        self, path: str, path_parts: list[str]
1645    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1646        """
1647        Browse AI Mix Sets (from /landing-blocks/mixes-waves).
1648
1649        :param path: Full browse path.
1650        :param path_parts: Split path parts after ://.
1651        :return: List of folders or tracks.
1652        """
1653        mixes_data = await self._get_mixes_waves_cached()
1654        return await self._browse_wave_categories(
1655            path, path_parts, mixes_data or [], MY_WAVES_SET_FOLDER_ID
1656        )
1657
1658    @use_cache(600)
1659    async def _get_tag_playlists_as_browse(
1660        self, tag_id: str
1661    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
1662        """
1663        Get playlists for a tag and return as browse items.
1664
1665        :param tag_id: Tag identifier (e.g. 'chill', '80s').
1666        :return: List of Playlist objects.
1667        """
1668        self.logger.debug("Fetching playlists for tag: %s", tag_id)
1669        playlists = await self.client.get_tag_playlists(tag_id)
1670        self.logger.debug("Got %d playlists for tag %s", len(playlists), tag_id)
1671        result: list[Playlist] = []
1672        for playlist in playlists:
1673            try:
1674                result.append(parse_playlist(self, playlist))
1675            except InvalidDataError as err:
1676                self.logger.debug("Error parsing tag playlist: %s", err)
1677        self.logger.debug("Parsed %d playlists for tag %s", len(result), tag_id)
1678        return result
1679
1680    # Search
1681
1682    @use_cache(3600 * 24 * 14)
1683    async def search(
1684        self, search_query: str, media_types: list[MediaType], limit: int = 5
1685    ) -> SearchResults:
1686        """
1687        Perform search on KION Music.
1688
1689        :param search_query: The search query.
1690        :param media_types: List of media types to search for.
1691        :param limit: Maximum number of results per type.
1692        :return: SearchResults with found items.
1693        """
1694        result = SearchResults()
1695
1696        # Determine search type based on requested media types
1697        # Map MediaType to Kion API search type
1698        type_mapping = {
1699            MediaType.TRACK: "track",
1700            MediaType.ALBUM: "album",
1701            MediaType.ARTIST: "artist",
1702            MediaType.PLAYLIST: "playlist",
1703        }
1704        requested_types = [type_mapping[mt] for mt in media_types if mt in type_mapping]
1705
1706        # Use specific type if only one requested, otherwise search all
1707        search_type = requested_types[0] if len(requested_types) == 1 else "all"
1708
1709        search_result = await self.client.search(search_query, search_type=search_type, limit=limit)
1710        if not search_result:
1711            return result
1712
1713        # Parse tracks
1714        if MediaType.TRACK in media_types and search_result.tracks:
1715            for track in search_result.tracks.results[:limit]:
1716                try:
1717                    result.tracks = [*result.tracks, parse_track(self, track)]
1718                except InvalidDataError as err:
1719                    self.logger.debug("Error parsing track: %s", err)
1720
1721        # Parse albums
1722        if MediaType.ALBUM in media_types and search_result.albums:
1723            for album in search_result.albums.results[:limit]:
1724                try:
1725                    result.albums = [*result.albums, parse_album(self, album)]
1726                except InvalidDataError as err:
1727                    self.logger.debug("Error parsing album: %s", err)
1728
1729        # Parse artists
1730        if MediaType.ARTIST in media_types and search_result.artists:
1731            for artist in search_result.artists.results[:limit]:
1732                try:
1733                    result.artists = [*result.artists, parse_artist(self, artist)]
1734                except InvalidDataError as err:
1735                    self.logger.debug("Error parsing artist: %s", err)
1736
1737        # Parse playlists
1738        if MediaType.PLAYLIST in media_types and search_result.playlists:
1739            for playlist in search_result.playlists.results[:limit]:
1740                try:
1741                    result.playlists = [*result.playlists, parse_playlist(self, playlist)]
1742                except InvalidDataError as err:
1743                    self.logger.debug("Error parsing playlist: %s", err)
1744
1745        return result
1746
1747    # Get single items
1748
1749    @use_cache(3600 * 24 * 30)
1750    async def get_artist(self, prov_artist_id: str) -> Artist:
1751        """
1752        Get artist details by ID, enriched with description and listener stats.
1753
1754        :param prov_artist_id: The provider artist ID.
1755        :return: Artist object.
1756        :raises MediaNotFoundError: If artist not found.
1757        """
1758        artist, about = await asyncio.gather(
1759            self.client.get_artist(prov_artist_id),
1760            self.client.get_artist_about(prov_artist_id),
1761        )
1762        if not artist:
1763            raise MediaNotFoundError(f"Artist {prov_artist_id} not found")
1764        return parse_artist(self, artist, about=about)
1765
1766    @use_cache(3600 * 24 * 30)
1767    async def get_album(self, prov_album_id: str) -> Album:
1768        """
1769        Get album details by ID.
1770
1771        :param prov_album_id: The provider album ID.
1772        :return: Album object.
1773        :raises MediaNotFoundError: If album not found.
1774        """
1775        album = await self.client.get_album(prov_album_id)
1776        if not album:
1777            raise MediaNotFoundError(f"Album {prov_album_id} not found")
1778        return parse_album(self, album)
1779
1780    async def get_track(self, prov_track_id: str) -> Track:
1781        """
1782        Get track details by ID.
1783
1784        Supports composite item_id (track_id@station_id) for My Mix tracks;
1785        only the track_id part is used for the API. Normalizes the ID before
1786        caching to avoid duplicate cache entries.
1787
1788        :param prov_track_id: The provider track ID (or track_id@station_id).
1789        :return: Track object.
1790        :raises MediaNotFoundError: If track not found.
1791        """
1792        track_id, _ = _parse_radio_item_id(prov_track_id)
1793        return await self._get_track_cached(track_id)
1794
1795    @use_cache(3600 * 24 * 30)
1796    async def _get_track_cached(self, track_id: str) -> Track:
1797        """
1798        Get track details by normalized ID (cached).
1799
1800        :param track_id: Normalized track ID (without station suffix).
1801        :return: Track object.
1802        :raises MediaNotFoundError: If track not found.
1803        """
1804        raw_track = await self.client.get_track(track_id)
1805        if not raw_track:
1806            raise MediaNotFoundError(f"Track {track_id} not found")
1807
1808        # Use the already-fetched track object to avoid a duplicate API call
1809        lyrics, lyrics_synced = await self.client.get_track_lyrics_from_track(raw_track)
1810
1811        return parse_track(self, raw_track, lyrics=lyrics, lyrics_synced=lyrics_synced)
1812
1813    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
1814        """
1815        Get playlist details by ID.
1816
1817        Supports virtual playlists MY_WAVE_PLAYLIST_ID (My Mix) and
1818        LIKED_TRACKS_PLAYLIST_ID (Liked Tracks). Real playlists use format "owner_id:kind".
1819
1820        :param prov_playlist_id: The provider playlist ID (format: "owner_id:kind",
1821            my_wave, or liked_tracks).
1822        :return: Playlist object.
1823        :raises MediaNotFoundError: If playlist not found.
1824        """
1825        # Virtual playlists - constructed locally (no API call). translation_key localizes
1826        # the name for the connection locale at serialization; the English name is kept (not
1827        # dropped like browse/recommendation folders) because a playable item's name is also
1828        # read outside outbound API serialization (e.g. queue / now-playing metadata).
1829        if prov_playlist_id == MY_WAVE_PLAYLIST_ID:
1830            return Playlist(
1831                item_id=MY_WAVE_PLAYLIST_ID,
1832                provider=self.instance_id,
1833                name="My Mix",
1834                translation_key=MY_WAVE_PLAYLIST_ID,
1835                owner=get_canonical_provider_name(self),
1836                provider_mappings={
1837                    ProviderMapping(
1838                        item_id=MY_WAVE_PLAYLIST_ID,
1839                        provider_domain=self.domain,
1840                        provider_instance=self.instance_id,
1841                        is_unique=True,
1842                    )
1843                },
1844                is_editable=False,
1845            )
1846
1847        if prov_playlist_id == LIKED_TRACKS_PLAYLIST_ID:
1848            return Playlist(
1849                item_id=LIKED_TRACKS_PLAYLIST_ID,
1850                provider=self.instance_id,
1851                name="My Favorites",
1852                translation_key=LIKED_TRACKS_PLAYLIST_ID,
1853                owner=get_canonical_provider_name(self),
1854                provider_mappings={
1855                    ProviderMapping(
1856                        item_id=LIKED_TRACKS_PLAYLIST_ID,
1857                        provider_domain=self.domain,
1858                        provider_instance=self.instance_id,
1859                        is_unique=True,
1860                    )
1861                },
1862                is_editable=False,
1863            )
1864
1865        # Real playlists - use cached method
1866        return await self._get_real_playlist(prov_playlist_id)
1867
1868    @use_cache(3600 * 24 * 30)
1869    async def _get_real_playlist(self, prov_playlist_id: str) -> Playlist:
1870        """
1871        Get real playlist details by ID (cached).
1872
1873        :param prov_playlist_id: The provider playlist ID (format: "owner_id:kind").
1874        :return: Playlist object.
1875        :raises MediaNotFoundError: If playlist not found.
1876        """
1877        # Parse the playlist ID (format: owner_id:kind)
1878        if PLAYLIST_ID_SPLITTER in prov_playlist_id:
1879            owner_id, kind = prov_playlist_id.split(PLAYLIST_ID_SPLITTER, 1)
1880        else:
1881            owner_id = str(self.client.user_id)
1882            kind = prov_playlist_id
1883
1884        playlist = await self.client.get_playlist(owner_id, kind)
1885        if not playlist:
1886            raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found")
1887        return parse_playlist(self, playlist)
1888
1889    @use_cache(3600 * 3, allow_expired_cache=True)
1890    async def _get_my_wave_playlist_tracks(self, page: int) -> list[Track]:
1891        """
1892        Get My Mix tracks for virtual playlist (uses cursor for page > 0).
1893
1894        Fetches MY_WAVE_BATCH_SIZE Rotor API batches per page call to reduce
1895        the number of round-trips when the player controller paginates through pages.
1896
1897        :param page: Page number (0 = first batch, 1+ = next batches via queue cursor).
1898        :return: List of Track objects for this page.
1899        """
1900        async with self._my_wave_lock:
1901            max_tracks_config = int(
1902                self.config.get_value(CONF_MY_WAVE_MAX_TRACKS) or 150  # type: ignore[arg-type]
1903            )
1904
1905            # Reset seen tracks on first page
1906            if page == 0:
1907                self._my_wave_seen_track_ids = set()
1908
1909            queue: str | int | None = None
1910            if page > 0:
1911                queue = self._my_wave_playlist_next_cursor
1912                if not queue:
1913                    return []
1914
1915            # Check if we've already reached the limit
1916            if len(self._my_wave_seen_track_ids) >= max_tracks_config:
1917                return []
1918
1919            tracks: list[Track] = []
1920            next_cursor: str | None = None
1921
1922            # Fetch MY_WAVE_BATCH_SIZE Rotor API batches per page to reduce API round-trips
1923            for _ in range(MY_WAVE_BATCH_SIZE):
1924                if len(self._my_wave_seen_track_ids) >= max_tracks_config:
1925                    break
1926
1927                raw_tracks, batch_id = await self.client.get_my_wave_tracks(queue=queue)
1928                if batch_id:
1929                    self._my_wave_batch_id = batch_id
1930                if not self._my_wave_radio_started_sent and raw_tracks:
1931                    sent = await self.client.send_rotor_station_feedback(
1932                        ROTOR_STATION_MY_MIX,
1933                        "radioStarted",
1934                        batch_id=batch_id,
1935                    )
1936                    if sent:
1937                        self._my_wave_radio_started_sent = True
1938
1939                if not raw_tracks:
1940                    break
1941
1942                first_track_id_this_batch = None
1943                for yt in raw_tracks:
1944                    if len(self._my_wave_seen_track_ids) >= max_tracks_config:
1945                        break
1946
1947                    track = self._parse_my_wave_track(yt, self._my_wave_seen_track_ids)
1948                    if track is None:
1949                        continue
1950
1951                    tracks.append(track)
1952                    track_id = track.item_id.split(RADIO_TRACK_ID_SEP, 1)[0]
1953                    if first_track_id_this_batch is None:
1954                        first_track_id_this_batch = track_id
1955
1956                if first_track_id_this_batch is not None:
1957                    next_cursor = first_track_id_this_batch
1958                    queue = first_track_id_this_batch
1959                else:
1960                    # All tracks in this batch were duplicates or failed to parse
1961                    break
1962
1963            # Store cursor for next page call (None clears pagination so next call returns [])
1964            self._my_wave_playlist_next_cursor = next_cursor
1965            return tracks
1966
1967    @use_cache(3600 * 3, allow_expired_cache=True)
1968    async def _get_liked_tracks_playlist_tracks(self, page: int) -> list[Track]:
1969        """
1970        Get liked tracks for virtual playlist (sorted in reverse chronological order).
1971
1972        :param page: Page number (0 = all tracks limited by config, >0 = empty for pagination).
1973        :return: List of Track objects.
1974        """
1975        # Liked tracks API returns all tracks at once, so only return tracks on page 0
1976        if page > 0:
1977            return []
1978
1979        max_tracks_config = int(
1980            self.config.get_value(CONF_LIKED_TRACKS_MAX_TRACKS) or 500  # type: ignore[arg-type]
1981        )
1982
1983        # Fetch liked tracks (already sorted in reverse chronological order by api_client)
1984        track_shorts = await self.client.get_liked_tracks()
1985        if not track_shorts:
1986            self.logger.debug("No liked tracks found")
1987            return []
1988
1989        # Apply max tracks limit
1990        track_shorts = track_shorts[:max_tracks_config]
1991
1992        # Fetch full track details in batches
1993        track_ids = [str(ts.track_id) for ts in track_shorts if ts.track_id]
1994
1995        batch_size = TRACK_BATCH_SIZE
1996        full_tracks = []
1997        for i in range(0, len(track_ids), batch_size):
1998            batch_ids = track_ids[i : i + batch_size]
1999            batch_result = await self.client.get_tracks(batch_ids)
2000            full_tracks.extend(batch_result)
2001
2002        # Create track ID to full track mapping by track ID directly
2003        track_map = {}
2004        for t in full_tracks:
2005            if hasattr(t, "id") and t.id:
2006                track_map[str(t.id)] = t
2007
2008        # Parse tracks in the original order (reverse chronological)
2009        tracks = []
2010        for track_id in track_ids:
2011            # track_id may be compound "trackId:albumId", extract base ID for lookup
2012            base_id = track_id.split(":")[0] if ":" in track_id else track_id
2013            found = track_map.get(track_id) or track_map.get(base_id)
2014            if found:
2015                try:
2016                    tracks.append(parse_track(self, found))
2017                except InvalidDataError as err:
2018                    self.logger.debug("Error parsing liked track %s: %s", track_id, err)
2019
2020        self.logger.debug("Liked tracks: fetched %s, parsed %s", len(track_shorts), len(tracks))
2021        return tracks
2022
2023    # Get related items
2024
2025    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
2026    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
2027        """
2028        Get album tracks.
2029
2030        :param prov_album_id: The provider album ID.
2031        :return: List of Track objects.
2032        """
2033        album = await self.client.get_album_with_tracks(prov_album_id)
2034        if not album or not album.volumes:
2035            return []
2036
2037        tracks = []
2038        for volume_index, volume in enumerate(album.volumes):
2039            for track_index, track in enumerate(volume):
2040                try:
2041                    parsed_track = parse_track(self, track)
2042                    parsed_track.disc_number = volume_index + 1
2043                    parsed_track.track_number = track_index + 1
2044                    tracks.append(parsed_track)
2045                except InvalidDataError as err:
2046                    self.logger.debug("Error parsing album track: %s", err)
2047        return tracks
2048
2049    @use_cache(3600 * 3, allow_expired_cache=True)
2050    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
2051        """
2052        Get similar tracks using Kion Rotor station for this track.
2053
2054        Uses rotor station track:{id} so MA radio mode gets Kion recommendations.
2055
2056        :param prov_track_id: Provider track ID (plain or track_id@station_id).
2057        :param limit: Maximum number of tracks to return.
2058        :return: List of similar Track objects.
2059        """
2060        track_id, _ = _parse_radio_item_id(prov_track_id)
2061        station_id = f"track:{track_id}"
2062        raw_tracks, _ = await self.client.get_rotor_station_tracks(station_id, queue=None)
2063        tracks = []
2064        for yt in raw_tracks[:limit]:
2065            try:
2066                tracks.append(parse_track(self, yt))
2067            except InvalidDataError as err:
2068                self.logger.debug("Error parsing similar track: %s", err)
2069        return tracks
2070
2071    @use_cache(3600 * 3)
2072    async def get_similar_artists(self, prov_artist_id: str, limit: int = 25) -> list[Artist]:
2073        """
2074        Get artists similar to the given one via Kion artists/similar endpoint.
2075
2076        :param prov_artist_id: Provider artist ID.
2077        :param limit: Maximum number of artists to return.
2078        :return: List of similar Artist objects.
2079        """
2080        raw_artists = await self.client.get_similar_artists(prov_artist_id, limit=limit)
2081        artists: list[Artist] = []
2082        for ya in raw_artists:
2083            try:
2084                artists.append(parse_artist(self, ya))
2085            except InvalidDataError as err:
2086                self.logger.debug("Error parsing similar artist: %s", err)
2087        return artists
2088
2089    @use_cache(600)
2090    async def _get_my_wave_recommendations(self) -> RecommendationFolder | None:
2091        """
2092        Get My Mix recommendation folder with personalized tracks.
2093
2094        :return: RecommendationFolder with My Mix tracks, or None if empty.
2095        """
2096        max_tracks_config = int(
2097            self.config.get_value(CONF_MY_WAVE_MAX_TRACKS) or 150  # type: ignore[arg-type]
2098        )
2099        batch_size_config = MY_WAVE_BATCH_SIZE
2100
2101        seen_track_ids: set[str] = set()
2102        items: list[Track] = []
2103        queue: str | int | None = None
2104
2105        for _ in range(batch_size_config):
2106            if len(seen_track_ids) >= max_tracks_config:
2107                break
2108
2109            raw_tracks, _ = await self.client.get_my_wave_tracks(queue=queue)
2110            if not raw_tracks:
2111                break
2112
2113            first_track_id_this_batch = None
2114            for yt in raw_tracks:
2115                if len(seen_track_ids) >= max_tracks_config:
2116                    break
2117
2118                track = self._parse_my_wave_track(yt, seen_ids=seen_track_ids)
2119                if track is None:
2120                    continue
2121
2122                items.append(track)
2123                track_id = track.item_id.split(RADIO_TRACK_ID_SEP, 1)[0]
2124                if first_track_id_this_batch is None:
2125                    first_track_id_this_batch = track_id
2126
2127            queue = first_track_id_this_batch
2128            if not queue:
2129                break
2130
2131        if not items:
2132            return None
2133
2134        initial_tracks_limit = DISCOVERY_INITIAL_TRACKS
2135        if len(items) > initial_tracks_limit:
2136            items = items[:initial_tracks_limit]
2137
2138        # Recommendation folders keep their English name (not dropped like browse folders):
2139        # MusicProvider.browse() re-wraps them into plain BrowseFolders for the
2140        # "<provider>://recommendations" listing, where a bare translation_key would resolve
2141        # under the wrong media group. translation_key still localizes the recommendations view.
2142        return RecommendationFolder(
2143            item_id=MY_WAVE_PLAYLIST_ID,
2144            provider=self.instance_id,
2145            name="My Mix",
2146            translation_key=MY_WAVE_PLAYLIST_ID,
2147            items=UniqueList(items),
2148            icon="mdi-waveform",
2149        )
2150
2151    @use_cache(1800)
2152    async def _get_feed_recommendations(self) -> RecommendationFolder | None:
2153        """
2154        Get personalized feed playlists (Playlist of the Day, DejaVu, etc.).
2155
2156        :return: RecommendationFolder with generated playlists, or None if unavailable.
2157        """
2158        feed = await self.client.get_feed()
2159        if not feed or not feed.generated_playlists:
2160            return None
2161        items: list[Playlist] = []
2162        for gen_playlist in feed.generated_playlists:
2163            if gen_playlist.data and gen_playlist.ready:
2164                try:
2165                    items.append(parse_playlist(self, gen_playlist.data))
2166                except InvalidDataError as err:
2167                    self.logger.debug("Error parsing feed playlist: %s", err)
2168        if not items:
2169            return None
2170        return RecommendationFolder(
2171            item_id="feed",
2172            provider=self.instance_id,
2173            name="Made for you",
2174            translation_key="made_for_you",
2175            items=UniqueList(items),
2176            icon="mdi-account-music",
2177        )
2178
2179    @use_cache(3600)
2180    async def _get_chart_recommendations(self) -> RecommendationFolder | None:
2181        """
2182        Get chart tracks (hot tracks of the month).
2183
2184        :return: RecommendationFolder with chart tracks, or None if unavailable.
2185        """
2186        chart_info = await self.client.get_chart()
2187        if not chart_info or not chart_info.chart:
2188            return None
2189        playlist = chart_info.chart
2190        if not playlist.tracks:
2191            return None
2192        # TrackShort objects in chart context have .track (full Track) and .chart (position)
2193        tracks: list[Track] = []
2194        for track_short in playlist.tracks[:20]:
2195            track_obj = getattr(track_short, "track", None)
2196            if not track_obj:
2197                continue
2198            try:
2199                tracks.append(parse_track(self, track_obj))
2200            except InvalidDataError as err:
2201                self.logger.debug("Error parsing chart track: %s", err)
2202        if not tracks:
2203            return None
2204        return RecommendationFolder(
2205            item_id="chart",
2206            provider=self.instance_id,
2207            name="Chart",
2208            translation_key="chart",
2209            items=UniqueList(tracks),
2210            icon="mdi-chart-line",
2211        )
2212
2213    @use_cache(3600)
2214    async def _get_new_releases_recommendations(self) -> RecommendationFolder | None:
2215        """
2216        Get new album releases.
2217
2218        :return: RecommendationFolder with new albums, or None if unavailable.
2219        """
2220        releases = await self.client.get_new_releases()
2221        if not releases or not releases.new_releases:
2222            return None
2223        # new_releases is a list of album IDs (int) — need to batch-fetch full details
2224        album_ids = [str(aid) for aid in releases.new_releases[:20]]
2225        if not album_ids:
2226            return None
2227        full_albums = await self.client.get_albums(album_ids)
2228        if not full_albums:
2229            return None
2230        albums: list[Album] = []
2231        for album in full_albums:
2232            try:
2233                albums.append(parse_album(self, album))
2234            except InvalidDataError as err:
2235                self.logger.debug("Error parsing new release album: %s", err)
2236        if not albums:
2237            return None
2238        return RecommendationFolder(
2239            item_id="new_releases",
2240            provider=self.instance_id,
2241            name="New Releases",
2242            translation_key="new_releases",
2243            items=UniqueList(albums),
2244            icon="mdi-new-box",
2245        )
2246
2247    @use_cache(3600)
2248    async def _get_new_playlists_recommendations(self) -> RecommendationFolder | None:
2249        """
2250        Get new editorial playlists.
2251
2252        :return: RecommendationFolder with new playlists, or None if unavailable.
2253        """
2254        result = await self.client.get_new_playlists()
2255        if not result or not result.new_playlists:
2256            return None
2257        # new_playlists is a list of PlaylistId objects (uid, kind) — fetch full details
2258        playlist_ids = [
2259            f"{pid.uid}:{pid.kind}"
2260            for pid in result.new_playlists[:20]
2261            if hasattr(pid, "uid") and hasattr(pid, "kind")
2262        ]
2263        if not playlist_ids:
2264            return None
2265        full_playlists = await self.client.get_playlists(playlist_ids)
2266        if not full_playlists:
2267            return None
2268        playlists: list[Playlist] = []
2269        for playlist in full_playlists:
2270            try:
2271                playlists.append(parse_playlist(self, playlist))
2272            except InvalidDataError as err:
2273                self.logger.debug("Error parsing new playlist: %s", err)
2274        if not playlists:
2275            return None
2276        return RecommendationFolder(
2277            item_id="new_playlists",
2278            provider=self.instance_id,
2279            name="New Playlists",
2280            translation_key="new_playlists",
2281            items=UniqueList(playlists),
2282            icon="mdi-playlist-star",
2283        )
2284
2285    @use_cache(3600)
2286    async def _get_top_picks_recommendations(self) -> RecommendationFolder | None:
2287        """
2288        Get Top Picks recommendation folder (tag: top).
2289
2290        :return: RecommendationFolder with top playlists, or None if unavailable.
2291        """
2292        playlists = await self.client.get_tag_playlists("top")
2293        if not playlists:
2294            return None
2295        items: list[Playlist] = []
2296        for playlist in playlists[:10]:
2297            try:
2298                items.append(parse_playlist(self, playlist))
2299            except InvalidDataError as err:
2300                self.logger.debug("Error parsing top picks playlist: %s", err)
2301        if not items:
2302            return None
2303        return RecommendationFolder(
2304            item_id="top_picks",
2305            provider=self.instance_id,
2306            name="Top Picks",
2307            translation_key="top_picks",
2308            items=UniqueList(items),
2309            icon="mdi-star",
2310        )
2311
2312    @use_cache(1800)
2313    async def _get_mood_mix_recommendations(self, mood_tag: str) -> RecommendationFolder | None:
2314        """
2315        Get Mood Mix recommendation folder for a specific tag.
2316
2317        :param mood_tag: Preselected mood tag slug.
2318        :return: RecommendationFolder with mood playlists, or None if unavailable.
2319        """
2320        playlists = await self.client.get_tag_playlists(mood_tag)
2321        if not playlists:
2322            self.logger.debug("No playlists for mood tag %s, skipping recommendation", mood_tag)
2323            return None
2324        items: list[Playlist] = []
2325        for playlist in playlists[:8]:
2326            try:
2327                items.append(parse_playlist(self, playlist))
2328            except InvalidDataError as err:
2329                self.logger.debug("Error parsing mood playlist: %s", err)
2330        if not items:
2331            return None
2332        tag_name = self._media_source_name("folder", _media_label_key(mood_tag)) or mood_tag.title()
2333        return RecommendationFolder(
2334            item_id="mood_mix",
2335            provider=self.instance_id,
2336            name=f"Mood Mix: {tag_name}",
2337            translation_key="mood_mix",
2338            translation_params=[tag_name],
2339            items=UniqueList(items),
2340            icon="mdi-emoticon-outline",
2341        )
2342
2343    @use_cache(1800)
2344    async def _get_activity_mix_recommendations(
2345        self, activity_tag: str
2346    ) -> RecommendationFolder | None:
2347        """
2348        Get Activity Mix recommendation folder for a specific tag.
2349
2350        :param activity_tag: Preselected activity tag slug.
2351        :return: RecommendationFolder with activity playlists, or None if unavailable.
2352        """
2353        playlists = await self.client.get_tag_playlists(activity_tag)
2354        if not playlists:
2355            self.logger.debug(
2356                "No playlists for activity tag %s, skipping recommendation", activity_tag
2357            )
2358            return None
2359        items: list[Playlist] = []
2360        for playlist in playlists[:8]:
2361            try:
2362                items.append(parse_playlist(self, playlist))
2363            except InvalidDataError as err:
2364                self.logger.debug("Error parsing activity playlist: %s", err)
2365        if not items:
2366            return None
2367        tag_name = (
2368            self._media_source_name("folder", _media_label_key(activity_tag))
2369            or activity_tag.title()
2370        )
2371        return RecommendationFolder(
2372            item_id="activity_mix",
2373            provider=self.instance_id,
2374            name=f"Activity Mix: {tag_name}",
2375            translation_key="activity_mix",
2376            translation_params=[tag_name],
2377            items=UniqueList(items),
2378            icon="mdi-run",
2379        )
2380
2381    @use_cache(3600 * 6)
2382    async def _get_seasonal_mix_recommendations(self) -> RecommendationFolder | None:
2383        """
2384        Get Seasonal Mix recommendation folder (based on current month).
2385
2386        :return: RecommendationFolder with seasonal playlists, or None if unavailable.
2387        """
2388        # Determine current season tag
2389        current_month = utc().month
2390        seasonal_tag = TAG_SEASONAL_MAP.get(current_month, "autumn")
2391
2392        # Validate the seasonal tag; fall back to autumn if not available
2393        if not await self._validate_tag(seasonal_tag):
2394            seasonal_tag = "autumn"
2395
2396        playlists = await self.client.get_tag_playlists(seasonal_tag)
2397        if not playlists:
2398            return None
2399        items: list[Playlist] = []
2400        for playlist in playlists[:8]:
2401            try:
2402                items.append(parse_playlist(self, playlist))
2403            except InvalidDataError as err:
2404                self.logger.debug("Error parsing seasonal playlist: %s", err)
2405        if not items:
2406            return None
2407        tag_name = (
2408            self._media_source_name("folder", _media_label_key(seasonal_tag))
2409            or seasonal_tag.title()
2410        )
2411        return RecommendationFolder(
2412            item_id="seasonal_mix",
2413            provider=self.instance_id,
2414            name=f"Seasonal: {tag_name}",
2415            translation_key="seasonal_mix",
2416            translation_params=[tag_name],
2417            items=UniqueList(items),
2418            icon="mdi-weather-sunny",
2419        )
2420
2421    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
2422        """
2423        Get playlist tracks.
2424
2425        :param prov_playlist_id: The provider playlist ID (format: "owner_id:kind",
2426            my_wave, or liked_tracks).
2427        :param page: Page number for pagination.
2428        :return: List of Track objects.
2429        """
2430        self.logger.debug(
2431            "get_playlist_tracks called: prov_playlist_id=%s, page=%s", prov_playlist_id, page
2432        )
2433
2434        if prov_playlist_id == MY_WAVE_PLAYLIST_ID:
2435            self.logger.debug("Fetching My Mix tracks")
2436            return await self._get_my_wave_playlist_tracks(page)
2437
2438        if prov_playlist_id == LIKED_TRACKS_PLAYLIST_ID:
2439            self.logger.debug("Fetching Liked Tracks for virtual playlist")
2440            result = await self._get_liked_tracks_playlist_tracks(page)
2441            self.logger.debug("Liked Tracks playlist returned %s tracks", len(result))
2442            return result
2443
2444        return await self._get_regular_playlist_tracks(prov_playlist_id, page)
2445
2446    @use_cache(3600 * 3, allow_expired_cache=True)
2447    async def _get_regular_playlist_tracks(self, prov_playlist_id: str, page: int) -> list[Track]:
2448        """
2449        Get the tracks of a regular (non-virtual) playlist.
2450
2451        :param prov_playlist_id: The provider playlist ID (format: "owner_id:kind").
2452        :param page: Page number for pagination.
2453        :return: List of Track objects.
2454        """
2455        # KION Music API returns all playlist tracks in one call (no server-side pagination).
2456        # Return empty list for page > 0 so the controller pagination loop terminates.
2457        if page > 0:
2458            return []
2459
2460        # Parse the playlist ID (format: owner_id:kind)
2461        if PLAYLIST_ID_SPLITTER in prov_playlist_id:
2462            owner_id, kind = prov_playlist_id.split(PLAYLIST_ID_SPLITTER, 1)
2463        else:
2464            owner_id = str(self.client.user_id)
2465            kind = prov_playlist_id
2466
2467        playlist = await self.client.get_playlist(owner_id, kind)
2468        if not playlist:
2469            return []
2470
2471        # API sometimes returns playlist without tracks; fetch them explicitly if needed
2472        tracks_list = playlist.tracks or []
2473        track_count = getattr(playlist, "track_count", None) or 0
2474        if not tracks_list and track_count > 0:
2475            self.logger.debug(
2476                "Playlist %s/%s: track_count=%s but no tracks in response, "
2477                "calling fetch_tracks_async",
2478                owner_id,
2479                kind,
2480                track_count,
2481            )
2482            try:
2483                tracks_list = await playlist.fetch_tracks_async()
2484            except Exception as err:
2485                self.logger.warning("fetch_tracks_async failed for %s/%s: %s", owner_id, kind, err)
2486            if not tracks_list:
2487                raise ResourceTemporarilyUnavailable(
2488                    "Playlist tracks not available; try again later"
2489                )
2490
2491        if not tracks_list:
2492            return []
2493
2494        # Kion returns TrackShort objects, we need to fetch full track info
2495        track_ids = [
2496            str(track.track_id) if hasattr(track, "track_id") else str(track.id)
2497            for track in tracks_list
2498            if track
2499        ]
2500        if not track_ids:
2501            return []
2502
2503        # Fetch full track details in batches to avoid timeouts
2504        batch_size = TRACK_BATCH_SIZE
2505        full_tracks = []
2506        for i in range(0, len(track_ids), batch_size):
2507            batch = track_ids[i : i + batch_size]
2508            batch_result = await self.client.get_tracks(batch)
2509            if not batch_result:
2510                self.logger.warning(
2511                    "Received empty result for playlist %s tracks batch %s-%s",
2512                    prov_playlist_id,
2513                    i,
2514                    i + len(batch) - 1,
2515                )
2516                raise ResourceTemporarilyUnavailable(
2517                    "Playlist tracks not fully available; try again later"
2518                )
2519            full_tracks.extend(batch_result)
2520
2521        if track_ids and not full_tracks:
2522            raise ResourceTemporarilyUnavailable("Failed to load track details; try again later")
2523
2524        tracks = []
2525        for track in full_tracks:
2526            try:
2527                tracks.append(parse_track(self, track))
2528            except InvalidDataError as err:
2529                self.logger.debug("Error parsing playlist track: %s", err)
2530        return tracks
2531
2532    @use_cache(3600 * 24 * 7, allow_expired_cache=True)
2533    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
2534        """
2535        Get artist's albums.
2536
2537        :param prov_artist_id: The provider artist ID.
2538        :return: List of Album objects.
2539        """
2540        albums = await self.client.get_artist_albums(prov_artist_id)
2541        result = []
2542        for album in albums:
2543            try:
2544                result.append(parse_album(self, album))
2545            except InvalidDataError as err:
2546                self.logger.debug("Error parsing artist album: %s", err)
2547        return result
2548
2549    @use_cache(3600 * 24 * 7, allow_expired_cache=True)
2550    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
2551        """
2552        Get artist's top tracks.
2553
2554        :param prov_artist_id: The provider artist ID.
2555        :return: List of Track objects.
2556        """
2557        tracks = await self.client.get_artist_tracks(prov_artist_id)
2558        result = []
2559        for track in tracks:
2560            try:
2561                result.append(parse_track(self, track))
2562            except InvalidDataError as err:
2563                self.logger.debug("Error parsing artist track: %s", err)
2564        return result
2565
2566    # Library methods
2567
2568    async def get_library_artists(self) -> AsyncGenerator[Artist]:
2569        """Retrieve library artists from KION Music."""
2570        artists = await self.client.get_liked_artists()
2571        for artist in artists:
2572            try:
2573                yield parse_artist(self, artist)
2574            except InvalidDataError as err:
2575                # only raised for a missing artist id, so the item is unidentifiable
2576                self.report_skipped_sync_item(MediaType.ARTIST, None, err)
2577
2578    async def get_library_albums(self) -> AsyncGenerator[Album]:
2579        """Retrieve library albums from KION Music."""
2580        batch_size = TRACK_BATCH_SIZE
2581        albums = await self.client.get_liked_albums(batch_size=batch_size)
2582        for album in albums:
2583            try:
2584                yield parse_album(self, album)
2585            except InvalidDataError as err:
2586                # album.id may still be usable even if one of its artists is not
2587                item_id = str(album.id) if album.id is not None else None
2588                self.report_skipped_sync_item(MediaType.ALBUM, item_id, err)
2589
2590    async def get_library_tracks(self) -> AsyncGenerator[Track]:
2591        """Retrieve library tracks from KION Music."""
2592        track_shorts = await self.client.get_liked_tracks()
2593        if not track_shorts:
2594            return
2595
2596        # Fetch full track details in batches
2597        track_ids = [str(ts.track_id) for ts in track_shorts if ts.track_id]
2598        batch_size = TRACK_BATCH_SIZE
2599        for i in range(0, len(track_ids), batch_size):
2600            batch_ids = track_ids[i : i + batch_size]
2601            full_tracks = await self.client.get_tracks(batch_ids)
2602            for track in full_tracks:
2603                try:
2604                    yield parse_track(self, track)
2605                except InvalidDataError as err:
2606                    # track.id may still be usable even if its artist/album is not
2607                    item_id = str(track.id) if track.id is not None else None
2608                    self.report_skipped_sync_item(MediaType.TRACK, item_id, err)
2609
2610    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
2611        """
2612        Retrieve library playlists from KION Music.
2613
2614        Includes virtual playlists (My Mix and Liked Tracks if enabled), user-created playlists,
2615        and user-liked editorial playlists (returned by a separate API endpoint).
2616        """
2617        yield await self.get_playlist(MY_WAVE_PLAYLIST_ID)
2618        yield await self.get_playlist(LIKED_TRACKS_PLAYLIST_ID)
2619        seen_ids: set[str] = set()
2620        # User-created playlists
2621        playlists = await self.client.get_user_playlists()
2622        for playlist in playlists:
2623            try:
2624                parsed = parse_playlist(self, playlist)
2625                seen_ids.add(parsed.item_id)
2626                yield parsed
2627            except InvalidDataError as err:
2628                # mirrors the "owner_id:kind" id parse_playlist() derives
2629                owner_id = str(playlist.owner.uid) if playlist.owner else str(self.client.user_id)
2630                self.report_skipped_sync_item(
2631                    MediaType.PLAYLIST, f"{owner_id}:{playlist.kind}", err
2632                )
2633        # User-liked editorial playlists (not in users_playlists_list)
2634        liked_playlists = await self.client.get_liked_playlists()
2635        for playlist in liked_playlists:
2636            try:
2637                parsed = parse_playlist(self, playlist)
2638                if parsed.item_id not in seen_ids:
2639                    yield parsed
2640            except InvalidDataError as err:
2641                # mirrors the "owner_id:kind" id parse_playlist() derives
2642                owner_id = str(playlist.owner.uid) if playlist.owner else str(self.client.user_id)
2643                self.report_skipped_sync_item(
2644                    MediaType.PLAYLIST, f"{owner_id}:{playlist.kind}", err
2645                )
2646
2647    # Library edit methods
2648
2649    async def library_add(self, item: MediaItemType) -> bool:
2650        """
2651        Add item to library.
2652
2653        :param item: The media item to add.
2654        :return: True if successful.
2655        """
2656        prov_item_id = self._get_provider_item_id(item)
2657        if not prov_item_id:
2658            return False
2659        track_id, _ = _parse_radio_item_id(prov_item_id)
2660
2661        if item.media_type == MediaType.TRACK:
2662            return await self.client.like_track(track_id)
2663        if item.media_type == MediaType.ALBUM:
2664            return await self.client.like_album(prov_item_id)
2665        if item.media_type == MediaType.ARTIST:
2666            return await self.client.like_artist(prov_item_id)
2667        return False
2668
2669    async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
2670        """
2671        Remove item from library.
2672
2673        :param prov_item_id: The provider item ID (may be track_id@station_id for tracks).
2674        :param media_type: The media type.
2675        :return: True if successful.
2676        """
2677        track_id, _ = _parse_radio_item_id(prov_item_id)
2678        if media_type == MediaType.TRACK:
2679            return await self.client.unlike_track(track_id)
2680        if media_type == MediaType.ALBUM:
2681            return await self.client.unlike_album(prov_item_id)
2682        if media_type == MediaType.ARTIST:
2683            return await self.client.unlike_artist(prov_item_id)
2684        return False
2685
2686    def _get_provider_item_id(self, item: MediaItemType) -> str | None:
2687        """Get provider item ID from media item."""
2688        for mapping in item.provider_mappings:
2689            if mapping.provider_instance == self.instance_id:
2690                return mapping.item_id
2691        return item.item_id if item.provider == self.instance_id else None
2692
2693    # Streaming
2694
2695    async def get_stream_details(
2696        self, item_id: str, media_type: MediaType = MediaType.TRACK
2697    ) -> StreamDetails:
2698        """
2699        Get stream details for a track.
2700
2701        :param item_id: The track ID (or track_id@station_id for My Mix).
2702        :param media_type: The media type (should be TRACK).
2703        :return: StreamDetails for the track.
2704        """
2705        return await self.streaming.get_stream_details(item_id)
2706
2707    async def get_audio_stream(
2708        self, streamdetails: StreamDetails, seek_position: int = 0
2709    ) -> AsyncGenerator[bytes]:
2710        """
2711        Return the audio stream for the provider item.
2712
2713        Uses windowed Range-request streaming to prevent Kion CDN drops.
2714        Handles both raw (direct) and encrypted (encraw) transports.
2715
2716        :param streamdetails: Stream details with URL and optional decryption key.
2717        :param seek_position: Seek position in seconds (handled by provider for raw transport).
2718        :return: Async generator yielding audio chunks.
2719        """
2720        async for chunk in self.streaming.get_audio_stream(streamdetails, seek_position):
2721            yield chunk
2722
2723    async def get_rotor_station_tracks(
2724        self, station_id: str, queue: str | int | None = None
2725    ) -> tuple[list[Any], str | None]:
2726        """
2727        Fetch tracks from a rotor station (My Mix, similar, etc.).
2728
2729        Wrapper around client.get_rotor_station_tracks for use by ynison plugin.
2730        """
2731        return await self.client.get_rotor_station_tracks(station_id, queue=queue)
2732
2733    def get_quality(self) -> str:
2734        """
2735        Return the configured audio quality tier (e.g. 'balanced', 'superb').
2736
2737        Mirrors the legacy-value normalization used by the streaming layer:
2738        older configs store the lossless tier as ``"lossless"``, while the
2739        current canonical value is ``QUALITY_LOSSLESS`` (``"superb"``).
2740        External callers (e.g. the ynison plugin wrapper) see the same
2741        normalized value the streaming code would resolve to.
2742        """
2743        quality = str(self.config.get_value(CONF_QUALITY) or "").strip().lower()
2744        if quality == "lossless":
2745            quality = QUALITY_LOSSLESS
2746        return quality
2747
2748    async def resolve_image(self, path: str) -> str | bytes:
2749        """
2750        Resolve wave cover image with background color fill for transparent PNGs.
2751
2752        If the image URL has an associated background color (stored in _wave_bg_colors),
2753        downloads the PNG from Kion CDN and composites it on a solid color background
2754        using Pillow, returning JPEG bytes. Falls back to the original URL on any error.
2755
2756        :param path: Image URL (may include #rrggbb fragment used as cache key).
2757        :return: Composited JPEG bytes, or original path string as fallback.
2758        """
2759        bg_color = self._wave_bg_colors.get(path)
2760        if not bg_color:
2761            return path
2762
2763        # Strip the #color fragment before fetching the actual image
2764        fetch_url = path.split("#", maxsplit=1)[0] if "#" in path else path
2765        try:
2766            async with self.mass.http_session.get(fetch_url) as resp:
2767                resp.raise_for_status()
2768                raw = await resp.read()
2769        except Exception as err:
2770            self.logger.debug("Failed to fetch wave cover %s: %s", fetch_url, err)
2771            return fetch_url
2772
2773        def _composite() -> bytes:
2774            bg_clean = bg_color.lstrip("#")
2775            try:
2776                r = int(bg_clean[0:2], 16)
2777                g = int(bg_clean[2:4], 16)
2778                b = int(bg_clean[4:6], 16)
2779            except ValueError, IndexError:
2780                return raw
2781            fg = PilImage.open(BytesIO(raw)).convert("RGBA")
2782            bg = PilImage.new("RGBA", fg.size, (r, g, b, 255))
2783            bg.paste(fg, mask=fg)
2784            out = BytesIO()
2785            bg.convert("RGB").save(out, "JPEG", quality=92)
2786            return out.getvalue()
2787
2788        try:
2789            return await asyncio.to_thread(_composite)
2790        except Exception as err:
2791            self.logger.debug("Wave cover composite failed for %s: %s", fetch_url, err)
2792            return fetch_url
2793
2794    async def on_played(
2795        self,
2796        media_type: MediaType,
2797        prov_item_id: str,
2798        fully_played: bool,
2799        position: int,
2800        media_item: MediaItemType,
2801        is_playing: bool = False,
2802    ) -> None:
2803        """
2804        Report playback for rotor feedback when the track is from My Mix.
2805
2806        Sends trackStarted when the track is currently playing (is_playing=True).
2807        trackFinished/skip are sent from on_streamed to use accurate seconds_streamed.
2808        """
2809        if media_type != MediaType.TRACK:
2810            return
2811        track_id, station_id = _parse_radio_item_id(prov_item_id)
2812        if not station_id:
2813            return
2814        if is_playing:
2815            if station_id == ROTOR_STATION_MY_MIX:
2816                batch_id = self._my_wave_batch_id
2817            else:
2818                state = self._wave_states.get(station_id)
2819                batch_id = state.batch_id if state else None
2820            await self.client.send_rotor_station_feedback(
2821                station_id,
2822                "trackStarted",
2823                track_id=track_id,
2824                batch_id=batch_id,
2825            )
2826
2827    async def on_streamed(self, streamdetails: StreamDetails) -> None:
2828        """
2829        Report stream completion for My Mix rotor feedback.
2830
2831        Sends trackFinished or skip with actual seconds_streamed so Kion
2832        can improve recommendations.
2833        """
2834        track_id, station_id = _parse_radio_item_id(streamdetails.item_id)
2835        if not station_id:
2836            return
2837        seconds = int(streamdetails.seconds_streamed or 0)
2838        duration = streamdetails.duration or 0
2839        feedback_type = "trackFinished" if duration and seconds >= max(0, duration - 10) else "skip"
2840        if station_id == ROTOR_STATION_MY_MIX:
2841            batch_id = self._my_wave_batch_id
2842        else:
2843            state = self._wave_states.get(station_id)
2844            batch_id = state.batch_id if state else None
2845        await self.client.send_rotor_station_feedback(
2846            station_id,
2847            feedback_type,
2848            track_id=track_id,
2849            total_played_seconds=seconds,
2850            batch_id=batch_id,
2851        )
2852
2853    async def _rotating_row_tag_subtitle(self, category: str) -> str | None:
2854        """
2855        Return the display label of the current rotating tag for a mood/activity row.
2856
2857        Cache-only read of the validated tag list (rows must stay free of backend I/O):
2858        returns None - no subtitle - until an items fetch has warmed that cache.
2859
2860        :param category: Tag category ('mood' or 'activity').
2861        """
2862        # key mirrors the @use_cache key construction on _get_valid_tags_for_category:
2863        # the wrapped function's __name__ (preserved by functools.wraps, so it survives
2864        # renames) plus its positional args, joined by dots
2865        tags, _, found = await self.mass.cache.get_with_freshness(
2866            f"{self._get_valid_tags_for_category.__name__}.{category}",
2867            provider=self.instance_id,
2868            include_expired=True,
2869        )
2870        if not found or not tags:
2871            return None
2872        tag = self._rotating_row_tag(category, tags)
2873        return self._media_source_name("folder", _media_label_key(tag)) or tag.title()
2874
2875    def _rotating_row_tag(self, category: str, valid_tags: list[str]) -> str:
2876        """
2877        Deterministically pick the current hour's tag for a mood/activity row.
2878
2879        Rows and items derive the same tag independently - no shared state, so
2880        concurrent clients (or multiple users on one instance) can never make the
2881        served items mismatch the row subtitle. The pick rotates hourly and
2882        differs per provider instance.
2883
2884        :param category: Tag category the tags belong to.
2885        :param valid_tags: Non-empty list of valid tag slugs to pick from.
2886        """
2887        hour_bucket = int(utc().timestamp()) // 3600
2888        seed = f"{self.instance_id}.{category}.{hour_bucket}".encode()
2889        return sorted(valid_tags)[zlib.crc32(seed) % len(valid_tags)]
2890