music-assistant-server

1"""Zvuk Music provider implementation."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any
6from urllib.parse import urlparse
7
8import aiohttp
9from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
10from music_assistant_models.enums import (
11    ConfigEntryType,
12    ContentType,
13    MediaType,
14    ProviderFeature,
15    StreamType,
16)
17from music_assistant_models.errors import (
18    InvalidDataError,
19    LoginFailed,
20    MediaNotFoundError,
21    ProviderUnavailableError,
22    ResourceTemporarilyUnavailable,
23)
24from music_assistant_models.media_items import (
25    Album,
26    Artist,
27    AudioFormat,
28    BrowseFolder,
29    ItemMapping,
30    MediaItemMetadata,
31    MediaItemType,
32    Playlist,
33    RecommendationFolder,
34    SearchResults,
35    Track,
36    UniqueList,
37)
38from music_assistant_models.streamdetails import StreamDetails
39
40from music_assistant.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER
41from music_assistant.controllers.cache import use_cache
42from music_assistant.models.music_provider import MusicProvider
43
44from .api_client import ZvukMusicClient
45from .constants import (
46    CONF_QUALITY,
47    CONF_TOKEN,
48    DEFAULT_LIMIT,
49    PLAYLIST_TRACK_FETCH_LIMIT,
50    PLAYLIST_TRACKS_PAGE_SIZE,
51    QUALITY_HIGH,
52    QUALITY_LOSSLESS,
53    SYNTHESIS_PLAYLIST_IDS,
54)
55from .parsers import parse_album, parse_artist, parse_playlist, parse_track
56
57if TYPE_CHECKING:
58    from collections.abc import AsyncGenerator
59
60
61class ZvukMusicProvider(MusicProvider):
62    """Implementation of a Zvuk Music MusicProvider."""
63
64    _client: ZvukMusicClient | None = None
65
66    @property
67    def client(self) -> ZvukMusicClient:
68        """Return the Zvuk Music client."""
69        if self._client is None:
70            raise ProviderUnavailableError("Provider not initialized")
71        return self._client
72
73    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
74        """Return Config entries to configure this provider."""
75        return (
76            CONF_ENTRY_UNOFFICIAL_PROVIDER,
77            ConfigEntry(
78                key=CONF_QUALITY,
79                type=ConfigEntryType.STRING,
80                options=[
81                    ConfigValueOption(QUALITY_HIGH),
82                    ConfigValueOption(QUALITY_LOSSLESS),
83                ],
84                default_value=QUALITY_HIGH,
85            ),
86        )
87
88    async def handle_async_init(self) -> None:
89        """Handle async initialization of the provider."""
90        token = self.get_setup_value(CONF_TOKEN)
91        if not token:
92            raise LoginFailed("No Zvuk Music token provided")
93
94        self._client = ZvukMusicClient(str(token))
95        await self._client.connect()
96        self.logger.info("Successfully connected to Zvuk Music")
97
98    async def unload(self, is_removed: bool = False) -> None:
99        """
100        Handle unload/close of the provider.
101
102        :param is_removed: Whether the provider is being removed.
103        """
104        if self._client:
105            await self._client.disconnect()
106        self._client = None
107        await super().unload(is_removed)
108
109    def get_item_mapping(self, media_type: MediaType | str, key: str, name: str) -> ItemMapping:
110        """
111        Create a generic item mapping.
112
113        :param media_type: The media type.
114        :param key: The item ID.
115        :param name: The item name.
116        :return: An ItemMapping instance.
117        """
118        if isinstance(media_type, str):
119            media_type = MediaType(media_type)
120        return ItemMapping(
121            media_type=media_type,
122            item_id=key,
123            provider=self.instance_id,
124            name=name,
125        )
126
127    # Search
128
129    @use_cache(3600 * 24 * 14)
130    async def search(
131        self, search_query: str, media_types: list[MediaType], limit: int = 5
132    ) -> SearchResults:
133        """
134        Perform search on Zvuk Music.
135
136        :param search_query: The search query.
137        :param media_types: List of media types to search for.
138        :param limit: Maximum number of results per type.
139        :return: SearchResults with found items.
140        """
141        result = SearchResults()
142
143        search_result = await self.client.search(
144            search_query,
145            limit=limit,
146            search_tracks=MediaType.TRACK in media_types,
147            search_artists=MediaType.ARTIST in media_types,
148            search_releases=MediaType.ALBUM in media_types,
149            search_playlists=MediaType.PLAYLIST in media_types,
150        )
151        if not search_result:
152            return result
153
154        # Parse tracks
155        if MediaType.TRACK in media_types and search_result.tracks:
156            for track in search_result.tracks.items[:limit]:
157                try:
158                    result.tracks = [*result.tracks, parse_track(self, track)]
159                except InvalidDataError as err:
160                    self.logger.debug("Error parsing track: %s", err)
161
162        # Parse albums (Zvuk releases)
163        if MediaType.ALBUM in media_types and search_result.releases:
164            for release in search_result.releases.items[:limit]:
165                try:
166                    result.albums = [*result.albums, parse_album(self, release)]
167                except InvalidDataError as err:
168                    self.logger.debug("Error parsing album: %s", err)
169
170        # Parse artists
171        if MediaType.ARTIST in media_types and search_result.artists:
172            for artist in search_result.artists.items[:limit]:
173                try:
174                    result.artists = [*result.artists, parse_artist(self, artist)]
175                except InvalidDataError as err:
176                    self.logger.debug("Error parsing artist: %s", err)
177
178        # Parse playlists
179        if MediaType.PLAYLIST in media_types and search_result.playlists:
180            for playlist in search_result.playlists.items[:limit]:
181                try:
182                    result.playlists = [*result.playlists, parse_playlist(self, playlist)]
183                except InvalidDataError as err:
184                    self.logger.debug("Error parsing playlist: %s", err)
185
186        return result
187
188    # Get single items
189
190    @use_cache(3600 * 24 * 30)
191    async def get_artist(self, prov_artist_id: str) -> Artist:
192        """
193        Get artist details by ID.
194
195        :param prov_artist_id: The provider artist ID.
196        :return: Artist object.
197        :raises MediaNotFoundError: If artist not found.
198        """
199        artist = await self.client.get_artist(prov_artist_id)
200        if not artist:
201            raise MediaNotFoundError(f"Artist {prov_artist_id} not found")
202        return parse_artist(self, artist)
203
204    @use_cache(3600 * 24 * 30)
205    async def get_album(self, prov_album_id: str) -> Album:
206        """
207        Get album details by ID.
208
209        :param prov_album_id: The provider album ID.
210        :return: Album object.
211        :raises MediaNotFoundError: If album not found.
212        """
213        release = await self.client.get_release(prov_album_id)
214        if not release:
215            raise MediaNotFoundError(f"Album {prov_album_id} not found")
216        return parse_album(self, release)
217
218    @use_cache(3600 * 24 * 30)
219    async def get_track(self, prov_track_id: str) -> Track:
220        """
221        Get track details by ID.
222
223        :param prov_track_id: The provider track ID.
224        :return: Track object.
225        :raises MediaNotFoundError: If track not found.
226        """
227        track = await self.client.get_track(prov_track_id)
228        if not track:
229            raise MediaNotFoundError(f"Track {prov_track_id} not found")
230        return parse_track(self, track)
231
232    @use_cache(3600 * 24 * 30)
233    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
234        """
235        Get playlist details by ID.
236
237        :param prov_playlist_id: The provider playlist ID.
238        :return: Playlist object.
239        :raises MediaNotFoundError: If playlist not found.
240        """
241        playlist = await self.client.get_playlist(prov_playlist_id)
242        if not playlist:
243            raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found")
244        return parse_playlist(self, playlist)
245
246    # Get related items
247
248    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
249    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
250        """
251        Get album tracks.
252
253        :param prov_album_id: The provider album ID.
254        :return: List of Track objects.
255        """
256        release = await self.client.get_release(prov_album_id)
257        if not release or not release.tracks:
258            return []
259
260        tracks = []
261        for index, track in enumerate(release.tracks):
262            try:
263                parsed_track = parse_track(self, track)
264                parsed_track.disc_number = 1
265                parsed_track.track_number = index + 1
266                tracks.append(parsed_track)
267            except InvalidDataError as err:
268                self.logger.debug("Error parsing album track: %s", err)
269        return tracks
270
271    @use_cache(3600 * 3, allow_expired_cache=True)
272    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
273        """
274        Get playlist tracks.
275
276        :param prov_playlist_id: The provider playlist ID.
277        :param page: Page number for pagination.
278        :return: List of Track objects.
279        """
280        offset = page * PLAYLIST_TRACKS_PAGE_SIZE
281        simple_tracks = await self.client.get_playlist_tracks(
282            prov_playlist_id, limit=PLAYLIST_TRACKS_PAGE_SIZE, offset=offset
283        )
284        if not simple_tracks:
285            return []
286
287        # Fetch full track details from SimpleTrack IDs
288        track_ids = [str(t.id) for t in simple_tracks if t.id]
289        if not track_ids:
290            return []
291
292        full_tracks = await self.client.get_tracks(track_ids)
293        tracks = []
294        for track in full_tracks:
295            try:
296                tracks.append(parse_track(self, track))
297            except InvalidDataError as err:
298                self.logger.debug("Error parsing playlist track: %s", err)
299        return tracks
300
301    @use_cache(3600 * 24 * 7, allow_expired_cache=True)
302    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
303        """
304        Get artist's albums.
305
306        :param prov_artist_id: The provider artist ID.
307        :return: List of Album objects.
308        """
309        artists = await self.client.get_artist_releases(prov_artist_id, limit=DEFAULT_LIMIT)
310        if not artists:
311            return []
312
313        result = []
314        for artist in artists:
315            for release in artist.releases:
316                try:
317                    result.append(parse_album(self, release))
318                except InvalidDataError as err:
319                    self.logger.debug("Error parsing artist album: %s", err)
320        return result
321
322    @use_cache(3600 * 24 * 7, allow_expired_cache=True)
323    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
324        """
325        Get artist's top tracks.
326
327        :param prov_artist_id: The provider artist ID.
328        :return: List of Track objects.
329        """
330        artists = await self.client.get_artist_top_tracks(prov_artist_id, limit=DEFAULT_LIMIT)
331        if not artists:
332            return []
333
334        result = []
335        for artist in artists:
336            for track in artist.popular_tracks:
337                try:
338                    result.append(parse_track(self, track))
339                except InvalidDataError as err:
340                    self.logger.debug("Error parsing artist track: %s", err)
341        return result
342
343    @use_cache(3600 * 24 * 7, allow_expired_cache=True)
344    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
345        """
346        Get similar tracks based on related releases of the track's album.
347
348        Uses the Zvuk ``release.related`` field to find similar releases and samples tracks from
349        them. Only called if provider supports ProviderFeature.SIMILAR_TRACKS.
350
351        :param prov_track_id: The provider track ID.
352        :param limit: Maximum number of similar tracks to return.
353        :return: List of Track objects.
354        """
355        track = await self.client.get_track(prov_track_id)
356        if not track or not track.release:
357            return []
358
359        release = await self.client.get_release(str(track.release.id))
360        if not release or not getattr(release, "related", None):
361            return []
362
363        result: list[Track] = []
364        for related_release in release.related:
365            if len(result) >= limit:
366                break
367            if not getattr(related_release, "id", None):
368                continue
369            related_full = await self.client.get_release(str(related_release.id))
370            if not related_full or not related_full.tracks:
371                continue
372            for t in related_full.tracks[:2]:
373                try:
374                    result.append(parse_track(self, t))
375                except InvalidDataError as err:
376                    self.logger.debug("Error parsing similar track: %s", err)
377                if len(result) >= limit:
378                    break
379
380        return result[:limit]
381
382    # Library methods
383
384    async def get_library_artists(self) -> AsyncGenerator[Artist]:
385        """Retrieve library artists from Zvuk Music."""
386        collection = await self.client.get_collection()
387        if not collection or not collection.artists:
388            return
389        ids = [str(item.id) for item in collection.artists if item.id]
390        async for artist in self._iter_batched(
391            ids, self.client.get_artists, parse_artist, "artist"
392        ):
393            yield artist
394
395    async def get_library_albums(self) -> AsyncGenerator[Album]:
396        """Retrieve library albums from Zvuk Music."""
397        collection = await self.client.get_collection()
398        if not collection or not collection.releases:
399            return
400        ids = [str(item.id) for item in collection.releases if item.id]
401        async for album in self._iter_batched(ids, self.client.get_releases, parse_album, "album"):
402            yield album
403
404    async def get_library_tracks(self) -> AsyncGenerator[Track]:
405        """Retrieve library tracks from Zvuk Music."""
406        collection = await self.client.get_collection()
407        if not collection or not collection.tracks:
408            return
409        ids = [str(item.id) for item in collection.tracks if item.id]
410        async for track in self._iter_batched(ids, self.client.get_tracks, parse_track, "track"):
411            yield track
412
413    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
414        """
415        Retrieve library playlists from Zvuk Music.
416
417        Yields user's own playlists followed by Zvuk's personalized synthesis
418        playlists («Плейлисты для вас»: IDs 3, 4, 6, 11, 12, 13, 14, 15).
419        """
420        collection_items = await self.client.get_user_playlists()
421        if collection_items:
422            ids = [str(item.id) for item in collection_items if item.id]
423            async for playlist in self._iter_batched(
424                ids, self.client.get_playlists, parse_playlist, "playlist"
425            ):
426                yield playlist
427
428        # Synthesis playlists — personalized AI playlists («Плейлисты для вас»)
429        synthesis_playlists = await self.client.get_short_playlists(SYNTHESIS_PLAYLIST_IDS)
430        for simple_pl in synthesis_playlists:
431            try:
432                yield parse_playlist(self, simple_pl)
433            except InvalidDataError as err:
434                self.logger.debug("Error parsing synthesis playlist: %s", err)
435
436    async def get_recommendations(self) -> list[RecommendationFolder]:
437        """
438        Return the available recommendation rows, without items.
439
440        Two rows:
441        - "for_you" ("Made for you"): Zvuk's AI-generated personalized playlists.
442        - "editorial" ("Collections"): Editorial genre-themed curated playlists.
443        """
444        return [
445            RecommendationFolder(
446                item_id="for_you",
447                provider=self.instance_id,
448                name="Made for you",
449                translation_key="made_for_you",
450                icon="mdi-playlist-music",
451            ),
452            RecommendationFolder(
453                item_id="editorial",
454                provider=self.instance_id,
455                name="Collections",
456                subtitle="Editorial playlists from Zvuk by genre",
457                translation_key="editorial",
458                icon="mdi-music-box-multiple",
459            ),
460        ]
461
462    async def get_recommendation_items(
463        self, item_id: str
464    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
465        """
466        Return the items for a single recommendation row.
467
468        :param item_id: The item_id of the row, as returned by get_recommendations.
469        """
470        if item_id == "for_you":
471            return UniqueList(await self._get_for_you_playlists())
472        if item_id == "editorial":
473            return UniqueList(await self._get_editorial_playlists())
474        return UniqueList()
475
476    async def browse(self, path: str) -> list[MediaItemType | ItemMapping | BrowseFolder]:
477        """
478        Browse provider content as a hierarchical folder tree.
479
480        Root level exposes two folders:
481        - "Made for you": Zvuk AI-generated personalized playlists.
482        - "Collections": Editorial genre-themed curated playlists.
483
484        :param path: Browse path (e.g. ``provider_id://`` or ``provider_id://for_you``).
485        :return: List of playlists or BrowseFolders.
486        """
487        if ProviderFeature.BROWSE not in self.supported_features:
488            raise NotImplementedError
489
490        subpath: str | None = None
491        if "://" in path:
492            after_scheme = path.split("://", 1)[1].strip("/")
493            if after_scheme:
494                subpath = after_scheme.split("/", 1)[0]
495
496        if subpath == "for_you":
497            return list(await self._get_for_you_playlists())
498
499        if subpath == "editorial":
500            return list(await self._get_editorial_playlists())
501
502        if subpath is not None:
503            raise KeyError(f"Unsupported browse path: {path}")
504
505        # Root level — return top-level folders
506        base = f"{self.instance_id}://"
507        return [
508            BrowseFolder(
509                item_id="for_you",
510                provider=self.instance_id,
511                path=f"{base}for_you",
512                name="Made for you",
513                translation_key="made_for_you",
514            ),
515            BrowseFolder(
516                item_id="editorial",
517                provider=self.instance_id,
518                path=f"{base}editorial",
519                name="Collections",
520                translation_key="editorial",
521            ),
522        ]
523
524    async def get_track_metadata(self, track: Track) -> MediaItemMetadata | None:
525        """
526        Fetch lyrics for a track from Zvuk's lyrics API.
527
528        Called by MA when ``ProviderFeature.TRACK_METADATA`` is declared.
529        Returns LRC-synced lyrics (``lrc_lyrics``) when the API returns type
530        ``'subtitle'``, otherwise plain text (``lyrics``). Returns ``None`` if
531        the track has no lyrics. Any API errors are caught and return ``None``
532        so that optional metadata enrichment never fails the broader media load.
533
534        :param track: The MA Track object. ``item_id`` is used to call the API.
535        :return: MediaItemMetadata with lyrics, or None.
536        """
537        track_id = track.item_id
538        try:
539            result = await self.client.get_lyrics(track_id)
540        except (ResourceTemporarilyUnavailable, ProviderUnavailableError, LoginFailed) as err:
541            self.logger.debug("Failed to fetch lyrics for track %s: %s", track_id, err)
542            return None
543        except Exception as err:
544            self.logger.debug(
545                "Unexpected error while fetching lyrics for track %s: %s", track_id, err
546            )
547            return None
548        if not result or not result.lyrics:
549            return None
550
551        metadata = MediaItemMetadata()
552        if result.is_synced:
553            metadata.lrc_lyrics = result.lyrics
554        else:
555            metadata.lyrics = result.lyrics
556        return metadata
557
558    async def resolve_image(self, path: str) -> str | bytes:
559        """
560        Fetch a Zvuk image with optional authentication.
561
562        Called by MA when a ``MediaItemImage`` has ``remotely_accessible=False``.
563        Static playlist avatar images (``/static/avatar/playlist/...``) require
564        Zvuk auth cookies and cannot be fetched anonymously.
565
566        Auth token is only sent to trusted Zvuk domains (``zvuk.com``,
567        ``cdn.zvuk.com``, ``cdn-image.zvuk.com``) to prevent token leakage.
568        Images from Sbercloud CDN (``*.sbercloud.ru``) are fetched without auth
569        as they are publicly accessible.
570
571        :param path: Full image URL.
572        :return: Raw image bytes on success, original URL string as fallback.
573        """
574        _zvuk_auth_hosts = frozenset({"zvuk.com", "cdn.zvuk.com", "cdn-image.zvuk.com"})
575        parsed = urlparse(path)
576        hostname = parsed.hostname or ""
577        is_zvuk = hostname in _zvuk_auth_hosts
578        is_sbercloud = hostname.endswith(".sbercloud.ru")
579        if not is_zvuk and not is_sbercloud:
580            self.logger.warning("Refusing to fetch image from untrusted host: %s", hostname)
581            return str(path)
582        headers: dict[str, str] = {
583            "User-Agent": (
584                "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
585                "AppleWebKit/537.36 (KHTML, like Gecko) "
586                "Chrome/120.0.0.0 Safari/537.36"
587            ),
588        }
589        if is_zvuk:
590            token = self.get_setup_value(CONF_TOKEN)
591            if not token:
592                return str(path)
593            headers["X-Auth-Token"] = str(token)
594            headers["Referer"] = "https://zvuk.com/"
595            headers["Origin"] = "https://zvuk.com"
596        try:
597            async with self.mass.http_session.get(path, headers=headers) as resp:
598                if resp.status == 200:
599                    return bytes(await resp.read())
600        except (aiohttp.ClientError, TimeoutError) as err:
601            self.logger.debug("Failed to resolve image %s: %s", path, err)
602        return str(path)
603
604    # Library edit methods
605
606    async def library_add(self, item: MediaItemType) -> bool:
607        """
608        Add item to library.
609
610        :param item: The media item to add.
611        :return: True if successful.
612        """
613        prov_item_id = self._get_provider_item_id(item)
614        if not prov_item_id:
615            return False
616
617        if item.media_type == MediaType.TRACK:
618            return await self.client.like_track(prov_item_id)
619        if item.media_type == MediaType.ALBUM:
620            return await self.client.like_release(prov_item_id)
621        if item.media_type == MediaType.ARTIST:
622            return await self.client.like_artist(prov_item_id)
623        if item.media_type == MediaType.PLAYLIST:
624            return await self.client.like_playlist(prov_item_id)
625        return False
626
627    async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
628        """
629        Remove item from library.
630
631        :param prov_item_id: The provider item ID.
632        :param media_type: The media type.
633        :return: True if successful.
634        """
635        if media_type == MediaType.TRACK:
636            return await self.client.unlike_track(prov_item_id)
637        if media_type == MediaType.ALBUM:
638            return await self.client.unlike_release(prov_item_id)
639        if media_type == MediaType.ARTIST:
640            return await self.client.unlike_artist(prov_item_id)
641        if media_type == MediaType.PLAYLIST:
642            return await self.client.unlike_playlist(prov_item_id)
643        return False
644
645    # Playlist management
646
647    async def create_playlist(self, name: str, media_types: set[MediaType]) -> Playlist:
648        """
649        Create a new playlist.
650
651        :param name: Playlist name.
652        :param media_types: Ignored — Zvuk playlists are always track-based.
653        :return: The created Playlist object.
654        """
655        playlist_id = await self.client.create_playlist(name)
656        playlist = await self.client.get_playlist(playlist_id)
657        if not playlist:
658            raise MediaNotFoundError(f"Created playlist {playlist_id} not found")
659        return parse_playlist(self, playlist)
660
661    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
662        """
663        Add tracks to a playlist.
664
665        :param prov_playlist_id: The provider playlist ID.
666        :param prov_track_ids: List of track IDs to add.
667        """
668        await self.client.add_tracks_to_playlist(prov_playlist_id, prov_track_ids)
669
670    async def remove_playlist_tracks(
671        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
672    ) -> None:
673        """
674        Remove tracks from a playlist by position.
675
676        :param prov_playlist_id: The provider playlist ID.
677        :param positions_to_remove: Tuple of track positions (0-based) to remove.
678        """
679        # Fetch current tracks and filter out the ones at given positions
680        simple_tracks = await self.client.get_playlist_tracks(
681            prov_playlist_id, limit=PLAYLIST_TRACK_FETCH_LIMIT
682        )
683        remove_positions = set(positions_to_remove)
684        remaining_ids = [
685            str(t.id) for i, t in enumerate(simple_tracks) if t.id and i not in remove_positions
686        ]
687        await self.client.update_playlist(prov_playlist_id, remaining_ids)
688
689    # Streaming
690
691    async def get_stream_details(
692        self, item_id: str, media_type: MediaType = MediaType.TRACK
693    ) -> StreamDetails:
694        """
695        Get stream details for a track.
696
697        Uses /api/tiny/track/stream to get a direct (non-DRM) URL. When lossless is
698        requested, always tries "flac" quality first, then falls back through "high"
699        (320kbps MP3) → "mid" (128kbps MP3). The ``has_flac`` field from the API is
700        not reliable enough to skip the FLAC attempt.
701
702        :param item_id: The track ID.
703        :param media_type: The media type (should be TRACK).
704        :return: StreamDetails for the track.
705        :raises MediaNotFoundError: If stream URL cannot be obtained.
706        """
707        quality_pref = self.config.get_value(CONF_QUALITY)
708        quality_str = str(quality_pref) if quality_pref is not None else QUALITY_LOSSLESS
709
710        # Fetch track metadata for duration.
711        # has_flac is read for diagnostics only — it is not reliable enough to
712        # skip the FLAC attempt, so it does not affect the quality fallback chain.
713        track = await self.client.get_track(item_id)
714        duration: int | None = None
715        has_flac: bool | None = None
716        if track is not None:
717            if getattr(track, "duration", None) is not None:
718                duration = int(track.duration)
719            if getattr(track, "has_flac", None) is not None:
720                has_flac = bool(track.has_flac)
721
722        # Build quality fallback chain using /api/tiny/track/stream (plain, non-DRM URLs).
723        # This endpoint returns {"result": {"stream": "https://..."}} for FLAC/MP3 qualities.
724        # zvuk-dl-rs uses the same endpoint to download lossless FLAC.
725        self.logger.debug(
726            "Stream request for track %s: quality_pref=%s has_flac=%s (diagnostic only)",
727            item_id,
728            quality_str,
729            has_flac,
730        )
731        if quality_str == QUALITY_LOSSLESS:
732            quality_chain = [
733                ("flac", ContentType.FLAC, 0),
734                ("high", ContentType.MP3, 320),
735                ("mid", ContentType.MP3, 128),
736            ]
737        else:
738            quality_chain = [("high", ContentType.MP3, 320), ("mid", ContentType.MP3, 128)]
739
740        url: str | None = None
741        content_type = ContentType.UNKNOWN
742        bitrate = 0
743
744        for q_str, q_content_type, q_bitrate in quality_chain:
745            try:
746                url = await self.client.get_direct_stream_url(item_id, q_str)
747            except (ResourceTemporarilyUnavailable, ProviderUnavailableError) as err:
748                self.logger.warning(
749                    "Error getting stream URL for track %s quality=%s: %s",
750                    item_id,
751                    q_str,
752                    err,
753                )
754                continue
755            self.logger.debug(
756                "Stream URL for track %s quality=%s: %s",
757                item_id,
758                q_str,
759                "OK" if url else "None",
760            )
761            if url:
762                content_type = q_content_type
763                bitrate = q_bitrate
764                break
765
766        if not url:
767            raise MediaNotFoundError(f"No stream URL available for track {item_id}")
768
769        return StreamDetails(
770            item_id=item_id,
771            provider=self.instance_id,
772            audio_format=AudioFormat(
773                content_type=content_type,
774                bit_rate=bitrate,
775            ),
776            stream_type=StreamType.HTTP,
777            path=url,
778            duration=duration,
779            allow_seek=True,
780            can_seek=True,
781        )
782
783    async def _iter_batched(
784        self,
785        ids: list[str],
786        fetcher: Any,
787        parser: Any,
788        item_type: str,
789    ) -> AsyncGenerator[Any]:
790        """
791        Yield parsed items by fetching ``ids`` in batches of DEFAULT_LIMIT.
792
793        :param ids: List of item IDs to fetch.
794        :param fetcher: Async callable that accepts a list of IDs and returns a list of raw items.
795        :param parser: Callable(provider, raw_item) → MA media item.
796        :param item_type: Human-readable type name for debug log messages.
797        """
798        for i in range(0, len(ids), DEFAULT_LIMIT):
799            batch = ids[i : i + DEFAULT_LIMIT]
800            items = await fetcher(batch)
801            for item in items:
802                try:
803                    yield parser(self, item)
804                except InvalidDataError as err:
805                    self.logger.debug("Error parsing library %s: %s", item_type, err)
806
807    async def _get_for_you_playlists(self) -> list[Playlist]:
808        """Fetch and parse Zvuk's personalized synthesis playlists («Плейлисты для вас»)."""
809        synthesis_playlists = await self.client.get_short_playlists(SYNTHESIS_PLAYLIST_IDS)
810        result: list[Playlist] = []
811        for simple_pl in synthesis_playlists:
812            try:
813                result.append(parse_playlist(self, simple_pl))
814            except InvalidDataError as err:
815                self.logger.debug("Error parsing synthesis playlist: %s", err)
816        return result
817
818    async def _get_editorial_playlists(self) -> list[Playlist]:
819        """Fetch and parse Zvuk's editorial curated playlists («Подборки»)."""
820        editorial_ids = await self.client.get_editorial_playlist_ids()
821        if not editorial_ids:
822            return []
823        full_playlists = await self.client.get_playlists(editorial_ids[:DEFAULT_LIMIT])
824        result: list[Playlist] = []
825        for full_pl in full_playlists:
826            try:
827                result.append(parse_playlist(self, full_pl))
828            except InvalidDataError as err:
829                self.logger.debug("Error parsing editorial playlist: %s", err)
830        return result
831
832    def _get_provider_item_id(self, item: MediaItemType) -> str | None:
833        """Get provider item ID from media item."""
834        for mapping in item.provider_mappings:
835            if mapping.provider_instance == self.instance_id:
836                return mapping.item_id
837        return item.item_id if item.provider == self.instance_id else None
838