music-assistant-server

1"""The provider class for Open Subsonic."""
2
3from __future__ import annotations
4
5from datetime import datetime
6from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar
7
8from libopensonic import AsyncConnection as SonicConnection
9from libopensonic import Extensions as OpenSubsonicExtensions
10from libopensonic.errors import (
11    AuthError,
12    CredentialError,
13    DataNotFoundError,
14    ParameterError,
15    SonicError,
16)
17from libopensonic.media import PodcastChannel
18from music_assistant_models.config_entries import ConfigEntry
19from music_assistant_models.enums import ConfigEntryType, ContentType, MediaType, StreamType
20from music_assistant_models.errors import (
21    ActionUnavailable,
22    LoginFailed,
23    MediaNotFoundError,
24    ProviderPermissionDenied,
25    UnsupportedFeaturedException,
26)
27from music_assistant_models.media_items import (
28    Album,
29    Artist,
30    AudioFormat,
31    BrowseFolder,
32    ItemMapping,
33    MediaItemType,
34    Playlist,
35    Podcast,
36    PodcastEpisode,
37    ProviderMapping,
38    Radio,
39    RecommendationFolder,
40    SearchResults,
41    Track,
42    UniqueList,
43)
44from music_assistant_models.streamdetails import StreamDetails
45
46from music_assistant.constants import (
47    CONF_PASSWORD,
48    CONF_PATH,
49    CONF_PORT,
50    CONF_USERNAME,
51    UNKNOWN_ARTIST,
52)
53from music_assistant.controllers.cache import use_cache
54from music_assistant.models.music_provider import MusicProvider
55
56from .parsers import (
57    EP_CHAN_SEP,
58    NAVI_VARIOUS_PREFIX,
59    UNKNOWN_ARTIST_ID,
60    parse_album,
61    parse_artist,
62    parse_epsiode,
63    parse_playlist,
64    parse_podcast,
65    parse_radio,
66    parse_structured_lyrics,
67    parse_track,
68)
69
70if TYPE_CHECKING:
71    from collections.abc import AsyncGenerator
72
73    from libopensonic.media import AlbumID3 as SonicAlbum
74    from libopensonic.media import ArtistWithAlbumsID3 as SonicArtist
75    from libopensonic.media import Bookmark as SonicBookmark
76    from libopensonic.media import Child as SonicItem
77    from libopensonic.media import InternetRadioStation as SonicRadio
78    from libopensonic.media import Lyrics as SonicLyrics
79    from libopensonic.media import OpenSubsonicExtension, StructuredLyrics
80    from libopensonic.media import Playlist as SonicPlaylist
81    from libopensonic.media import PodcastEpisode as SonicEpisode
82
83CONF_BASE_URL = "baseURL"
84CONF_API_KEY = "api_key"
85CONF_ENABLE_PODCASTS = "enable_podcasts"
86CONF_ENABLE_RADIO_STATIONS = "enable_radio_stations"
87CONF_ENABLE_LEGACY_AUTH = "enable_legacy_auth"
88CONF_RECO_FAVES = "recommend_favorites"
89CONF_NEW_ALBUMS = "recommend_new"
90CONF_PLAYED_ALBUMS = "recommend_played"
91CONF_RECO_SIZE = "recommendation_count"
92CONF_PAGE_SIZE = "pagination_size"
93CONF_RAW_FILE = "request_raw_file"
94
95CACHE_CATEGORY_PODCAST_CHANNEL = 1
96CACHE_CATEGORY_PODCAST_EPISODES = 2
97
98Param = ParamSpec("Param")
99RetType = TypeVar("RetType")
100
101
102class OpenSonicProvider(MusicProvider):
103    """Provider for Open Subsonic servers."""
104
105    conn: SonicConnection
106    _enable_podcasts: bool = True
107    _enable_radio_stations: bool = True
108    _show_faves: bool = True
109    _show_new: bool = True
110    _show_played: bool = True
111    _reco_limit: int = 10
112    _pagination_size: int = 200
113    _id_lyrics: bool = False
114    _direct_podcast_episode: bool = False
115    _raw_file: bool = True
116
117    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
118        """Return Config entries to setup this provider."""
119        return (
120            ConfigEntry(
121                key=CONF_ENABLE_PODCASTS,
122                type=ConfigEntryType.BOOLEAN,
123                required=True,
124                default_value=True,
125            ),
126            ConfigEntry(
127                key=CONF_ENABLE_RADIO_STATIONS,
128                type=ConfigEntryType.BOOLEAN,
129                required=True,
130                default_value=True,
131            ),
132            ConfigEntry(
133                key=CONF_ENABLE_LEGACY_AUTH,
134                type=ConfigEntryType.BOOLEAN,
135                required=True,
136                default_value=False,
137            ),
138            ConfigEntry(
139                key=CONF_RECO_FAVES,
140                type=ConfigEntryType.BOOLEAN,
141                required=True,
142                default_value=True,
143            ),
144            ConfigEntry(
145                key=CONF_NEW_ALBUMS,
146                type=ConfigEntryType.BOOLEAN,
147                required=True,
148                default_value=True,
149            ),
150            ConfigEntry(
151                key=CONF_PLAYED_ALBUMS,
152                type=ConfigEntryType.BOOLEAN,
153                required=True,
154                default_value=True,
155            ),
156            ConfigEntry(
157                key=CONF_RECO_SIZE,
158                type=ConfigEntryType.INTEGER,
159                required=True,
160                default_value=10,
161            ),
162            ConfigEntry(
163                key=CONF_RAW_FILE,
164                type=ConfigEntryType.BOOLEAN,
165                required=False,
166                default_value=True,
167            ),
168            ConfigEntry(
169                key=CONF_PAGE_SIZE,
170                type=ConfigEntryType.INTEGER,
171                required=True,
172                default_value=200,
173                advanced=True,
174            ),
175        )
176
177    async def handle_async_init(self) -> None:
178        """Set up the music provider and test the connection."""
179        port = self.get_setup_value(CONF_PORT)
180        port = int(str(port)) if port is not None else 443
181        path = self.get_setup_value(CONF_PATH)
182
183        if path is None:
184            path = ""
185
186        api_key = self.get_setup_value(CONF_API_KEY)
187        username = self.get_setup_value(CONF_USERNAME)
188        password = self.get_setup_value(CONF_PASSWORD)
189
190        if api_key:
191            self.conn = SonicConnection(
192                str(self.get_setup_value(CONF_BASE_URL)),
193                api_key=str(api_key),
194                port=port,
195                server_path=str(path),
196                use_get=True,
197                app_name="Music Assistant",
198            )
199        elif username and password:
200            self.conn = SonicConnection(
201                str(self.get_setup_value(CONF_BASE_URL)),
202                username=str(username),
203                password=str(password),
204                legacy_auth=bool(self.config.get_value(CONF_ENABLE_LEGACY_AUTH)),
205                port=port,
206                server_path=str(path),
207                use_get=True,
208                app_name="Music Assistant",
209            )
210        else:
211            msg = f"No credentials for {self.get_setup_value(CONF_BASE_URL)}, provide an API key or username and password."
212            raise LoginFailed(
213                msg,
214                translation_key="connect_failed",
215                translation_owner=self.translation_owner,
216                translation_args=[self.get_setup_value(CONF_BASE_URL)],
217            )
218
219        try:
220            success = await self.conn.ping()
221            if not success:
222                raise CredentialError
223        except (AuthError, CredentialError) as e:
224            msg = (
225                f"Failed to connect to {self.get_setup_value(CONF_BASE_URL)}, check your settings."
226            )
227            raise LoginFailed(
228                msg,
229                translation_key="connect_failed",
230                translation_owner=self.translation_owner,
231                translation_args=[self.get_setup_value(CONF_BASE_URL)],
232            ) from e
233
234        try:
235            extensions: list[OpenSubsonicExtension] = await self.conn.get_open_subsonic_extensions()
236            for entry in extensions:
237                if entry.name == OpenSubsonicExtensions.STRUCTURED_LYRICS:
238                    self._id_lyrics = True
239                elif entry.name == OpenSubsonicExtensions.GET_PODCAST_EPISODE:
240                    self._direct_podcast_episode = True
241        except OSError:
242            self.logger.info("Failed to query server for OpenSubsonic extensions")
243
244        self._enable_podcasts = bool(self.config.get_value(CONF_ENABLE_PODCASTS))
245        self._enable_radio_stations = bool(self.config.get_value(CONF_ENABLE_RADIO_STATIONS, True))
246        self._show_faves = bool(self.config.get_value(CONF_RECO_FAVES))
247        self._show_new = bool(self.config.get_value(CONF_NEW_ALBUMS))
248        self._show_played = bool(self.config.get_value(CONF_PLAYED_ALBUMS))
249        self._reco_limit = int(str(self.config.get_value(CONF_RECO_SIZE)))
250        self._pagination_size = int(str(self.config.get_value(CONF_PAGE_SIZE)))
251        self._pagination_size = min(self._pagination_size, 500)
252        self._raw_file = bool(self.config.get_value(CONF_RAW_FILE))
253
254    async def unload(self, is_removed: bool = False) -> None:
255        """Unload the provider."""
256        await super().unload(is_removed)
257        await self.conn.cleanup()
258
259    @property
260    def is_streaming_provider(self) -> bool:
261        """
262        Return True if the provider is a streaming provider.
263
264        This literally means that the catalog is not the same as the library contents.
265        For local based providers (files, plex), the catalog is the same as the library content.
266        It also means that data is if this provider is NOT a streaming provider,
267        data cross instances is unique, the catalog and library differs per instance.
268
269        Setting this to True will only query one instance of the provider for search and lookups.
270        Setting this to False will query all instances of this provider for search and lookups.
271        """
272        return False
273
274    async def get_recommendations(self) -> list[RecommendationFolder]:
275        """
276        Get this provider's available recommendation rows, without items.
277
278        These can be favorited items, recently added albums, newest podcast episodes,
279        and most played albums.  What is included is configured with the provider.
280        """
281        recos: list[RecommendationFolder] = []
282        if self._enable_podcasts:
283            recos.append(
284                RecommendationFolder(
285                    item_id="subsonic_newest_podcasts",
286                    provider=self.instance_id,
287                    name="Newest Podcast Episodes",
288                    translation_key="episodes_recently_added",
289                )
290            )
291        if self._show_faves:
292            recos.append(
293                RecommendationFolder(
294                    item_id="subsonic_starred_albums",
295                    provider=self.instance_id,
296                    name="Starred Items",
297                    translation_key="starred_items",
298                )
299            )
300        if self._show_new:
301            recos.append(
302                RecommendationFolder(
303                    item_id="subsonic_new_albums",
304                    provider=self.instance_id,
305                    name="New Albums",
306                    translation_key="recently_added_albums",
307                )
308            )
309        if self._show_played:
310            recos.append(
311                RecommendationFolder(
312                    item_id="subsonic_most_played",
313                    provider=self.instance_id,
314                    name="Most Played Albums",
315                    translation_key="most_played_albums",
316                )
317            )
318        return recos
319
320    async def get_recommendation_items(
321        self, item_id: str
322    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
323        """
324        Get the items for a single recommendation row.
325
326        :param item_id: The item_id of the row, as returned by get_recommendations.
327        """
328        folder: RecommendationFolder | None = None
329        if item_id == "subsonic_newest_podcasts" and self._enable_podcasts:
330            folder = await self._podcast_recommendations()
331        elif item_id == "subsonic_starred_albums" and self._show_faves:
332            folder = await self._favorites_recommendation()
333        elif item_id == "subsonic_new_albums" and self._show_new:
334            folder = await self._new_recommendations()
335        elif item_id == "subsonic_most_played" and self._show_played:
336            folder = await self._played_recommendations()
337        if folder is None:
338            return UniqueList()
339        return folder.items
340
341    async def resolve_image(self, path: str) -> bytes | Any:
342        """Return the image."""
343        self.logger.debug("Requesting cover art for '%s'", path)
344
345        try:
346            art = await self.conn.get_cover_art(path)
347            return await art.content.read()
348        except DataNotFoundError:
349            self.logger.warning("Unable to locate a cover image for %s", path)
350            return None
351
352    @use_cache(3600 * 3)  # cache for 3 hours
353    async def search(
354        self, search_query: str, media_types: list[MediaType], limit: int = 20
355    ) -> SearchResults:
356        """Search the sonic library."""
357        artists = limit if MediaType.ARTIST in media_types else 0
358        albums = limit if MediaType.ALBUM in media_types else 0
359        songs = limit if MediaType.TRACK in media_types else 0
360        if not (artists or albums or songs):
361            return SearchResults()
362        answer = await self.conn.search3(
363            query=search_query,
364            artist_count=artists,
365            artist_offset=0,
366            album_count=albums,
367            album_offset=0,
368            song_count=songs,
369            song_offset=0,
370        )
371
372        if answer.artist:
373            ar = [
374                parse_artist(self.instance_id, entry, logger=self.logger) for entry in answer.artist
375            ]
376        else:
377            ar = []
378
379        if answer.album:
380            al = [parse_album(self.logger, self.instance_id, entry) for entry in answer.album]
381        else:
382            al = []
383
384        if answer.song:
385            tr = []
386            for entry in answer.song:
387                self._set_loudness(entry)
388                lyrics: tuple[str, bool] | None = await self.get_track_lyrics(entry)
389                tr.append(parse_track(self.logger, self.instance_id, entry, lyrics=lyrics))
390        else:
391            tr = []
392
393        return SearchResults(artists=ar, albums=al, tracks=tr)
394
395    async def set_favorite(self, prov_item_id: str, media_type: MediaType, favorite: bool) -> None:
396        """Set or clear favorite on the server."""
397        # The subsonic spec does not support favorite-ing anything but artists, albums, and tracks
398        if media_type not in (MediaType.ARTIST, MediaType.ALBUM, MediaType.TRACK):
399            return
400
401        track_ids: list[str] = []
402        album_ids: list[str] = []
403        artist_ids: list[str] = []
404
405        if media_type == MediaType.ARTIST:
406            artist_ids.append(prov_item_id)
407        elif media_type == MediaType.ALBUM:
408            album_ids.append(prov_item_id)
409        elif media_type == MediaType.TRACK:
410            track_ids.append(prov_item_id)
411
412        if favorite:
413            await self.conn.star(sids=track_ids, album_ids=album_ids, artist_ids=artist_ids)
414        else:
415            await self.conn.unstar(sids=track_ids, album_ids=album_ids, artist_ids=artist_ids)
416
417    async def get_library_artists(self) -> AsyncGenerator[Artist]:
418        """Provide a generator for reading all artists."""
419        artists = await self.conn.get_artists()
420
421        if not artists.index:
422            return
423
424        for index in artists.index:
425            if not index.artist:
426                continue
427
428            for artist in index.artist:
429                yield parse_artist(self.instance_id, artist, logger=self.logger)
430
431    async def get_library_albums(self) -> AsyncGenerator[Album]:
432        """
433        Provide a generator for reading all artists.
434
435        Note the pagination, the open subsonic docs say that this method is limited to
436        returning 500 items per invocation.
437        """
438        offset = 0
439        size = self._pagination_size
440        albums = await self.conn.get_album_list2(
441            ltype="alphabeticalByArtist",
442            size=size,
443            offset=offset,
444        )
445        while albums:
446            for album in albums:
447                yield parse_album(self.logger, self.instance_id, album)
448            offset += size
449            albums = await self.conn.get_album_list2(
450                ltype="alphabeticalByArtist",
451                size=size,
452                offset=offset,
453            )
454
455    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
456        """Provide a generator for library playlists."""
457        results = await self.conn.get_playlists()
458        for entry in results:
459            yield parse_playlist(self.instance_id, entry)
460
461    async def get_library_radios(self) -> AsyncGenerator[Radio]:
462        """Provide a generator for library radio stations."""
463        if not self._enable_radio_stations:
464            return
465        stations: list[SonicRadio] = await self.conn.get_internet_radio_stations()
466        for entry in stations:
467            yield parse_radio(self.instance_id, entry)
468
469    async def get_radio(self, prov_radio_id: str) -> Radio:
470        """Return the requested radio station."""
471        async for station in self.get_library_radios():
472            if station.item_id == prov_radio_id:
473                return station
474        msg = f"Radio {prov_radio_id} not found"
475        raise MediaNotFoundError(msg)
476
477    async def get_library_tracks(self) -> AsyncGenerator[Track]:
478        """
479        Provide a generator for library tracks.
480
481        Note the lack of item count on this method.
482        """
483        query = ""
484        offset = 0
485        count = self._pagination_size
486        try:
487            results = await self.conn.search3(
488                query=query,
489                artist_count=0,
490                album_count=0,
491                song_offset=offset,
492                song_count=count,
493            )
494        except ParameterError:
495            # Older Navidrome does not accept an empty string and requires the empty quotes
496            query = '""'
497            results = await self.conn.search3(
498                query=query,
499                artist_count=0,
500                album_count=0,
501                song_offset=offset,
502                song_count=count,
503            )
504        while results.song:
505            album: Album | None = None
506            for entry in results.song:
507                aid = entry.album_id or entry.parent
508                if aid is not None and (album is None or album.item_id != aid):
509                    album = await self.get_album(prov_album_id=aid)
510                self._set_loudness(entry)
511                lyrics: tuple[str, bool] | None = await self.get_track_lyrics(entry)
512                yield parse_track(self.logger, self.instance_id, entry, album=album, lyrics=lyrics)
513            offset += count
514            results = await self.conn.search3(
515                query=query,
516                artist_count=0,
517                album_count=0,
518                song_offset=offset,
519                song_count=count,
520            )
521
522    @use_cache(3600 * 3)  # cache for 3 hours
523    async def get_album(self, prov_album_id: str) -> Album:
524        """Return the requested Album."""
525        try:
526            sonic_album: SonicAlbum = await self.conn.get_album(prov_album_id)
527            sonic_info = await self.conn.get_album_info2(aid=prov_album_id)
528        except (ParameterError, DataNotFoundError) as e:
529            msg = f"Album {prov_album_id} not found"
530            raise MediaNotFoundError(msg) from e
531
532        return parse_album(self.logger, self.instance_id, sonic_album, sonic_info)
533
534    @use_cache(3600 * 3)  # cache for 3 hours
535    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
536        """Return a list of tracks on the specified Album."""
537        try:
538            sonic_album: SonicAlbum = await self.conn.get_album(prov_album_id)
539        except (ParameterError, DataNotFoundError) as e:
540            msg = f"Album {prov_album_id} not found"
541            raise MediaNotFoundError(msg) from e
542        tracks = []
543        if sonic_album.song:
544            for sonic_song in sonic_album.song:
545                self._set_loudness(sonic_song)
546                lyrics: tuple[str, bool] | None = await self.get_track_lyrics(sonic_song)
547                tracks.append(parse_track(self.logger, self.instance_id, sonic_song, lyrics=lyrics))
548        return tracks
549
550    @use_cache(3600 * 3)  # cache for 3 hours
551    async def get_artist(self, prov_artist_id: str) -> Artist:
552        """Return the requested Artist."""
553        if prov_artist_id == UNKNOWN_ARTIST_ID:
554            return Artist(
555                item_id=UNKNOWN_ARTIST_ID,
556                name=UNKNOWN_ARTIST,
557                provider=self.instance_id,
558                provider_mappings={
559                    ProviderMapping(
560                        item_id=UNKNOWN_ARTIST_ID,
561                        provider_domain=self.domain,
562                        provider_instance=self.instance_id,
563                    )
564                },
565            )
566        if prov_artist_id.startswith(NAVI_VARIOUS_PREFIX):
567            # Special case for handling track artists on various artists album for Navidrome.
568            return Artist(
569                item_id=prov_artist_id,
570                name=prov_artist_id.removeprefix(NAVI_VARIOUS_PREFIX),
571                provider=self.instance_id,
572                provider_mappings={
573                    ProviderMapping(
574                        item_id=prov_artist_id,
575                        provider_domain=self.domain,
576                        provider_instance=self.instance_id,
577                    )
578                },
579            )
580
581        try:
582            sonic_artist: SonicArtist = await self.conn.get_artist(artist_id=prov_artist_id)
583            sonic_info = await self.conn.get_artist_info2(aid=prov_artist_id)
584        except (ParameterError, DataNotFoundError) as e:
585            msg = f"Artist {prov_artist_id} not found"
586            raise MediaNotFoundError(msg) from e
587        return parse_artist(self.instance_id, sonic_artist, sonic_info, logger=self.logger)
588
589    @use_cache(3600 * 3)  # cache for 3 hours
590    async def get_track(self, prov_track_id: str) -> Track:
591        """Return the specified track."""
592        try:
593            sonic_song: SonicItem = await self.conn.get_song(prov_track_id)
594        except (ParameterError, DataNotFoundError) as e:
595            msg = f"Item {prov_track_id} not found"
596            raise MediaNotFoundError(msg) from e
597        aid = sonic_song.album_id or sonic_song.parent
598        album: Album | None = None
599        if not aid:
600            self.logger.warning("Unable to find album id for track %s", sonic_song.id)
601        else:
602            album = await self.get_album(prov_album_id=aid)
603        self._set_loudness(sonic_song)
604        lyrics: tuple[str, bool] | None = await self.get_track_lyrics(sonic_song)
605        return parse_track(self.logger, self.instance_id, sonic_song, album=album, lyrics=lyrics)
606
607    @use_cache(3600 * 3)  # cache for 3 hours
608    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
609        """Return a list of all Albums by specified Artist."""
610        if prov_artist_id == UNKNOWN_ARTIST_ID or prov_artist_id.startswith(NAVI_VARIOUS_PREFIX):
611            return []
612
613        try:
614            sonic_artist: SonicArtist = await self.conn.get_artist(prov_artist_id)
615        except (ParameterError, DataNotFoundError) as e:
616            msg = f"Album {prov_artist_id} not found"
617            raise MediaNotFoundError(msg) from e
618        albums = []
619        if sonic_artist.album:
620            for entry in sonic_artist.album:
621                albums.append(parse_album(self.logger, self.instance_id, entry))
622        return albums
623
624    @use_cache(3600 * 3)  # cache for 3 hours
625    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
626        """Return the specified Playlist."""
627        try:
628            sonic_playlist: SonicPlaylist = await self.conn.get_playlist(prov_playlist_id)
629        except (ParameterError, DataNotFoundError) as e:
630            msg = f"Playlist {prov_playlist_id} not found"
631            raise MediaNotFoundError(msg) from e
632        return parse_playlist(self.instance_id, sonic_playlist)
633
634    @use_cache(3600 * 3)  # cache for 3 hours
635    async def get_podcast_episode(self, prov_episode_id: str) -> PodcastEpisode:
636        """Get (full) podcast episode details by id."""
637        podcast_id, _ = prov_episode_id.split(EP_CHAN_SEP)
638        async for episode in self.get_podcast_episodes(podcast_id):
639            if episode.item_id == prov_episode_id:
640                return episode
641        msg = f"Episode {prov_episode_id} not found"
642        raise MediaNotFoundError(msg)
643
644    async def get_podcast_episodes(
645        self,
646        prov_podcast_id: str,
647    ) -> AsyncGenerator[PodcastEpisode]:
648        """Get all Episodes for given podcast id."""
649        if not self._enable_podcasts:
650            return
651        channels = await self.conn.get_podcasts(inc_episodes=True, pid=prov_podcast_id)
652        channel = channels[0]
653        if not channel.episode:
654            return
655
656        for episode in channel.episode:
657            self._set_loudness(episode)
658            yield parse_epsiode(self.instance_id, episode, channel)
659
660    @use_cache(3600 * 3)  # cache for 3 hours
661    async def get_podcast(self, prov_podcast_id: str) -> Podcast:
662        """Get full Podcast details by id."""
663        if not self._enable_podcasts:
664            msg = "Podcasts are currently disabled in the provider configuration"
665            raise ActionUnavailable(msg)
666
667        channels = await self.conn.get_podcasts(inc_episodes=True, pid=prov_podcast_id)
668
669        return parse_podcast(self.instance_id, channels[0])
670
671    async def get_library_podcasts(self) -> AsyncGenerator[Podcast]:
672        """Retrieve library/subscribed podcasts from the provider."""
673        if self._enable_podcasts:
674            channels = await self.conn.get_podcasts(inc_episodes=True)
675
676            for channel in channels:
677                yield parse_podcast(self.instance_id, channel)
678
679    @use_cache(3600 * 3)  # cache for 3 hours
680    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
681        """Get playlist tracks."""
682        result: list[Track] = []
683        if page > 0:
684            # paging not supported, we always return the whole list at once
685            return result
686        try:
687            sonic_playlist: SonicPlaylist = await self.conn.get_playlist(prov_playlist_id)
688        except (ParameterError, DataNotFoundError) as e:
689            msg = f"Playlist {prov_playlist_id} not found"
690            raise MediaNotFoundError(msg) from e
691
692        if not sonic_playlist.entry:
693            return result
694
695        for index, sonic_song in enumerate(sonic_playlist.entry, 1):
696            # A playlist can hold thousands of tracks, so we must not trigger a per-track
697            # metadata fetch here: parse_track derives the album reference from the playlist
698            # entry itself, and lyrics are fetched on demand when a track is played (get_track).
699            # Fetching album + lyrics per entry turned a single getPlaylist call into thousands
700            # of serial requests, making large playlists take minutes to start.
701            self._set_loudness(sonic_song)
702            track = parse_track(self.logger, self.instance_id, sonic_song)
703            track.position = index
704            result.append(track)
705        return result
706
707    @use_cache(3600 * 3)  # cache for 3 hours
708    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
709        """Get the top listed tracks for a specified artist."""
710        # We have seen top tracks requested for the UNKNOWN_ARTIST ID, protect against that
711        if prov_artist_id == UNKNOWN_ARTIST_ID or prov_artist_id.startswith(NAVI_VARIOUS_PREFIX):
712            return []
713
714        try:
715            sonic_artist: SonicArtist = await self.conn.get_artist(prov_artist_id)
716        except DataNotFoundError as e:
717            msg = f"Artist {prov_artist_id} not found"
718            raise MediaNotFoundError(msg) from e
719        songs: list[SonicItem] = await self.conn.get_top_songs(sonic_artist.name)
720        tracks = []
721        for entry in songs:
722            self._set_loudness(entry)
723            lyrics: tuple[str, bool] | None = await self.get_track_lyrics(entry)
724            tracks.append(parse_track(self.logger, self.instance_id, entry, lyrics=lyrics))
725        return tracks
726
727    @use_cache(3600 * 3)  # cache for 3 hours
728    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
729        """Get tracks similar to selected track."""
730        try:
731            songs: list[SonicItem] = await self.conn.get_similar_songs(
732                iid=prov_track_id, count=limit
733            )
734        except DataNotFoundError as e:
735            # Subsonic returns an error here instead of an empty list, I don't think this
736            # should be an exception but there we are. Return an empty list because this
737            # exception means we didn't find anything similar.
738            self.logger.info(e)
739            return []
740        tracks = []
741        for entry in songs:
742            self._set_loudness(entry)
743            lyrics: tuple[str, bool] | None = await self.get_track_lyrics(entry)
744            tracks.append(parse_track(self.logger, self.instance_id, entry, lyrics=lyrics))
745        return tracks
746
747    async def create_playlist(self, name: str, media_types: set[MediaType]) -> Playlist:
748        """Create a new empty playlist on the server."""
749        if not await self.conn.create_playlist(name=name):
750            raise ProviderPermissionDenied(
751                "Please ensure you have permission to create playlists on your server"
752            )
753        pls: list[SonicPlaylist] = await self.conn.get_playlists()
754        for pl in pls:
755            if pl.name == name:
756                return parse_playlist(self.instance_id, pl)
757        raise MediaNotFoundError(
758            f"Failed to create playlist with name '{name}'",
759            translation_key="create_playlist_failed",
760            translation_owner=self.translation_owner,
761            translation_args=[name],
762        )
763
764    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
765        """
766        Append the listed tracks to the selected playlist.
767
768        Note that the configured user must own the playlist to edit this way.
769        """
770        try:
771            await self.conn.update_playlist(
772                lid=prov_playlist_id,
773                song_ids_to_add=prov_track_ids,
774            )
775        except SonicError as ex:
776            msg = f"Failed to add songs to {prov_playlist_id}, check your permissions."
777            raise ProviderPermissionDenied(msg) from ex
778
779    async def remove_playlist_tracks(
780        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
781    ) -> None:
782        """Remove selected positions from the playlist."""
783        idx_to_remove = [pos - 1 for pos in positions_to_remove]
784        try:
785            await self.conn.update_playlist(
786                lid=prov_playlist_id,
787                song_indices_to_remove=idx_to_remove,
788            )
789        except SonicError as ex:
790            msg = f"Failed to remove songs from {prov_playlist_id}, check your permissions."
791            raise ProviderPermissionDenied(msg) from ex
792
793    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
794        """Get the details needed to process a specified track."""
795        item: SonicItem | SonicEpisode
796        if media_type == MediaType.TRACK:
797            try:
798                item = await self.conn.get_song(item_id)
799            except (ParameterError, DataNotFoundError) as e:
800                msg = f"Item {item_id} not found"
801                raise MediaNotFoundError(msg) from e
802
803            mime_type = item.transcoded_content_type or item.content_type
804
805            self.logger.debug(
806                "Fetching stream details for id %s '%s' with format '%s'",
807                item.id,
808                item.title,
809                mime_type,
810            )
811
812        elif media_type == MediaType.PODCAST_EPISODE:
813            item = await self._get_podcast_episode(item_id)
814
815            mime_type = item.transcoded_content_type or item.content_type
816
817            self.logger.debug(
818                "Fetching stream details for podcast episode '%s' with format '%s'",
819                item.id,
820                item.content_type,
821            )
822        elif media_type == MediaType.RADIO:
823            async for station in self.get_library_radios():
824                if station.item_id == item_id:
825                    return StreamDetails(
826                        item_id=item_id,
827                        provider=self.instance_id,
828                        allow_seek=False,
829                        can_seek=False,
830                        media_type=MediaType.RADIO,
831                        audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
832                        stream_type=StreamType.HTTP,
833                        path=station.uri or "",
834                    )
835            msg = f"Radio {item_id} not found"
836            raise MediaNotFoundError(msg)
837        else:
838            msg = f"Unsupported media type encountered '{media_type}'"
839            raise UnsupportedFeaturedException(msg)
840
841        fmat = "raw" if self._raw_file else None
842        url, _ = self.conn.get_stream_url(item.id, tformat=fmat, estimate_length=True)
843
844        return StreamDetails(
845            item_id=item.id,
846            provider=self.instance_id,
847            allow_seek=True,
848            can_seek=True,
849            media_type=media_type,
850            audio_format=AudioFormat(
851                content_type=ContentType.try_parse(mime_type),
852                sample_rate=item.sampling_rate or 44100,
853                bit_depth=item.bit_depth or 16,
854                channels=item.channel_count or 2,
855            ),
856            stream_type=StreamType.HTTP,
857            path=url,
858            duration=item.duration or 0,
859        )
860
861    async def on_played(
862        self,
863        media_type: MediaType,
864        prov_item_id: str,
865        fully_played: bool,
866        position: int,
867        media_item: MediaItemType,
868        is_playing: bool = False,
869    ) -> None:
870        """
871        Handle callback when a (playable) media item has been played.
872
873        This is called by the Queue controller when;
874            - a track has been fully played
875            - a track has been stopped (or skipped) after being played
876            - every 30s when a track is playing
877
878        Fully played is True when the track has been played to the end.
879
880        Position is the last known position of the track in seconds, to sync resume state.
881        When fully_played is set to false and position is 0,
882        the user marked the item as unplayed in the UI.
883
884        is_playing is True when the track is currently playing.
885
886        media_item is the full media item details of the played/playing track.
887        """
888        if media_type != MediaType.PODCAST_EPISODE:
889            # We don't handle audio books in this provider so this is the only resummable media
890            # type we should see.
891            return
892
893        _, ep_id = prov_item_id.split(EP_CHAN_SEP)
894
895        if fully_played:
896            # We completed the episode and should delete our bookmark
897            try:
898                await self.conn.delete_bookmark(mid=ep_id)
899            except DataNotFoundError:
900                # We probably raced with something else deleting this bookmark, not really a problem
901                self.logger.info("Bookmark for item '%s' has already been deleted.", ep_id)
902            return
903
904        # Otherwise, create a new bookmark for this item or update the existing one
905        # MA provides a position in seconds but expects it back in milliseconds
906        await self.conn.create_bookmark(
907            mid=ep_id,
908            position=position * 1000,
909            comment="Music Assistant Bookmark",
910        )
911
912    async def get_resume_position(
913        self, item_id: str, media_type: MediaType
914    ) -> tuple[bool, int, datetime | None]:
915        """
916        Get progress (resume point) details for the given Audiobook or Podcast episode.
917
918        This is a separate call from the regular get_item call to ensure the resume position
919        is always up-to-date and because a lot providers have this info present on a dedicated
920        endpoint.
921
922        Will be called right before playback starts to ensure the resume position is correct.
923
924        Returns a boolean with the fully_played status
925        and an integer with the resume position in ms.
926        """
927        if media_type != MediaType.PODCAST_EPISODE:
928            raise NotImplementedError("AudioBooks are not supported by the Open Subsonic provider")
929
930        _, ep_id = item_id.split(EP_CHAN_SEP)
931
932        bookmarks: list[SonicBookmark] = await self.conn.get_bookmarks()
933
934        for mark in bookmarks:
935            if mark.entry.id == ep_id:
936                return (
937                    False,
938                    mark.position,
939                    datetime.fromisoformat(mark.created) if mark.created else None,
940                )
941        # If we get here, there is no bookmark
942        return (False, 0, None)
943
944    async def get_track_lyrics(self, track: SonicItem) -> tuple[str, bool] | None:
945        """
946        Get lyrics for a track.
947
948        Fetches lyrics from Subsonic server. Returns the lyrics text in LRC format
949        if the Lyrics are synced (have time stamp info) or raw text if not
950        """
951        # Server doesn't support to newer lyrics retrieval, fall back to the old one
952        if not self._id_lyrics:
953            try:
954                ly: SonicLyrics = await self.conn.get_lyrics(track.title, track.artist)
955            except DataNotFoundError:
956                self.logger.debug("Lyrics not found for '%s' by '%s'", track.title, track.artist)
957                return None
958            return (ly.value, False)
959
960        try:
961            lyrics: list[StructuredLyrics] = await self.conn.get_lyrics_by_song_id(track.id)
962        except DataNotFoundError:
963            self.logger.debug("Lyrics not found for '%s'", track.id)
964            return None
965        if not lyrics:
966            return None
967        return parse_structured_lyrics(lyrics[0])
968
969    async def _get_podcast_episode(self, eid: str) -> SonicEpisode:
970        chan_id, ep_id = eid.split(EP_CHAN_SEP)
971
972        if self._direct_podcast_episode:
973            try:
974                return await self.conn.get_podcast_episode(ep_id)
975            except DataNotFoundError as e:
976                msg = f"Can't find episode {ep_id} in podcast {chan_id}"
977                raise MediaNotFoundError(msg) from e
978
979        chan = await self.conn.get_podcasts(inc_episodes=True, pid=chan_id)
980
981        if not chan[0].episode:
982            raise MediaNotFoundError(f"Missing episode list for podcast channel '{chan[0].id}'")
983
984        for episode in chan[0].episode:
985            if episode.id == ep_id:
986                return episode
987
988        msg = f"Can't find episode {ep_id} in podcast {chan_id}"
989        raise MediaNotFoundError(msg)
990
991    def _set_loudness(self, item: SonicItem) -> None:
992        if item.replay_gain and item.replay_gain.track_gain is not None:
993            # Convert ReplayGain values (gain in dB) to integrated loudness (LUFS)
994            track_loudness = -18 - item.replay_gain.track_gain
995            album_loudness = (
996                -18 - item.replay_gain.album_gain
997                if item.replay_gain.album_gain is not None
998                else None
999            )
1000            self.mass.create_task(
1001                self.mass.streams.audio_analysis.set_track_loudness(
1002                    item.id,
1003                    self.instance_id,
1004                    track_loudness,
1005                    album_loudness,
1006                )
1007            )
1008
1009    async def _get_podcast_channel_async(self, chan_id: str) -> PodcastChannel | None:
1010        if cache := await self.mass.cache.get(
1011            key=chan_id,
1012            provider=self.instance_id,
1013            category=CACHE_CATEGORY_PODCAST_CHANNEL,
1014            base_class=PodcastChannel,
1015        ):
1016            return cache
1017        if channels := await self.conn.get_podcasts(inc_episodes=True, pid=chan_id):
1018            channel = channels[0]
1019            await self.mass.cache.set(
1020                key=chan_id,
1021                data=channel.to_dict(),
1022                provider=self.instance_id,
1023                expiration=600,
1024                category=CACHE_CATEGORY_PODCAST_CHANNEL,
1025            )
1026            return channel
1027        return None
1028
1029    @use_cache(3600 * 3, cache_checksum="v2", base_class=RecommendationFolder)
1030    async def _podcast_recommendations(self) -> RecommendationFolder:
1031        podcasts: RecommendationFolder = RecommendationFolder(
1032            item_id="subsonic_newest_podcasts",
1033            provider=self.instance_id,
1034            name="Newest Podcast Episodes",
1035            translation_key="episodes_recently_added",
1036        )
1037        sonic_episodes = await self.conn.get_newest_podcasts(count=self._reco_limit)
1038        for ep in sonic_episodes:
1039            if channel_info := await self._get_podcast_channel_async(ep.channel_id):
1040                self._set_loudness(ep)
1041                podcasts.items.append(parse_epsiode(self.instance_id, ep, channel_info))
1042        return podcasts
1043
1044    @use_cache(3600 * 3, cache_checksum="v2", base_class=RecommendationFolder)
1045    async def _favorites_recommendation(self) -> RecommendationFolder:
1046        faves: RecommendationFolder = RecommendationFolder(
1047            item_id="subsonic_starred_albums",
1048            provider=self.instance_id,
1049            name="Starred Items",
1050            translation_key="starred_items",
1051        )
1052        starred = await self.conn.get_starred2()
1053        if starred.album:
1054            for sonic_album in starred.album[: self._reco_limit]:
1055                faves.items.append(parse_album(self.logger, self.instance_id, sonic_album))
1056        if starred.artist:
1057            for sonic_artist in starred.artist[: self._reco_limit]:
1058                faves.items.append(parse_artist(self.instance_id, sonic_artist, logger=self.logger))
1059        if starred.song:
1060            for sonic_song in starred.song[: self._reco_limit]:
1061                self._set_loudness(sonic_song)
1062                lyrics: tuple[str, bool] | None = await self.get_track_lyrics(sonic_song)
1063                faves.items.append(
1064                    parse_track(self.logger, self.instance_id, sonic_song, lyrics=lyrics)
1065                )
1066        return faves
1067
1068    @use_cache(3600 * 3, cache_checksum="v2", base_class=RecommendationFolder)
1069    async def _new_recommendations(self) -> RecommendationFolder:
1070        new_stuff: RecommendationFolder = RecommendationFolder(
1071            item_id="subsonic_new_albums",
1072            provider=self.instance_id,
1073            name="New Albums",
1074            translation_key="recently_added_albums",
1075        )
1076        new_albums = await self.conn.get_album_list2(ltype="newest", size=self._reco_limit)
1077        for sonic_album in new_albums:
1078            new_stuff.items.append(parse_album(self.logger, self.instance_id, sonic_album))
1079        return new_stuff
1080
1081    @use_cache(3600 * 3, cache_checksum="v2", base_class=RecommendationFolder)
1082    async def _played_recommendations(self) -> RecommendationFolder:
1083        recent: RecommendationFolder = RecommendationFolder(
1084            item_id="subsonic_most_played",
1085            provider=self.instance_id,
1086            name="Most Played Albums",
1087            translation_key="most_played_albums",
1088        )
1089        albums = await self.conn.get_album_list2(ltype="frequent", size=self._reco_limit)
1090        for sonic_album in albums:
1091            recent.items.append(parse_album(self.logger, self.instance_id, sonic_album))
1092        return recent
1093