music-assistant-server

1"""Apple Music provider implementation."""
2
3from __future__ import annotations
4
5from collections.abc import Sequence
6from typing import TYPE_CHECKING, cast
7
8from music_assistant_models.media_items import (
9    Album,
10    Artist,
11    BrowseFolder,
12    ItemMapping,
13    MediaItemType,
14    Playlist,
15    RecommendationFolder,
16    SearchResults,
17    Track,
18    UniqueList,
19)
20
21from music_assistant.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER
22from music_assistant.models.music_provider import MusicProvider
23from music_assistant.models.recommendation_payload import RecommendationPayloadMixin
24
25from .api_client import AppleMusicAPIClient
26from .constants import (
27    CONF_MUSIC_APP_TOKEN,
28    CONF_MUSIC_USER_MANUAL_TOKEN,
29    CONF_MUSIC_USER_TOKEN,
30    MUSIC_APP_TOKEN,
31    SUPPORTED_FEATURES,
32)
33from .helpers import browse_playlists
34from .library import AppleMusicLibraryManager
35from .media import AppleMusicMediaManager
36from .recommendations import AppleMusicRecommendationManager
37from .streaming import AppleMusicStreamingManager
38
39if TYPE_CHECKING:
40    from collections.abc import AsyncGenerator
41
42    from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
43    from music_assistant_models.enums import MediaType
44    from music_assistant_models.provider import ProviderManifest
45    from music_assistant_models.streamdetails import StreamDetails
46
47    from music_assistant import MusicAssistant
48
49
50class AppleMusicProvider(RecommendationPayloadMixin, MusicProvider):
51    """Implementation of an Apple Music MusicProvider."""
52
53    _music_user_token: str | None = None
54    _music_app_token: str | None = None
55    _storefront: str | None = None
56
57    def __init__(
58        self,
59        mass: MusicAssistant,
60        manifest: ProviderManifest,
61        config: ProviderConfig,
62    ) -> None:
63        """Initialize Apple Music provider."""
64        super().__init__(mass, manifest, config, SUPPORTED_FEATURES)
65        self.api_client = AppleMusicAPIClient(self)
66        self.library_manager = AppleMusicLibraryManager(self)
67        self.media_manager = AppleMusicMediaManager(self)
68        self.recommendation_manager = AppleMusicRecommendationManager(self)
69        self.streaming_manager = AppleMusicStreamingManager(self)
70
71    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
72        """
73        Return Config entries to configure this provider.
74
75        Authentication is handled by the setup flow (see setup_flow.py); only the
76        informational note is surfaced here.
77        """
78        return (CONF_ENTRY_UNOFFICIAL_PROVIDER,)
79
80    async def handle_async_init(self) -> None:
81        """Handle async initialization of the provider."""
82        self._music_user_token = cast(
83            "str | None",
84            self.get_setup_value(CONF_MUSIC_USER_MANUAL_TOKEN)
85            or self.get_setup_value(CONF_MUSIC_USER_TOKEN),
86        )
87        # a stored/manual app token only exists because the setup flow found the
88        # bundled one empty or invalid, so it takes precedence over the bundled token
89        self._music_app_token = (
90            cast("str | None", self.get_setup_value(CONF_MUSIC_APP_TOKEN)) or MUSIC_APP_TOKEN
91        )
92        self._storefront = await self.api_client.get_user_storefront()
93        await self.streaming_manager.initialize()
94
95    # ------------------------------------------------------------------
96    # Browse / search / recommendations
97    # ------------------------------------------------------------------
98
99    async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
100        """Browse Apple Music with support for playlist folders and radio stations."""
101        sub_path = path.split("://", 1)[1] if "://" in path else ""
102        path_parts = [part for part in sub_path.split("/") if part]
103        if not path_parts:
104            items = list(await super().browse(path))
105            items.append(
106                BrowseFolder(
107                    item_id="stations",
108                    provider=self.instance_id,
109                    path=f"{self.instance_id}://stations",
110                    name="Radio Stations",
111                    translation_key="radio_stations",
112                )
113            )
114            return items
115        if path_parts[0] == "playlists":
116            return await browse_playlists(self, path, path_parts)
117        if path_parts[0] == "stations":
118            return await self.recommendation_manager.browse_stations()
119        return await super().browse(path)
120
121    async def search(
122        self,
123        search_query: str,
124        media_types: list[MediaType] | None,
125        limit: int = 5,
126    ) -> SearchResults:
127        """Perform search on musicprovider."""
128        return await self.media_manager.search(search_query, media_types, limit)
129
130    async def get_recommendations(self) -> list[RecommendationFolder]:
131        """Get this provider's available recommendation rows, without items."""
132        return await self._recommendation_rows_from_payload()
133
134    async def get_recommendation_items(
135        self, item_id: str
136    ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
137        """
138        Get the items for a single recommendation row.
139
140        :param item_id: The item_id of the row, as returned by get_recommendations.
141        """
142        return await self._recommendation_items_from_payload(item_id)
143
144    # ------------------------------------------------------------------
145    # Media item getters
146    # ------------------------------------------------------------------
147
148    async def get_artist(self, prov_artist_id: str) -> Artist:
149        """Get full artist details by id."""
150        return await self.media_manager.get_artist(prov_artist_id)
151
152    async def get_album(self, prov_album_id: str) -> Album:
153        """Get full album details by id."""
154        return await self.media_manager.get_album(prov_album_id)
155
156    async def get_track(self, prov_track_id: str) -> Track:
157        """Get full track details by id."""
158        return await self.media_manager.get_track(prov_track_id)
159
160    async def get_playlist(self, prov_playlist_id: str, is_favourite: bool = False) -> Playlist:
161        """Get full playlist details by id."""
162        if prov_playlist_id.startswith("ra."):
163            return await self.recommendation_manager.get_station_playlist(prov_playlist_id)
164        return await self.media_manager.get_playlist(prov_playlist_id, is_favourite)
165
166    async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
167        """Get all album tracks for given album id."""
168        return await self.media_manager.get_album_tracks(prov_album_id)
169
170    async def resolve_image(self, path: str) -> str | bytes:
171        """Resolve an artwork token to a freshly signed artwork URL."""
172        media_type, _, item_id = path.partition("/")
173        return await self.media_manager.get_artwork_url(media_type, item_id) or ""
174
175    async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
176        """Get all playlist tracks for given playlist id."""
177        return await self.media_manager.get_playlist_tracks(prov_playlist_id, page)
178
179    async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
180        """Get a list of all albums for the given artist."""
181        return await self.media_manager.get_artist_albums(prov_artist_id)
182
183    async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
184        """Get a list of 10 most popular tracks for the given artist."""
185        return await self.media_manager.get_artist_toptracks(prov_artist_id)
186
187    async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
188        """Retrieve a dynamic list of tracks based on the provided item."""
189        return await self.recommendation_manager.get_similar_tracks(prov_track_id, limit)
190
191    async def get_similar_artists(self, prov_artist_id: str, limit: int = 25) -> list[Artist]:
192        """Retrieve a list of artists similar to the provided artist."""
193        return await self.recommendation_manager.get_similar_artists(prov_artist_id, limit)
194
195    # ------------------------------------------------------------------
196    # Library generators
197    # ------------------------------------------------------------------
198
199    async def get_library_artists(self) -> AsyncGenerator[Artist]:
200        """Retrieve library artists from the provider."""
201        async for item in self.library_manager.get_library_artists():
202            yield item
203
204    async def get_library_albums(self) -> AsyncGenerator[Album]:
205        """Retrieve library albums from the provider."""
206        async for item in self.library_manager.get_library_albums():
207            yield item
208
209    async def get_library_tracks(self) -> AsyncGenerator[Track]:
210        """Retrieve library tracks from the provider."""
211        async for item in self.library_manager.get_library_tracks():
212            yield item
213
214    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
215        """Retrieve playlists from the provider."""
216        async for item in self.library_manager.get_library_playlists():
217            yield item
218
219    # ------------------------------------------------------------------
220    # Library mutations
221    # ------------------------------------------------------------------
222
223    async def library_add(self, item: MediaItemType) -> bool:
224        """Add item to library."""
225        await self.library_manager.library_add(item)
226        return True
227
228    async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
229        """Remove item from library."""
230        await self.library_manager.library_remove(prov_item_id, media_type)
231        return True
232
233    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
234        """Add track(s) to playlist."""
235        await self.library_manager.add_playlist_tracks(prov_playlist_id, prov_track_ids)
236
237    async def remove_playlist_tracks(
238        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
239    ) -> None:
240        """Remove track(s) from playlist."""
241        await self.library_manager.remove_playlist_tracks(prov_playlist_id, positions_to_remove)
242
243    async def set_favorite(self, prov_item_id: str, media_type: MediaType, favorite: bool) -> None:
244        """Set the favorite status of an item."""
245        await self.library_manager.set_favorite(prov_item_id, media_type, favorite)
246
247    # ------------------------------------------------------------------
248    # Streaming
249    # ------------------------------------------------------------------
250
251    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
252        """Return the content details for the given track when it will be streamed."""
253        return await self.streaming_manager.get_stream_details(item_id)
254
255    async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
256        """Fetch and parse the full recommendations payload (folders with items)."""
257        return await self.recommendation_manager.get_personal_recommendations()
258