/
/
/
1"""Catalog and media lookups for Apple Music."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, cast
6
7from music_assistant_models.enums import MediaType
8from music_assistant_models.errors import MediaNotFoundError, MusicAssistantError
9from music_assistant_models.media_items import (
10 Album,
11 Artist,
12 Playlist,
13 SearchResults,
14 Track,
15)
16
17from music_assistant.controllers.cache import use_cache
18from music_assistant.helpers.track_filter import filter_tracks
19
20from .constants import ARTWORK_CACHE_EXPIRATION, PARSED_ITEM_CACHE_CHECKSUM
21from .helpers.utils import is_catalog_id, is_library_id, translate_media_type_to_apple_type
22from .parsers import (
23 format_artwork_url,
24 parse_album,
25 parse_artist,
26 parse_playlist,
27 parse_track,
28)
29
30if TYPE_CHECKING:
31 from .provider import AppleMusicProvider
32
33
34class AppleMusicMediaManager:
35 """Handles catalog reads and search for Apple Music."""
36
37 def __init__(self, provider: AppleMusicProvider) -> None:
38 """Initialize media manager."""
39 self.provider = provider
40 self.mass = provider.mass
41 self.instance_id = provider.instance_id
42 self.domain = provider.domain
43 self.api = provider.api_client
44 self.logger = provider.logger
45
46 @use_cache(cache_checksum=PARSED_ITEM_CACHE_CHECKSUM)
47 async def search(
48 self,
49 search_query: str,
50 media_types: list[MediaType] | None,
51 limit: int = 5,
52 ) -> SearchResults:
53 """Perform search on the Apple Music catalog."""
54 endpoint = f"catalog/{self.provider._storefront}/search"
55 limit = min(limit, 25)
56 searchresult = SearchResults()
57 if not media_types:
58 return searchresult
59 searchtypes = []
60 if MediaType.ARTIST in media_types:
61 searchtypes.append("artists")
62 if MediaType.ALBUM in media_types:
63 searchtypes.append("albums")
64 if MediaType.TRACK in media_types:
65 searchtypes.append("songs")
66 if MediaType.PLAYLIST in media_types:
67 searchtypes.append("playlists")
68 if not searchtypes:
69 return searchresult
70 searchtype = ",".join(searchtypes)
71 search_query = search_query.replace("'", "")
72 response = await self.api.get_data(
73 endpoint, term=search_query, types=searchtype, limit=limit
74 )
75 if "artists" in response["results"]:
76 searchresult.artists = [
77 *searchresult.artists,
78 *(
79 parse_artist(self.provider, item)
80 for item in response["results"]["artists"]["data"]
81 ),
82 ]
83 if "albums" in response["results"]:
84 searchresult.albums = [
85 *searchresult.albums,
86 *(
87 cast("Album", parse_album(self.provider, item))
88 for item in response["results"]["albums"]["data"]
89 ),
90 ]
91 if "songs" in response["results"]:
92 searchresult.tracks = [
93 *searchresult.tracks,
94 *(
95 parse_track(self.provider, item)
96 for item in response["results"]["songs"]["data"]
97 ),
98 ]
99 if "playlists" in response["results"]:
100 searchresult.playlists = [
101 *searchresult.playlists,
102 *(
103 parse_playlist(self.provider, item)
104 for item in response["results"]["playlists"]["data"]
105 ),
106 ]
107 return searchresult
108
109 @use_cache(cache_checksum=PARSED_ITEM_CACHE_CHECKSUM)
110 async def get_artist(self, prov_artist_id: str) -> Artist:
111 """Get full artist details by id."""
112 endpoint = f"catalog/{self.provider._storefront}/artists/{prov_artist_id}"
113 response = await self.api.get_data(endpoint, extend="editorialNotes")
114 return cast("Artist", parse_artist(self.provider, response["data"][0]))
115
116 @use_cache(cache_checksum=PARSED_ITEM_CACHE_CHECKSUM)
117 async def get_album(self, prov_album_id: str) -> Album:
118 """Get full album details by id."""
119 if is_library_id(prov_album_id):
120 endpoint = f"me/library/albums/{prov_album_id}"
121 response = await self.api.get_data(endpoint, include="catalog,artists")
122 else:
123 endpoint = f"catalog/{self.provider._storefront}/albums/{prov_album_id}"
124 response = await self.api.get_data(endpoint, include="artists")
125 rating_response = await self.api.get_ratings([prov_album_id], MediaType.ALBUM)
126 is_favourite = rating_response.get(prov_album_id)
127 return cast("Album", parse_album(self.provider, response["data"][0], is_favourite))
128
129 @use_cache(cache_checksum=PARSED_ITEM_CACHE_CHECKSUM)
130 async def get_track(self, prov_track_id: str) -> Track:
131 """Get full track details by id."""
132 endpoint = f"catalog/{self.provider._storefront}/songs/{prov_track_id}"
133 response = await self.api.get_data(endpoint, include="artists,albums")
134 rating_response = await self.api.get_ratings([prov_track_id], MediaType.TRACK)
135 is_favourite = rating_response.get(prov_track_id)
136 return parse_track(self.provider, response["data"][0], is_favourite)
137
138 async def get_playlist(
139 self,
140 prov_playlist_id: str,
141 is_favourite: bool = False,
142 can_edit_hint: bool | None = None,
143 library_id_override: str | None = None,
144 ) -> Playlist:
145 """Get full playlist details by id."""
146 return await self._get_regular_playlist(
147 prov_playlist_id,
148 is_favourite,
149 can_edit_hint=can_edit_hint,
150 library_id_override=library_id_override,
151 )
152
153 @use_cache(ARTWORK_CACHE_EXPIRATION, cache_checksum=PARSED_ITEM_CACHE_CHECKSUM)
154 async def get_artwork_url(self, media_type: str, prov_item_id: str) -> str | None:
155 """
156 Return the current artwork URL for the given item, if any.
157
158 Blobstore artwork URLs are presigned with a limited lifetime, so they are
159 resolved on demand (and cached well below the signature lifetime) instead
160 of being persisted anywhere.
161
162 :param media_type: The media type value of the artwork token (album/track/...).
163 :param prov_item_id: The provider item id of the item the artwork belongs to.
164 """
165 try:
166 apple_type = translate_media_type_to_apple_type(MediaType(media_type))
167 except ValueError, MusicAssistantError:
168 return None
169 # playlists use globalId ("pl.") catalog ids; all other types use the
170 # library id format to tell library and catalog items apart
171 if media_type == MediaType.PLAYLIST.value:
172 in_library = not is_catalog_id(prov_item_id)
173 else:
174 in_library = is_library_id(prov_item_id)
175 if in_library:
176 endpoint = f"me/library/{apple_type}/{prov_item_id}"
177 else:
178 endpoint = f"catalog/{self.provider._storefront}/{apple_type}/{prov_item_id}"
179 response = await self.api.get_data(endpoint, include="catalog")
180 item_obj = response["data"][0]
181 attributes = item_obj.get("attributes") or {}
182 if not attributes.get("artwork"):
183 # library items may only carry artwork on their catalog counterpart
184 catalog_data = item_obj.get("relationships", {}).get("catalog", {}).get("data") or []
185 if catalog_data:
186 attributes = catalog_data[0].get("attributes") or {}
187 return format_artwork_url(attributes)
188
189 @use_cache(cache_checksum=PARSED_ITEM_CACHE_CHECKSUM)
190 async def _get_regular_playlist(
191 self,
192 prov_playlist_id: str,
193 is_favourite: bool = False,
194 can_edit_hint: bool | None = None,
195 library_id_override: str | None = None,
196 ) -> Playlist:
197 """Fetch and cache details for a regular (non-station) playlist."""
198 if not is_catalog_id(prov_playlist_id):
199 endpoint = f"me/library/playlists/{prov_playlist_id}"
200 else:
201 endpoint = f"catalog/{self.provider._storefront}/playlists/{prov_playlist_id}"
202 response = await self.api.get_data(endpoint)
203 return parse_playlist(
204 self.provider,
205 response["data"][0],
206 is_favourite,
207 can_edit_hint=can_edit_hint,
208 library_id_override=library_id_override,
209 )
210
211 @use_cache(cache_checksum=PARSED_ITEM_CACHE_CHECKSUM, allow_expired_cache=True)
212 async def get_album_tracks(self, prov_album_id: str) -> list[Track]:
213 """Get all album tracks for given album id."""
214 if is_library_id(prov_album_id):
215 endpoint = f"me/library/albums/{prov_album_id}/tracks"
216 response = await self.api.get_data(endpoint, include="catalog,artists")
217 else:
218 endpoint = f"catalog/{self.provider._storefront}/albums/{prov_album_id}/tracks"
219 response = await self.api.get_data(endpoint, include="artists")
220 album = await self.get_album(prov_album_id)
221 track_ids = [track_obj["id"] for track_obj in response["data"] if "id" in track_obj]
222 rating_response = await self.api.get_ratings(track_ids, MediaType.TRACK)
223 tracks = []
224 for track_obj in response["data"]:
225 if "id" not in track_obj:
226 continue
227 track = parse_track(self.provider, track_obj, rating_response.get(track_obj["id"]))
228 track.album = album
229 tracks.append(track)
230 return tracks
231
232 async def get_playlist_tracks(self, prov_playlist_id: str, page: int = 0) -> list[Track]:
233 """Get all playlist tracks for given playlist id."""
234 if prov_playlist_id.startswith("ra."):
235 if page > 0:
236 return []
237 return await self._get_station_tracks(prov_playlist_id)
238 return await self._get_playlist_tracks_cached(prov_playlist_id, page)
239
240 @use_cache(3600 * 3, cache_checksum=PARSED_ITEM_CACHE_CHECKSUM)
241 async def _get_playlist_tracks_cached(
242 self, prov_playlist_id: str, page: int = 0
243 ) -> list[Track]:
244 """Fetch and cache tracks for a regular (non-station) playlist."""
245 if is_catalog_id(prov_playlist_id):
246 endpoint = f"catalog/{self.provider._storefront}/playlists/{prov_playlist_id}/tracks"
247 else:
248 endpoint = f"me/library/playlists/{prov_playlist_id}/tracks"
249 result: list[Track] = []
250 page_size = 100
251 offset = page * page_size
252 response = await self.api.get_data(
253 endpoint, include="artists,catalog", limit=page_size, offset=offset
254 )
255 if not response or "data" not in response:
256 return result
257 playlist_track_ids = [track["id"] for track in response["data"] if track and track["id"]]
258 rating_response = await self.api.get_ratings(playlist_track_ids, MediaType.TRACK)
259 for index, track in enumerate(response["data"]):
260 if track and track["id"]:
261 is_favourite = rating_response.get(track["id"])
262 parsed_track = parse_track(self.provider, track, is_favourite)
263 parsed_track.position = offset + index + 1
264 result.append(parsed_track)
265 return result
266
267 async def _get_station_tracks(self, station_id: str) -> list[Track]:
268 """Fetch the next batch of tracks for a radio station."""
269 tracks = await self._fetch_station_tracks(station_id)
270 if not tracks:
271 # Apple may rotate station IDs for personal stations; try to resolve the current one.
272 fresh_id = await self.provider.recommendation_manager.resolve_station_id(station_id)
273 if fresh_id and fresh_id != station_id:
274 self.logger.debug(
275 "Station ID %s appears stale, retrying with refreshed ID %s",
276 station_id,
277 fresh_id,
278 )
279 tracks = await self._fetch_station_tracks(fresh_id)
280 return filter_tracks(tracks)
281
282 async def _fetch_station_tracks(self, station_id: str) -> list[Track]:
283 """Fetch tracks for a station ID from the Apple Music API."""
284 response = await self.api.post_data(
285 f"me/stations/next-tracks/{station_id}", include="artists"
286 )
287 tracks = response.get("data", [])
288 if not tracks:
289 self.logger.debug(
290 "No tracks returned for station_id=%s; errors=%s",
291 station_id,
292 response.get("errors"),
293 )
294 return []
295 track_ids = [t["id"] for t in tracks if t and t.get("id")]
296 rating_response = await self.api.get_ratings(track_ids, MediaType.TRACK)
297 return [
298 parse_track(self.provider, t, rating_response.get(t["id"]))
299 for t in tracks
300 if t and t.get("id")
301 ]
302
303 @use_cache(3600 * 24 * 7, cache_checksum=PARSED_ITEM_CACHE_CHECKSUM, allow_expired_cache=True)
304 async def get_artist_albums(self, prov_artist_id: str) -> list[Album]:
305 """Get a list of all albums for the given artist."""
306 endpoint = f"catalog/{self.provider._storefront}/artists/{prov_artist_id}/albums"
307 try:
308 response = await self.api.get_all_items(endpoint)
309 except MediaNotFoundError:
310 self.logger.info("No albums found for artist %s", prov_artist_id)
311 return []
312 album_ids = [album["id"] for album in response if album["id"]]
313 rating_response = await self.api.get_ratings(album_ids, MediaType.ALBUM)
314 albums = []
315 for album in response:
316 if not album["id"]:
317 continue
318 parsed = parse_album(self.provider, album, rating_response.get(album["id"]))
319 if parsed:
320 albums.append(cast("Album", parsed))
321 return albums
322
323 @use_cache(3600 * 24 * 7, cache_checksum=PARSED_ITEM_CACHE_CHECKSUM, allow_expired_cache=True)
324 async def get_artist_toptracks(self, prov_artist_id: str) -> list[Track]:
325 """Get a list of 10 most popular tracks for the given artist."""
326 endpoint = f"catalog/{self.provider._storefront}/artists/{prov_artist_id}/view/top-songs"
327 try:
328 response = await self.api.get_data(endpoint)
329 except MediaNotFoundError:
330 self.logger.info("No top tracks found for artist %s", prov_artist_id)
331 return []
332 track_ids = [track["id"] for track in response["data"] if track["id"]]
333 rating_response = await self.api.get_ratings(track_ids, MediaType.TRACK)
334 return [
335 parse_track(self.provider, track, rating_response.get(track["id"]))
336 for track in response["data"]
337 if track["id"]
338 ]
339