/
/
/
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 @property
72 def max_concurrent_streams(self) -> int:
73 """Apple Music accounts allow a single active playback session."""
74 return 1
75
76 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
77 """
78 Return Config entries to configure this provider.
79
80 Authentication is handled by the setup flow (see setup_flow.py); only the
81 informational note is surfaced here.
82 """
83 return (CONF_ENTRY_UNOFFICIAL_PROVIDER,)
84
85 async def handle_async_init(self) -> None:
86 """Handle async initialization of the provider."""
87 self._music_user_token = cast(
88 "str | None",
89 self.get_setup_value(CONF_MUSIC_USER_MANUAL_TOKEN)
90 or self.get_setup_value(CONF_MUSIC_USER_TOKEN),
91 )
92 # a stored/manual app token only exists because the setup flow found the
93 # bundled one empty or invalid, so it takes precedence over the bundled token
94 self._music_app_token = (
95 cast("str | None", self.get_setup_value(CONF_MUSIC_APP_TOKEN)) or MUSIC_APP_TOKEN
96 )
97 self._storefront = await self.api_client.get_user_storefront()
98 await self.streaming_manager.initialize()
99
100 # ------------------------------------------------------------------
101 # Browse / search / recommendations
102 # ------------------------------------------------------------------
103
104 async def browse(self, path: str) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
105 """Browse Apple Music with support for playlist folders and radio stations."""
106 sub_path = path.split("://", 1)[1] if "://" in path else ""
107 path_parts = [part for part in sub_path.split("/") if part]
108 if not path_parts:
109 items = list(await super().browse(path))
110 items.append(
111 BrowseFolder(
112 item_id="stations",
113 provider=self.instance_id,
114 path=f"{self.instance_id}://stations",
115 name="Radio Stations",
116 translation_key="radio_stations",
117 )
118 )
119 return items
120 if path_parts[0] == "playlists":
121 return await browse_playlists(self, path, path_parts)
122 if path_parts[0] == "stations":
123 return await self.recommendation_manager.browse_stations()
124 return await super().browse(path)
125
126 async def search(
127 self,
128 search_query: str,
129 media_types: list[MediaType] | None,
130 limit: int = 5,
131 ) -> SearchResults:
132 """Perform search on musicprovider."""
133 return await self.media_manager.search(search_query, media_types, limit)
134
135 async def get_recommendations(self) -> list[RecommendationFolder]:
136 """Get this provider's available recommendation rows, without items."""
137 return await self._recommendation_rows_from_payload()
138
139 async def get_recommendation_items(
140 self, item_id: str
141 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
142 """
143 Get the items for a single recommendation row.
144
145 :param item_id: The item_id of the row, as returned by get_recommendations.
146 """
147 return await self._recommendation_items_from_payload(item_id)
148
149 # ------------------------------------------------------------------
150 # Media item getters
151 # ------------------------------------------------------------------
152
153 async def get_artist(self, prov_artist_id: str) -> Artist:
154 """Get full artist details by id."""
155 return await self.media_manager.get_artist(prov_artist_id)
156
157 async def get_album(self, prov_album_id: str) -> Album:
158 """Get full album details by id."""
159 return await self.media_manager.get_album(prov_album_id)
160
161 async def get_track(self, prov_track_id: str) -> Track:
162 """Get full track details by id."""
163 return await self.media_manager.get_track(prov_track_id)
164
165 async def get_playlist(self, prov_playlist_id: str, is_favourite: bool = False) -> Playlist:
166 """Get full playlist details by id."""
167 if prov_playlist_id.startswith("ra."):
168 return await self.recommendation_manager.get_station_playlist(prov_playlist_id)
169 return await self.media_manager.get_playlist(prov_playlist_id, is_favourite)
170
171 async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
172 """Get all album tracks for given album id."""
173 return await self.media_manager.get_album_tracks(prov_album_id)
174
175 async def resolve_image(self, path: str) -> str | bytes:
176 """Resolve an artwork token to a freshly signed artwork URL."""
177 media_type, _, item_id = path.partition("/")
178 return await self.media_manager.get_artwork_url(media_type, item_id) or ""
179
180 async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
181 """Get all playlist tracks for given playlist id."""
182 return await self.media_manager.get_playlist_tracks(prov_playlist_id, page)
183
184 async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
185 """Get a list of all albums for the given artist."""
186 return await self.media_manager.get_artist_albums(prov_artist_id)
187
188 async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
189 """Get a list of 10 most popular tracks for the given artist."""
190 return await self.media_manager.get_artist_toptracks(prov_artist_id)
191
192 async def get_similar_tracks(self, prov_track_id: str, limit: int = 25) -> list[Track]:
193 """Retrieve a dynamic list of tracks based on the provided item."""
194 return await self.recommendation_manager.get_similar_tracks(prov_track_id, limit)
195
196 async def get_similar_artists(self, prov_artist_id: str, limit: int = 25) -> list[Artist]:
197 """Retrieve a list of artists similar to the provided artist."""
198 return await self.recommendation_manager.get_similar_artists(prov_artist_id, limit)
199
200 # ------------------------------------------------------------------
201 # Library generators
202 # ------------------------------------------------------------------
203
204 async def get_library_artists(self) -> AsyncGenerator[Artist]:
205 """Retrieve library artists from the provider."""
206 async for item in self.library_manager.get_library_artists():
207 yield item
208
209 async def get_library_albums(self) -> AsyncGenerator[Album]:
210 """Retrieve library albums from the provider."""
211 async for item in self.library_manager.get_library_albums():
212 yield item
213
214 async def get_library_tracks(self) -> AsyncGenerator[Track]:
215 """Retrieve library tracks from the provider."""
216 async for item in self.library_manager.get_library_tracks():
217 yield item
218
219 async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
220 """Retrieve playlists from the provider."""
221 async for item in self.library_manager.get_library_playlists():
222 yield item
223
224 # ------------------------------------------------------------------
225 # Library mutations
226 # ------------------------------------------------------------------
227
228 async def library_add(self, item: MediaItemType) -> bool:
229 """Add item to library."""
230 await self.library_manager.library_add(item)
231 return True
232
233 async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
234 """Remove item from library."""
235 await self.library_manager.library_remove(prov_item_id, media_type)
236 return True
237
238 async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
239 """Add track(s) to playlist."""
240 await self.library_manager.add_playlist_tracks(prov_playlist_id, prov_track_ids)
241
242 async def remove_playlist_tracks(
243 self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
244 ) -> None:
245 """Remove track(s) from playlist."""
246 await self.library_manager.remove_playlist_tracks(prov_playlist_id, positions_to_remove)
247
248 async def set_favorite(self, prov_item_id: str, media_type: MediaType, favorite: bool) -> None:
249 """Set the favorite status of an item."""
250 await self.library_manager.set_favorite(prov_item_id, media_type, favorite)
251
252 # ------------------------------------------------------------------
253 # Streaming
254 # ------------------------------------------------------------------
255
256 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
257 """Return the content details for the given track when it will be streamed."""
258 return await self.streaming_manager.get_stream_details(item_id)
259
260 async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
261 """Fetch and parse the full recommendations payload (folders with items)."""
262 return await self.recommendation_manager.get_personal_recommendations()
263