music-assistant-server

1"""
2Media operations manager for the Deezer provider.
3
4Handles library retrieval, search, item getters, content getters,
5library mutations, and playlist CRUD operations.
6"""
7
8from __future__ import annotations
9
10from collections.abc import AsyncGenerator, Awaitable, Callable
11from typing import TYPE_CHECKING, Any, Protocol
12
13from deezer_python_gql import GraphQLClientGraphQLMultiError
14from music_assistant_models.enums import MediaType
15from music_assistant_models.errors import MediaNotFoundError, UnsupportedFeaturedException
16from music_assistant_models.media_items import (
17    Album,
18    Artist,
19    Audiobook,
20    ItemMapping,
21    MediaItemType,
22    Playlist,
23    Podcast,
24    PodcastEpisode,
25    ProviderMapping,
26    Radio,
27    SearchResults,
28    Track,
29    UniqueList,
30)
31
32from music_assistant.controllers.cache import use_cache
33
34from .constants import (
35    AUDIOBOOK_CHAPTERS_PAGE_SIZE,
36    FAVORITES_PAGE_SIZE,
37    PERSONAL_ALBUM_PREFIX,
38    PERSONAL_ARTIST_PREFIX,
39)
40from .helpers import fetch_all_audiobook_chapter_edges, fetch_all_bookmarks
41from .parsers import (
42    apply_web_url,
43    parse_album,
44    parse_artist,
45    parse_audiobook,
46    parse_audiobook_chapters,
47    parse_audiobook_from_album,
48    parse_date,
49    parse_gw_track,
50    parse_playlist,
51    parse_podcast,
52    parse_podcast_episode,
53    parse_radio,
54    parse_track,
55)
56
57if TYPE_CHECKING:
58    from .provider import DeezerProvider
59
60
61# -- Protocols for typed pagination --
62
63
64class _PageInfo(Protocol):
65    @property
66    def has_next_page(self) -> bool: ...
67
68    @property
69    def end_cursor(self) -> str | None: ...
70
71
72class _Connection(Protocol):
73    @property
74    def edges(self) -> list[Any]: ...
75
76    @property
77    def page_info(self) -> _PageInfo: ...
78
79
80def _is_complexity_error(err: GraphQLClientGraphQLMultiError) -> bool:
81    """Check if a GraphQL error is a query complexity limit violation."""
82    return any("complexity" in e.message.lower() for e in err.errors)
83
84
85class DeezerMediaManager:
86    """Handles library sync, search, item getters, and mutations."""
87
88    def __init__(self, provider: DeezerProvider) -> None:
89        """Initialize media manager."""
90        self.provider = provider
91        self.mass = provider.mass
92        self.instance_id = provider.instance_id
93        self.domain = provider.domain
94        self.logger = provider.logger
95        self._audiobook_ids_in_favorites: set[str] | None = None
96
97    # -- Pagination helper --
98
99    async def _iter_paged(
100        self,
101        fetch: Callable[..., Awaitable[Any]],
102        extract: Callable[..., _Connection | None],
103    ) -> AsyncGenerator[Any]:
104        """Iterate a cursor-paginated connection, yielding edges with non-null nodes."""
105        cursor: str | None = None
106        while True:
107            result = await fetch(first=FAVORITES_PAGE_SIZE, after=cursor)
108            if result is None:
109                break
110            connection = extract(result)
111            if connection is None:
112                break
113            for edge in connection.edges:
114                if edge.node is not None:
115                    yield edge
116            if not connection.page_info.has_next_page:
117                break
118            cursor = connection.page_info.end_cursor
119
120    # -- Personal songs cache --
121
122    @use_cache(3600 * 24)
123    async def _get_personal_songs(self) -> list[dict[str, Any]]:
124        """Fetch all user-uploaded personal songs via the GW API (cached 24h)."""
125        all_songs: list[dict[str, Any]] = []
126        start = 0
127        page_size = 500
128        while True:
129            results = await self.provider.gw_client.get_personal_songs(start=start, nb=page_size)
130            data: list[dict[str, Any]] = results.get("data", [])
131            all_songs.extend(data)
132            if len(data) < page_size:
133                break
134            start += page_size
135        return all_songs
136
137    # -- Library retrieval --
138
139    async def get_library_artists(self) -> AsyncGenerator[Artist]:
140        """Retrieve all library artists from Deezer."""
141        async for edge in self._iter_paged(
142            self.provider.gql_client.get_favorite_artists,
143            lambda r: r.user_favorites.artists,
144        ):
145            item = parse_artist(self.provider, edge.node)
146            if edge.favorited_at:
147                item.date_added = parse_date(edge.favorited_at)
148            yield item
149        # Also include artists from user-uploaded personal songs
150        personal_songs = await self._get_personal_songs()
151        seen_artist_names: set[str] = set()
152        for song in personal_songs:
153            track = parse_gw_track(self.provider, song)
154            for artist in track.artists:
155                if isinstance(artist, Artist) and artist.name not in seen_artist_names:
156                    seen_artist_names.add(artist.name)
157                    yield artist
158
159    async def _get_audiobook_ids_in_albums(self) -> set[str]:
160        """Identify which favorite album IDs are actually audiobooks."""
161        # Deezer stores audiobook favorites in the albums list, not in the
162        # dedicated (deprecated) audiobook favorites endpoint. We use
163        # check_audiobook_ids to tell them apart. Result is cached for the
164        # lifetime of this manager instance so both get_library_albums and
165        # get_library_audiobooks can share it without extra API calls.
166        if self._audiobook_ids_in_favorites is not None:
167            return self._audiobook_ids_in_favorites
168        album_ids: list[str] = []
169        async for edge in self._iter_paged(
170            self.provider.gql_client.get_favorite_albums,
171            lambda r: r.user_favorites.albums,
172        ):
173            album_ids.append(edge.node.id)
174        if not album_ids:
175            self._audiobook_ids_in_favorites = set()
176        else:
177            self._audiobook_ids_in_favorites = await self.provider.gql_client.check_audiobook_ids(
178                album_ids
179            )
180        return self._audiobook_ids_in_favorites
181
182    async def get_library_albums(self) -> AsyncGenerator[Album]:
183        """Retrieve all library albums from Deezer."""
184        # Collect all favorite album edges in a single pass, then determine
185        # which are audiobooks via check_audiobook_ids, and yield the rest.
186        all_edges: list[Any] = []
187        async for edge in self._iter_paged(
188            self.provider.gql_client.get_favorite_albums,
189            lambda r: r.user_favorites.albums,
190        ):
191            all_edges.append(edge)
192        # Populate the favorites-audiobook cache (shared with get_library_audiobooks)
193        if self._audiobook_ids_in_favorites is None:
194            album_ids = [edge.node.id for edge in all_edges]
195            self._audiobook_ids_in_favorites = (
196                await self.provider.gql_client.check_audiobook_ids(album_ids)
197                if album_ids
198                else set()
199            )
200        for edge in all_edges:
201            if edge.node.id in self._audiobook_ids_in_favorites:
202                continue
203            item = parse_album(self.provider, edge.node)
204            if edge.favorited_at:
205                item.date_added = parse_date(edge.favorited_at)
206            yield item
207        # Also include albums from user-uploaded personal songs
208        personal_songs = await self._get_personal_songs()
209        seen_album_names: set[str] = set()
210        for song in personal_songs:
211            track = parse_gw_track(self.provider, song)
212            if isinstance(track.album, Album) and track.album.name not in seen_album_names:
213                seen_album_names.add(track.album.name)
214                yield track.album
215
216    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
217        """Retrieve all library playlists from Deezer."""
218        # User-owned playlists first
219        seen_ids: set[str] = set()
220        async for edge in self._iter_paged(
221            self.provider.gql_client.get_user_playlists,
222            lambda r: r.playlists,
223        ):
224            seen_ids.add(edge.node.id)
225            yield parse_playlist(self.provider, edge.node, is_editable=True)
226        # Favorited playlists (other users' playlists)
227        async for edge in self._iter_paged(
228            self.provider.gql_client.get_favorite_playlists,
229            lambda r: r.user_favorites.playlists,
230        ):
231            if edge.node.id in seen_ids:
232                continue
233            item = parse_playlist(self.provider, edge.node)
234            if edge.favorited_at:
235                item.date_added = parse_date(edge.favorited_at)
236            yield item
237
238    async def get_library_tracks(self) -> AsyncGenerator[Track]:
239        """Retrieve all library tracks from Deezer (favorites + personal uploads)."""
240        async for edge in self._iter_paged(
241            self.provider.gql_client.get_favorite_tracks,
242            lambda r: r.user_favorites.tracks,
243        ):
244            item = parse_track(self.provider, edge.node)
245            if edge.favorited_at:
246                item.date_added = parse_date(edge.favorited_at)
247            yield item
248        # Also include user-uploaded personal songs
249        personal_songs = await self._get_personal_songs()
250        for idx, song in enumerate(personal_songs, 1):
251            yield parse_gw_track(self.provider, song, position=idx)
252
253    async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
254        """Retrieve library/subscribed podcasts from Deezer."""
255        async for edge in self._iter_paged(
256            self.provider.gql_client.get_favorite_podcasts,
257            lambda r: r.user_favorites.podcasts,
258        ):
259            item = parse_podcast(self.provider, edge.node)
260            if edge.favorited_at:
261                item.date_added = parse_date(edge.favorited_at)
262            yield item
263
264    async def get_library_audiobooks(self) -> AsyncGenerator[Audiobook]:
265        """
266        Retrieve library/subscribed audiobooks from Deezer.
267
268        Checks both the dedicated (deprecated) audiobook favorites endpoint
269        and the regular favorite albums list, since Deezer stores audiobook
270        favorites in the albums list.
271        """
272        seen_ids: set[str] = set()
273        # 1. Dedicated audiobook favorites (deprecated but may still have entries)
274        result = await self.provider.gql_client.get_favorite_audiobooks()
275        if result is not None and result.favorites.raw_audiobooks is not None:
276            for raw in result.favorites.raw_audiobooks:
277                try:
278                    item = await self.get_audiobook(raw.id)
279                except MediaNotFoundError as err:
280                    self.provider.report_skipped_sync_item(MediaType.AUDIOBOOK, raw.id, err)
281                    continue
282                seen_ids.add(raw.id)
283                if raw.favorited_at:
284                    item.date_added = parse_date(raw.favorited_at)
285                yield item
286        # 2. Audiobooks stored as favorite albums
287        audiobook_ids = await self._get_audiobook_ids_in_albums()
288        for ab_id in audiobook_ids:
289            if ab_id in seen_ids:
290                continue
291            try:
292                yield await self.get_audiobook(ab_id)
293            except MediaNotFoundError as err:
294                self.provider.report_skipped_sync_item(MediaType.AUDIOBOOK, ab_id, err)
295                continue
296
297    # -- Search --
298
299    @use_cache(60 * 15)
300    async def search(
301        self, search_query: str, media_types: list[MediaType], limit: int = 5
302    ) -> SearchResults:
303        """Perform search on music provider."""
304        self.logger.debug("search called with media_types=%s", media_types)
305        need_albums = MediaType.ALBUM in media_types
306        need_audiobooks = MediaType.AUDIOBOOK in media_types
307
308        # Try with full limit first; on complexity error, retry with reduced limits.
309        attempts = [limit, max(limit // 2, 5), 5]
310        result = None
311        for idx, attempt_limit in enumerate(attempts):
312            try:
313                result = await self.provider.gql_client.search(
314                    query=search_query,
315                    tracks_first=attempt_limit if MediaType.TRACK in media_types else 0,
316                    albums_first=attempt_limit if (need_albums or need_audiobooks) else 0,
317                    artists_first=attempt_limit if MediaType.ARTIST in media_types else 0,
318                    playlists_first=attempt_limit if MediaType.PLAYLIST in media_types else 0,
319                    livestreams_first=attempt_limit if MediaType.RADIO in media_types else 0,
320                    podcasts_first=attempt_limit if MediaType.PODCAST in media_types else 0,
321                )
322                break
323            except GraphQLClientGraphQLMultiError as err:
324                if not _is_complexity_error(err):
325                    raise
326                if idx == len(attempts) - 1:
327                    self.logger.warning("Search complexity exceeded even at minimum limit")
328                    raise
329                self.logger.debug(
330                    "Search complexity exceeded at limit=%d, retrying with %d",
331                    attempt_limit,
332                    attempts[idx + 1],
333                )
334        search_results = SearchResults()
335        if result is None:
336            return search_results
337        if MediaType.TRACK in media_types:
338            search_results.tracks = [
339                parse_track(self.provider, edge.node)
340                for edge in result.results.tracks.edges
341                if edge.node is not None
342            ]
343        if need_albums or need_audiobooks:
344            album_nodes = [e.node for e in result.results.albums.edges if e.node is not None]
345            if album_nodes and need_audiobooks:
346                album_ids = [n.id for n in album_nodes]
347                audiobook_ids = await self.provider.gql_client.check_audiobook_ids(album_ids)
348                if need_albums:
349                    search_results.albums = [
350                        parse_album(self.provider, n)
351                        for n in album_nodes
352                        if n.id not in audiobook_ids
353                    ]
354                search_results.audiobooks = [
355                    parse_audiobook_from_album(self.provider, n)
356                    for n in album_nodes
357                    if n.id in audiobook_ids
358                ]
359            elif need_albums:
360                search_results.albums = [parse_album(self.provider, n) for n in album_nodes]
361        if MediaType.ARTIST in media_types:
362            search_results.artists = [
363                parse_artist(self.provider, edge.node)
364                for edge in result.results.artists.edges
365                if edge.node is not None
366            ]
367        if MediaType.PLAYLIST in media_types:
368            search_results.playlists = [
369                parse_playlist(self.provider, edge.node)
370                for edge in result.results.playlists.edges
371                if edge.node is not None
372            ]
373        if MediaType.RADIO in media_types:
374            search_results.radio = [
375                parse_radio(self.provider, edge.node)
376                for edge in result.results.livestreams.edges
377                if edge.node is not None
378            ]
379        if MediaType.PODCAST in media_types:
380            search_results.podcasts = [
381                parse_podcast(self.provider, edge.node)
382                for edge in result.results.podcasts.edges
383                if edge.node is not None
384            ]
385        return search_results
386
387    # -- Item getters --
388
389    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
390    async def get_artist(self, prov_artist_id: str) -> Artist:
391        """Get full artist details by id."""
392        if prov_artist_id.startswith(PERSONAL_ARTIST_PREFIX):
393            # Personal track artist — reconstruct from GW data
394            song_id = prov_artist_id.removeprefix(PERSONAL_ARTIST_PREFIX)
395            personal_songs = await self._get_personal_songs()
396            for song in personal_songs:
397                if str(song["SNG_ID"]) == song_id:
398                    return Artist(
399                        item_id=prov_artist_id,
400                        provider=self.instance_id,
401                        name=song.get("ART_NAME", ""),
402                        provider_mappings={
403                            ProviderMapping(
404                                item_id=prov_artist_id,
405                                provider_domain=self.domain,
406                                provider_instance=self.instance_id,
407                            )
408                        },
409                    )
410            raise MediaNotFoundError(f"Personal artist {prov_artist_id} not found")
411        result = await self.provider.gql_client.get_artist(artist_id=prov_artist_id)
412        if result is None:
413            raise MediaNotFoundError(f"Artist {prov_artist_id} not found on Deezer")
414        item = parse_artist(self.provider, result)
415        apply_web_url(item, result)
416        return item
417
418    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
419    async def get_album(self, prov_album_id: str) -> Album:
420        """Get full album details by id."""
421        if prov_album_id.startswith(PERSONAL_ALBUM_PREFIX):
422            # Personal track album — reconstruct from GW data
423            song_id = prov_album_id.removeprefix(PERSONAL_ALBUM_PREFIX)
424            personal_songs = await self._get_personal_songs()
425            for song in personal_songs:
426                if str(song["SNG_ID"]) == song_id:
427                    art_name = song.get("ART_NAME", "")
428                    personal_art_id = f"{PERSONAL_ARTIST_PREFIX}{song_id}"
429                    artists: UniqueList[Artist | ItemMapping] = UniqueList()
430                    if art_name:
431                        artists.append(
432                            ItemMapping(
433                                media_type=MediaType.ARTIST,
434                                item_id=personal_art_id,
435                                provider=self.instance_id,
436                                name=art_name,
437                            )
438                        )
439                    return Album(
440                        item_id=prov_album_id,
441                        provider=self.instance_id,
442                        name=song.get("ALB_TITLE", ""),
443                        artists=artists,
444                        provider_mappings={
445                            ProviderMapping(
446                                item_id=prov_album_id,
447                                provider_domain=self.domain,
448                                provider_instance=self.instance_id,
449                            )
450                        },
451                    )
452            raise MediaNotFoundError(f"Personal album {prov_album_id} not found")
453        result = await self.provider.gql_client.get_album(album_id=prov_album_id)
454        if result is None:
455            raise MediaNotFoundError(f"Album {prov_album_id} not found on Deezer")
456        item = parse_album(self.provider, result)
457        apply_web_url(item, result)
458        return item
459
460    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
461    async def get_track(self, prov_track_id: str) -> Track:
462        """Get full track details by id."""
463        try:
464            track_id_int = int(prov_track_id)
465        except ValueError as err:
466            raise MediaNotFoundError(f"Invalid Deezer track ID: {prov_track_id}") from err
467        # Personal tracks (negative IDs) don't exist in the GQL API
468        if track_id_int < 0:
469            personal_songs = await self._get_personal_songs()
470            for song in personal_songs:
471                if str(song["SNG_ID"]) == prov_track_id:
472                    return parse_gw_track(self.provider, song)
473            raise MediaNotFoundError(f"Personal track {prov_track_id} not found")
474        result = await self.provider.gql_client.get_track(track_id=prov_track_id)
475        if result is None:
476            raise MediaNotFoundError(f"Track {prov_track_id} not found on Deezer")
477        return parse_track(self.provider, result)
478
479    @use_cache(3600 * 24 * 30)
480    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
481        """Get full playlist details by id."""
482        if virtual := await self.provider.browse_manager.get_virtual_playlist(prov_playlist_id):
483            return virtual
484        result = await self.provider.gql_client.get_playlist(playlist_id=prov_playlist_id)
485        if result is None:
486            raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found on Deezer")
487        is_editable = result.owner is not None and result.owner.id == self.provider.user_id
488        return parse_playlist(self.provider, result, is_editable=is_editable)
489
490    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
491    async def get_radio(self, prov_radio_id: str) -> Radio:
492        """Get full radio/livestream details by id."""
493        result = await self.provider.gql_client.get_livestream(livestream_id=prov_radio_id)
494        if result is None:
495            raise MediaNotFoundError(f"Radio {prov_radio_id} not found on Deezer")
496        return parse_radio(self.provider, result)
497
498    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
499    async def get_podcast(self, prov_podcast_id: str) -> Podcast:
500        """Get full podcast details by id."""
501        result = await self.provider.gql_client.get_podcast(podcast_id=prov_podcast_id)
502        if result is None:
503            raise MediaNotFoundError(f"Podcast {prov_podcast_id} not found on Deezer")
504        podcast = parse_podcast(self.provider, result)
505        podcast.total_episodes = len(result.raw_episodes)
506        return podcast
507
508    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
509    async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
510        """Get (full) podcast episode details by id."""
511        result = await self.provider.gql_client.get_podcast_episode(
512            podcast_episode_id=prov_episode_id,
513        )
514        if result is None:
515            raise MediaNotFoundError(f"Podcast episode {prov_episode_id} not found on Deezer")
516        podcast_mapping = ItemMapping(
517            media_type=MediaType.PODCAST,
518            item_id=result.podcast.id,
519            provider=self.instance_id,
520            name=result.podcast.display_title,
521        )
522        podcast_image_url = (
523            result.podcast.cover.urls[0]
524            if result.podcast.cover and result.podcast.cover.urls
525            else None
526        )
527        return parse_podcast_episode(self.provider, result, podcast_mapping, 0, podcast_image_url)
528
529    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
530    async def get_audiobook(self, prov_audiobook_id: str) -> Audiobook:
531        """Get full audiobook details by id."""
532        result = await self.provider.gql_client.get_audiobook(
533            audiobook_id=prov_audiobook_id, chapters_first=AUDIOBOOK_CHAPTERS_PAGE_SIZE
534        )
535        if result is None:
536            raise MediaNotFoundError(f"Audiobook {prov_audiobook_id} not found on Deezer")
537        item = parse_audiobook(self.provider, result)
538        if result.chapters.page_info.has_next_page:
539            all_edges = await fetch_all_audiobook_chapter_edges(
540                self.provider.gql_client,
541                prov_audiobook_id,
542                initial_edges=result.chapters.edges,
543                initial_page_info=result.chapters.page_info,
544            )
545        else:
546            all_edges = result.chapters.edges
547        item.metadata.chapters = parse_audiobook_chapters(all_edges)
548        return item
549
550    # -- Content getters --
551
552    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
553    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
554        """Get all tracks in an album."""
555        if prov_album_id.startswith(PERSONAL_ALBUM_PREFIX):
556            # Personal album has no real Deezer album page
557            return []
558        result = await self.provider.gql_client.get_album(album_id=prov_album_id)
559        if result is None:
560            return []
561        all_edges = list(result.tracks.edges)
562        while result.tracks.page_info.has_next_page:
563            result = await self.provider.gql_client.get_album(
564                album_id=prov_album_id,
565                tracks_after=result.tracks.page_info.end_cursor,
566            )
567            if result is None:
568                break
569            all_edges.extend(result.tracks.edges)
570        return [
571            parse_track(self.provider, edge.node, position=idx)
572            for idx, edge in enumerate(all_edges, 1)
573            if edge.node is not None
574        ]
575
576    async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
577        """Get all episodes for a given podcast with current resume state."""
578        episodes = await self._fetch_podcast_episodes(prov_podcast_id)
579        if not episodes:
580            return
581        bookmarks = await fetch_all_bookmarks(self.provider.gql_client)
582        for ep in episodes:
583            ep.fully_played = False
584            ep.resume_position_ms = 0
585            if ep.item_id in bookmarks:
586                ep.fully_played, ep.resume_position_ms = bookmarks[ep.item_id]
587            yield ep
588
589    @use_cache(3600)
590    async def _fetch_podcast_episodes(self, prov_podcast_id: str) -> list[PodcastEpisode]:
591        """Fetch all episodes for a podcast (cached 1h)."""
592        # Two-layer caching strategy:
593        # - Outer (this decorator, 1h): avoids repeated cache lookups during
594        #   rapid navigation (e.g., user browsing back and forth between podcasts).
595        # - Inner (per-episode, 30 days): prevents re-fetching episode details
596        #   that rarely change. When the outer cache expires, only genuinely new
597        #   episodes require an API call.
598        result = await self.provider.gql_client.get_podcast(
599            podcast_id=prov_podcast_id, episodes_first=0
600        )
601        if result is None:
602            return []
603        podcast_mapping = ItemMapping(
604            media_type=MediaType.PODCAST,
605            item_id=result.id,
606            provider=self.instance_id,
607            name=result.display_title,
608        )
609        podcast_image_url: str | None = None
610        if result.cover and result.cover.urls:
611            podcast_image_url = result.cover.urls[0]
612        episode_ids = result.raw_episodes
613        if not episode_ids:
614            return []
615
616        cache = self.mass.cache
617        episode_cache_ttl = 3600 * 24 * 30  # 30 days
618
619        # Resolve cached vs uncached episode IDs
620        cached_episodes: dict[str, PodcastEpisode] = {}
621        uncached_ids: list[str] = []
622        for eid in episode_ids:
623            cache_key = f"podcast_episode.{eid}"
624            cached = await cache.get(cache_key, provider=self.instance_id)
625            if cached is not None:
626                cached_episodes[eid] = PodcastEpisode.from_dict(cached)
627            else:
628                uncached_ids.append(eid)
629
630        # Batch-fetch only uncached episodes
631        batch_size = 50
632        for i in range(0, len(uncached_ids), batch_size):
633            batch = uncached_ids[i : i + batch_size]
634            fetched = await self.provider.gql_client.get_podcast_episodes_by_ids(ids=batch)
635            for ep in fetched:
636                if ep is not None:
637                    parsed = parse_podcast_episode(
638                        self.provider, ep, podcast_mapping, 0, podcast_image_url
639                    )
640                    cached_episodes[ep.id] = parsed
641                    self.mass.create_task(
642                        cache.set(
643                            key=f"podcast_episode.{ep.id}",
644                            data=parsed.to_dict(),
645                            expiration=episode_cache_ttl,
646                            provider=self.instance_id,
647                        )
648                    )
649
650        # Build final list in original order with correct positions
651        episodes: list[PodcastEpisode] = []
652        position = 0
653        for eid in episode_ids:
654            if eid in cached_episodes:
655                position += 1
656                cached_ep = cached_episodes[eid]
657                cached_ep.position = position
658                episodes.append(cached_ep)
659        return episodes
660
661    @use_cache(3600 * 24 * 7, allow_expired_cache=True)
662    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
663        """Get albums by an artist."""
664        if prov_artist_id.startswith(PERSONAL_ARTIST_PREFIX):
665            # Personal artist has no real Deezer artist page
666            return []
667        result = await self.provider.gql_client.get_artist(artist_id=prov_artist_id)
668        if result is None:
669            return []
670        all_edges = list(result.albums.edges)
671        while result.albums.page_info.has_next_page:
672            result = await self.provider.gql_client.get_artist(
673                artist_id=prov_artist_id,
674                albums_after=result.albums.page_info.end_cursor,
675            )
676            if result is None:
677                break
678            all_edges.extend(result.albums.edges)
679        return [
680            parse_album(self.provider, edge.node) for edge in all_edges if edge.node is not None
681        ]
682
683    async def get_artist_topalbums(self, prov_artist_id: str) -> list[Album]:
684        """Get top albums of an artist, ranked by popularity."""
685        albums = await self.get_artist_albums(prov_artist_id)
686        return sorted(albums, key=lambda album: album.metadata.popularity or 0, reverse=True)
687
688    @use_cache(3600 * 24 * 7, allow_expired_cache=True)
689    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
690        """Get top tracks of an artist."""
691        if prov_artist_id.startswith(PERSONAL_ARTIST_PREFIX):
692            # Personal artist has no real Deezer artist page
693            return []
694        result = await self.provider.gql_client.get_artist(artist_id=prov_artist_id)
695        if result is None or result.top_tracks is None:
696            return []
697        all_edges = list(result.top_tracks.edges)
698        while result.top_tracks is not None and result.top_tracks.page_info.has_next_page:
699            result = await self.provider.gql_client.get_artist(
700                artist_id=prov_artist_id,
701                top_tracks_after=result.top_tracks.page_info.end_cursor,
702            )
703            if result is None or result.top_tracks is None:
704                break
705            all_edges.extend(result.top_tracks.edges)
706        return [
707            parse_track(self.provider, edge.node) for edge in all_edges if edge.node is not None
708        ]
709
710    @use_cache(3600 * 24, allow_expired_cache=True)
711    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
712        """Retrieve a dynamic list of tracks based on the provided item."""
713        result = await self.provider.gql_client.get_similar_tracks(track_id=prov_track_id, nb=limit)
714        if result is None:
715            return []
716        return [parse_track(self.provider, t) for t in result.recommended_tracks if t is not None]
717
718    @use_cache(3600 * 24, allow_expired_cache=True)
719    async def get_similar_artists(self, prov_artist_id: str, limit: int = 25) -> list[Artist]:
720        """Retrieve a list of artists similar to the provided artist."""
721        if prov_artist_id.startswith(PERSONAL_ARTIST_PREFIX):
722            return []
723        result = await self.provider.gql_client.get_similar_artists(
724            artist_id=prov_artist_id, first=limit
725        )
726        if result is None or result.related_artist is None:
727            return []
728        return [
729            parse_artist(self.provider, edge.node)
730            for edge in result.related_artist.edges
731            if edge.node is not None
732        ]
733
734    # -- Library mutations --
735
736    async def library_add(self, item: MediaItemType) -> bool:
737        """Add an item to the provider's library/favorites."""
738        if item.media_type == MediaType.ARTIST:
739            await self.provider.gql_client.add_artist_to_favorite(artist_id=item.item_id)
740        elif item.media_type == MediaType.ALBUM:
741            await self.provider.gql_client.add_album_to_favorite(album_id=item.item_id)
742        elif item.media_type == MediaType.TRACK:
743            await self.provider.gql_client.add_track_to_favorite(track_id=item.item_id)
744        elif item.media_type == MediaType.PLAYLIST:
745            await self.provider.gql_client.add_playlist_to_favorite(playlist_id=item.item_id)
746        elif item.media_type == MediaType.PODCAST:
747            await self.provider.gql_client.add_podcast_to_favorite(podcast_id=item.item_id)
748        elif item.media_type == MediaType.AUDIOBOOK:
749            await self.provider.gql_client.add_album_to_favorite(album_id=item.item_id)
750        else:
751            raise UnsupportedFeaturedException(
752                f"Unsupported media type for library_add: {item.media_type}"
753            )
754        return True
755
756    async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
757        """Remove an item from the provider's library/favorites."""
758        if media_type == MediaType.ARTIST:
759            await self.provider.gql_client.remove_artist_from_favorite(artist_id=prov_item_id)
760        elif media_type == MediaType.ALBUM:
761            await self.provider.gql_client.remove_album_from_favorite(album_id=prov_item_id)
762        elif media_type == MediaType.TRACK:
763            await self.provider.gql_client.remove_track_from_favorite(track_id=prov_item_id)
764        elif media_type == MediaType.PLAYLIST:
765            await self.provider.gql_client.remove_playlist_from_favorite(playlist_id=prov_item_id)
766        elif media_type == MediaType.PODCAST:
767            await self.provider.gql_client.remove_podcast_from_favorite(podcast_id=prov_item_id)
768        elif media_type == MediaType.AUDIOBOOK:
769            await self.provider.gql_client.remove_album_from_favorite(album_id=prov_item_id)
770        else:
771            raise UnsupportedFeaturedException(
772                f"Unsupported media type for library_remove: {media_type}"
773            )
774        return True
775
776    # -- Playlist CRUD --
777
778    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
779        """Add track(s) to playlist."""
780        await self.provider.gql_client.add_tracks_to_playlist(
781            playlist_id=prov_playlist_id, track_ids=prov_track_ids
782        )
783        await self.provider.browse_manager.invalidate_playlist_cache(prov_playlist_id)
784
785    async def remove_playlist_tracks(
786        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
787    ) -> None:
788        """Remove track(s) from playlist."""
789        playlist_tracks = await self.provider.browse_manager.get_playlist_tracks(
790            prov_playlist_id, 0
791        )
792        track_ids = [
793            track.item_id for track in playlist_tracks if track.position in positions_to_remove
794        ]
795        if track_ids:
796            await self.provider.gql_client.remove_tracks_from_playlist(
797                playlist_id=prov_playlist_id, track_ids=track_ids
798            )
799            await self.provider.browse_manager.invalidate_playlist_cache(prov_playlist_id)
800
801    async def create_playlist(self, name: str, media_types: set[MediaType]) -> Playlist:
802        """Create a new playlist on provider with given name."""
803        result = await self.provider.gql_client.create_playlist(
804            title=name, is_private=False, is_collaborative=False
805        )
806        if result.playlist is None:
807            msg = f"Failed to create playlist '{name}' on Deezer"
808            raise MediaNotFoundError(msg)
809        playlist = await self.provider.gql_client.get_playlist(playlist_id=result.playlist.id)
810        if playlist is None:
811            msg = f"Created playlist {result.playlist.id} not found on Deezer"
812            raise MediaNotFoundError(msg)
813        return parse_playlist(self.provider, playlist, is_editable=True)
814