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:
280                    continue
281                seen_ids.add(raw.id)
282                if raw.favorited_at:
283                    item.date_added = parse_date(raw.favorited_at)
284                yield item
285        # 2. Audiobooks stored as favorite albums
286        audiobook_ids = await self._get_audiobook_ids_in_albums()
287        for ab_id in audiobook_ids:
288            if ab_id in seen_ids:
289                continue
290            try:
291                yield await self.get_audiobook(ab_id)
292            except MediaNotFoundError:
293                continue
294
295    # -- Search --
296
297    @use_cache(60 * 15)
298    async def search(
299        self, search_query: str, media_types: list[MediaType], limit: int = 5
300    ) -> SearchResults:
301        """Perform search on music provider."""
302        self.logger.debug("search called with media_types=%s", media_types)
303        need_albums = MediaType.ALBUM in media_types
304        need_audiobooks = MediaType.AUDIOBOOK in media_types
305
306        # Try with full limit first; on complexity error, retry with reduced limits.
307        attempts = [limit, max(limit // 2, 5), 5]
308        result = None
309        for idx, attempt_limit in enumerate(attempts):
310            try:
311                result = await self.provider.gql_client.search(
312                    query=search_query,
313                    tracks_first=attempt_limit if MediaType.TRACK in media_types else 0,
314                    albums_first=attempt_limit if (need_albums or need_audiobooks) else 0,
315                    artists_first=attempt_limit if MediaType.ARTIST in media_types else 0,
316                    playlists_first=attempt_limit if MediaType.PLAYLIST in media_types else 0,
317                    livestreams_first=attempt_limit if MediaType.RADIO in media_types else 0,
318                    podcasts_first=attempt_limit if MediaType.PODCAST in media_types else 0,
319                )
320                break
321            except GraphQLClientGraphQLMultiError as err:
322                if not _is_complexity_error(err):
323                    raise
324                if idx == len(attempts) - 1:
325                    self.logger.warning("Search complexity exceeded even at minimum limit")
326                    raise
327                self.logger.debug(
328                    "Search complexity exceeded at limit=%d, retrying with %d",
329                    attempt_limit,
330                    attempts[idx + 1],
331                )
332        search_results = SearchResults()
333        if result is None:
334            return search_results
335        if MediaType.TRACK in media_types:
336            search_results.tracks = [
337                parse_track(self.provider, edge.node)
338                for edge in result.results.tracks.edges
339                if edge.node is not None
340            ]
341        if need_albums or need_audiobooks:
342            album_nodes = [e.node for e in result.results.albums.edges if e.node is not None]
343            if album_nodes and need_audiobooks:
344                album_ids = [n.id for n in album_nodes]
345                audiobook_ids = await self.provider.gql_client.check_audiobook_ids(album_ids)
346                if need_albums:
347                    search_results.albums = [
348                        parse_album(self.provider, n)
349                        for n in album_nodes
350                        if n.id not in audiobook_ids
351                    ]
352                search_results.audiobooks = [
353                    parse_audiobook_from_album(self.provider, n)
354                    for n in album_nodes
355                    if n.id in audiobook_ids
356                ]
357            elif need_albums:
358                search_results.albums = [parse_album(self.provider, n) for n in album_nodes]
359        if MediaType.ARTIST in media_types:
360            search_results.artists = [
361                parse_artist(self.provider, edge.node)
362                for edge in result.results.artists.edges
363                if edge.node is not None
364            ]
365        if MediaType.PLAYLIST in media_types:
366            search_results.playlists = [
367                parse_playlist(self.provider, edge.node)
368                for edge in result.results.playlists.edges
369                if edge.node is not None
370            ]
371        if MediaType.RADIO in media_types:
372            search_results.radio = [
373                parse_radio(self.provider, edge.node)
374                for edge in result.results.livestreams.edges
375                if edge.node is not None
376            ]
377        if MediaType.PODCAST in media_types:
378            search_results.podcasts = [
379                parse_podcast(self.provider, edge.node)
380                for edge in result.results.podcasts.edges
381                if edge.node is not None
382            ]
383        return search_results
384
385    # -- Item getters --
386
387    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
388    async def get_artist(self, prov_artist_id: str) -> Artist:
389        """Get full artist details by id."""
390        if prov_artist_id.startswith(PERSONAL_ARTIST_PREFIX):
391            # Personal track artist — reconstruct from GW data
392            song_id = prov_artist_id.removeprefix(PERSONAL_ARTIST_PREFIX)
393            personal_songs = await self._get_personal_songs()
394            for song in personal_songs:
395                if str(song["SNG_ID"]) == song_id:
396                    return Artist(
397                        item_id=prov_artist_id,
398                        provider=self.instance_id,
399                        name=song.get("ART_NAME", ""),
400                        provider_mappings={
401                            ProviderMapping(
402                                item_id=prov_artist_id,
403                                provider_domain=self.domain,
404                                provider_instance=self.instance_id,
405                            )
406                        },
407                    )
408            raise MediaNotFoundError(f"Personal artist {prov_artist_id} not found")
409        result = await self.provider.gql_client.get_artist(artist_id=prov_artist_id)
410        if result is None:
411            raise MediaNotFoundError(f"Artist {prov_artist_id} not found on Deezer")
412        item = parse_artist(self.provider, result)
413        apply_web_url(item, result)
414        return item
415
416    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
417    async def get_album(self, prov_album_id: str) -> Album:
418        """Get full album details by id."""
419        if prov_album_id.startswith(PERSONAL_ALBUM_PREFIX):
420            # Personal track album — reconstruct from GW data
421            song_id = prov_album_id.removeprefix(PERSONAL_ALBUM_PREFIX)
422            personal_songs = await self._get_personal_songs()
423            for song in personal_songs:
424                if str(song["SNG_ID"]) == song_id:
425                    art_name = song.get("ART_NAME", "")
426                    personal_art_id = f"{PERSONAL_ARTIST_PREFIX}{song_id}"
427                    artists: UniqueList[Artist | ItemMapping] = UniqueList()
428                    if art_name:
429                        artists.append(
430                            ItemMapping(
431                                media_type=MediaType.ARTIST,
432                                item_id=personal_art_id,
433                                provider=self.instance_id,
434                                name=art_name,
435                            )
436                        )
437                    return Album(
438                        item_id=prov_album_id,
439                        provider=self.instance_id,
440                        name=song.get("ALB_TITLE", ""),
441                        artists=artists,
442                        provider_mappings={
443                            ProviderMapping(
444                                item_id=prov_album_id,
445                                provider_domain=self.domain,
446                                provider_instance=self.instance_id,
447                            )
448                        },
449                    )
450            raise MediaNotFoundError(f"Personal album {prov_album_id} not found")
451        result = await self.provider.gql_client.get_album(album_id=prov_album_id)
452        if result is None:
453            raise MediaNotFoundError(f"Album {prov_album_id} not found on Deezer")
454        item = parse_album(self.provider, result)
455        apply_web_url(item, result)
456        return item
457
458    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
459    async def get_track(self, prov_track_id: str) -> Track:
460        """Get full track details by id."""
461        try:
462            track_id_int = int(prov_track_id)
463        except ValueError as err:
464            raise MediaNotFoundError(f"Invalid Deezer track ID: {prov_track_id}") from err
465        # Personal tracks (negative IDs) don't exist in the GQL API
466        if track_id_int < 0:
467            personal_songs = await self._get_personal_songs()
468            for song in personal_songs:
469                if str(song["SNG_ID"]) == prov_track_id:
470                    return parse_gw_track(self.provider, song)
471            raise MediaNotFoundError(f"Personal track {prov_track_id} not found")
472        result = await self.provider.gql_client.get_track(track_id=prov_track_id)
473        if result is None:
474            raise MediaNotFoundError(f"Track {prov_track_id} not found on Deezer")
475        return parse_track(self.provider, result)
476
477    @use_cache(3600 * 24 * 30)
478    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
479        """Get full playlist details by id."""
480        if virtual := await self.provider.browse_manager.get_virtual_playlist(prov_playlist_id):
481            return virtual
482        result = await self.provider.gql_client.get_playlist(playlist_id=prov_playlist_id)
483        if result is None:
484            raise MediaNotFoundError(f"Playlist {prov_playlist_id} not found on Deezer")
485        is_editable = result.owner is not None and result.owner.id == self.provider.user_id
486        return parse_playlist(self.provider, result, is_editable=is_editable)
487
488    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
489    async def get_radio(self, prov_radio_id: str) -> Radio:
490        """Get full radio/livestream details by id."""
491        result = await self.provider.gql_client.get_livestream(livestream_id=prov_radio_id)
492        if result is None:
493            raise MediaNotFoundError(f"Radio {prov_radio_id} not found on Deezer")
494        return parse_radio(self.provider, result)
495
496    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
497    async def get_podcast(self, prov_podcast_id: str) -> Podcast:
498        """Get full podcast details by id."""
499        result = await self.provider.gql_client.get_podcast(podcast_id=prov_podcast_id)
500        if result is None:
501            raise MediaNotFoundError(f"Podcast {prov_podcast_id} not found on Deezer")
502        podcast = parse_podcast(self.provider, result)
503        podcast.total_episodes = len(result.raw_episodes)
504        return podcast
505
506    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
507    async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
508        """Get (full) podcast episode details by id."""
509        result = await self.provider.gql_client.get_podcast_episode(
510            podcast_episode_id=prov_episode_id,
511        )
512        if result is None:
513            raise MediaNotFoundError(f"Podcast episode {prov_episode_id} not found on Deezer")
514        podcast_mapping = ItemMapping(
515            media_type=MediaType.PODCAST,
516            item_id=result.podcast.id,
517            provider=self.instance_id,
518            name=result.podcast.display_title,
519        )
520        podcast_image_url = (
521            result.podcast.cover.urls[0]
522            if result.podcast.cover and result.podcast.cover.urls
523            else None
524        )
525        return parse_podcast_episode(self.provider, result, podcast_mapping, 0, podcast_image_url)
526
527    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
528    async def get_audiobook(self, prov_audiobook_id: str) -> Audiobook:
529        """Get full audiobook details by id."""
530        result = await self.provider.gql_client.get_audiobook(
531            audiobook_id=prov_audiobook_id, chapters_first=AUDIOBOOK_CHAPTERS_PAGE_SIZE
532        )
533        if result is None:
534            raise MediaNotFoundError(f"Audiobook {prov_audiobook_id} not found on Deezer")
535        item = parse_audiobook(self.provider, result)
536        if result.chapters.page_info.has_next_page:
537            all_edges = await fetch_all_audiobook_chapter_edges(
538                self.provider.gql_client,
539                prov_audiobook_id,
540                initial_edges=result.chapters.edges,
541                initial_page_info=result.chapters.page_info,
542            )
543        else:
544            all_edges = result.chapters.edges
545        item.metadata.chapters = parse_audiobook_chapters(all_edges)
546        return item
547
548    # -- Content getters --
549
550    @use_cache(3600 * 24 * 30, allow_expired_cache=True)
551    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
552        """Get all tracks in an album."""
553        if prov_album_id.startswith(PERSONAL_ALBUM_PREFIX):
554            # Personal album has no real Deezer album page
555            return []
556        result = await self.provider.gql_client.get_album(album_id=prov_album_id)
557        if result is None:
558            return []
559        all_edges = list(result.tracks.edges)
560        while result.tracks.page_info.has_next_page:
561            result = await self.provider.gql_client.get_album(
562                album_id=prov_album_id,
563                tracks_after=result.tracks.page_info.end_cursor,
564            )
565            if result is None:
566                break
567            all_edges.extend(result.tracks.edges)
568        return [
569            parse_track(self.provider, edge.node, position=idx)
570            for idx, edge in enumerate(all_edges, 1)
571            if edge.node is not None
572        ]
573
574    async def get_podcast_episodes(self, prov_podcast_id: str) -> AsyncGenerator[PodcastEpisode]:
575        """Get all episodes for a given podcast with current resume state."""
576        episodes = await self._fetch_podcast_episodes(prov_podcast_id)
577        if not episodes:
578            return
579        bookmarks = await fetch_all_bookmarks(self.provider.gql_client)
580        for ep in episodes:
581            ep.fully_played = False
582            ep.resume_position_ms = 0
583            if ep.item_id in bookmarks:
584                ep.fully_played, ep.resume_position_ms = bookmarks[ep.item_id]
585            yield ep
586
587    @use_cache(3600)
588    async def _fetch_podcast_episodes(self, prov_podcast_id: str) -> list[PodcastEpisode]:
589        """Fetch all episodes for a podcast (cached 1h)."""
590        # Two-layer caching strategy:
591        # - Outer (this decorator, 1h): avoids repeated cache lookups during
592        #   rapid navigation (e.g., user browsing back and forth between podcasts).
593        # - Inner (per-episode, 30 days): prevents re-fetching episode details
594        #   that rarely change. When the outer cache expires, only genuinely new
595        #   episodes require an API call.
596        result = await self.provider.gql_client.get_podcast(
597            podcast_id=prov_podcast_id, episodes_first=0
598        )
599        if result is None:
600            return []
601        podcast_mapping = ItemMapping(
602            media_type=MediaType.PODCAST,
603            item_id=result.id,
604            provider=self.instance_id,
605            name=result.display_title,
606        )
607        podcast_image_url: str | None = None
608        if result.cover and result.cover.urls:
609            podcast_image_url = result.cover.urls[0]
610        episode_ids = result.raw_episodes
611        if not episode_ids:
612            return []
613
614        cache = self.mass.cache
615        episode_cache_ttl = 3600 * 24 * 30  # 30 days
616
617        # Resolve cached vs uncached episode IDs
618        cached_episodes: dict[str, PodcastEpisode] = {}
619        uncached_ids: list[str] = []
620        for eid in episode_ids:
621            cache_key = f"podcast_episode.{eid}"
622            cached = await cache.get(cache_key, provider=self.instance_id)
623            if cached is not None:
624                cached_episodes[eid] = PodcastEpisode.from_dict(cached)
625            else:
626                uncached_ids.append(eid)
627
628        # Batch-fetch only uncached episodes
629        batch_size = 50
630        for i in range(0, len(uncached_ids), batch_size):
631            batch = uncached_ids[i : i + batch_size]
632            fetched = await self.provider.gql_client.get_podcast_episodes_by_ids(ids=batch)
633            for ep in fetched:
634                if ep is not None:
635                    parsed = parse_podcast_episode(
636                        self.provider, ep, podcast_mapping, 0, podcast_image_url
637                    )
638                    cached_episodes[ep.id] = parsed
639                    self.mass.create_task(
640                        cache.set(
641                            key=f"podcast_episode.{ep.id}",
642                            data=parsed.to_dict(),
643                            expiration=episode_cache_ttl,
644                            provider=self.instance_id,
645                        )
646                    )
647
648        # Build final list in original order with correct positions
649        episodes: list[PodcastEpisode] = []
650        position = 0
651        for eid in episode_ids:
652            if eid in cached_episodes:
653                position += 1
654                cached_ep = cached_episodes[eid]
655                cached_ep.position = position
656                episodes.append(cached_ep)
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