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