/
/
/
1"""
2Apple Music musicprovider support for MusicAssistant.
3
4TODO MUSIC_APP_TOKEN expires after 6 months so should have a distribution mechanism outside
5 compulsory application updates. It is only a semi-private key in JWT format so code be refreshed
6 daily by a GitHub action and downloaded by the provider each initialise.
7TODO Widevine keys can be obtained dynamically from Apple Music API rather than copied into Docker
8 build. This is undocumented but @maxlyth has a working example.
9TODO MUSIC_USER_TOKEN must be refreshed (~min 180 days) and needs mechanism to prompt user to
10 re-authenticate in browser.
11TODO Current provider ignores private tracks that are not available in the storefront catalog as
12 streamable url is derived from the catalog id. It is undecumented but @maxlyth has a working
13 example to get a streamable url from the library id.
14"""
15
16from __future__ import annotations
17
18import base64
19import json
20import os
21import pathlib
22import re
23import time
24from typing import TYPE_CHECKING, Any
25
26import aiofiles
27from aiohttp import web
28from aiohttp.client_exceptions import ClientError
29from music_assistant_models.config_entries import ConfigEntry, ConfigValueType
30from music_assistant_models.enums import (
31 AlbumType,
32 ConfigEntryType,
33 ContentType,
34 ExternalID,
35 ImageType,
36 MediaType,
37 ProviderFeature,
38 StreamType,
39)
40from music_assistant_models.errors import (
41 LoginFailed,
42 MediaNotFoundError,
43 MusicAssistantError,
44 ResourceTemporarilyUnavailable,
45)
46from music_assistant_models.media_items import (
47 Album,
48 Artist,
49 AudioFormat,
50 ItemMapping,
51 MediaItemImage,
52 MediaItemType,
53 Playlist,
54 ProviderMapping,
55 SearchResults,
56 Track,
57 UniqueList,
58)
59from music_assistant_models.streamdetails import StreamDetails
60from pywidevine import PSSH, Cdm, Device, DeviceTypes
61from pywidevine.license_protocol_pb2 import WidevinePsshData
62from shortuuid import uuid
63
64from music_assistant.controllers.cache import use_cache
65from music_assistant.helpers.app_vars import app_var
66from music_assistant.helpers.auth import AuthenticationHelper
67from music_assistant.helpers.json import json_loads
68from music_assistant.helpers.playlists import fetch_playlist
69from music_assistant.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
70from music_assistant.helpers.util import infer_album_type, parse_title_and_version
71from music_assistant.models.music_provider import MusicProvider
72
73if TYPE_CHECKING:
74 from collections.abc import AsyncGenerator
75
76 from music_assistant_models.config_entries import ProviderConfig
77 from music_assistant_models.provider import ProviderManifest
78
79 from music_assistant import MusicAssistant
80 from music_assistant.models import ProviderInstanceType
81
82
83SUPPORTED_FEATURES = {
84 ProviderFeature.LIBRARY_ARTISTS,
85 ProviderFeature.LIBRARY_ALBUMS,
86 ProviderFeature.LIBRARY_TRACKS,
87 ProviderFeature.LIBRARY_PLAYLISTS,
88 ProviderFeature.BROWSE,
89 ProviderFeature.SEARCH,
90 ProviderFeature.ARTIST_ALBUMS,
91 ProviderFeature.ARTIST_TOPTRACKS,
92 ProviderFeature.SIMILAR_TRACKS,
93 ProviderFeature.LIBRARY_ALBUMS_EDIT,
94 ProviderFeature.LIBRARY_ARTISTS_EDIT,
95 ProviderFeature.LIBRARY_PLAYLISTS_EDIT,
96 ProviderFeature.LIBRARY_TRACKS_EDIT,
97 ProviderFeature.FAVORITE_ALBUMS_EDIT,
98 ProviderFeature.FAVORITE_TRACKS_EDIT,
99 ProviderFeature.FAVORITE_PLAYLISTS_EDIT,
100}
101
102MUSIC_APP_TOKEN = app_var(8)
103WIDEVINE_BASE_PATH = "/usr/local/bin/widevine_cdm"
104DECRYPT_CLIENT_ID_FILENAME = "client_id.bin"
105DECRYPT_PRIVATE_KEY_FILENAME = "private_key.pem"
106UNKNOWN_PLAYLIST_NAME = "Unknown Apple Music Playlist"
107
108CONF_MUSIC_APP_TOKEN = "music_app_token"
109CONF_MUSIC_USER_TOKEN = "music_user_token"
110CONF_MUSIC_USER_MANUAL_TOKEN = "music_user_manual_token"
111CONF_MUSIC_USER_TOKEN_TIMESTAMP = "music_user_token_timestamp"
112CACHE_CATEGORY_DECRYPT_KEY = 1
113
114
115async def setup(
116 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
117) -> ProviderInstanceType:
118 """Initialize provider(instance) with given configuration."""
119 return AppleMusicProvider(mass, manifest, config, SUPPORTED_FEATURES)
120
121
122async def get_config_entries(
123 mass: MusicAssistant,
124 instance_id: str | None = None,
125 action: str | None = None,
126 values: dict[str, ConfigValueType] | None = None,
127) -> tuple[ConfigEntry, ...]:
128 """
129 Return Config entries to setup this provider.
130
131 instance_id: id of an existing provider instance (None if new instance setup).
132 action: [optional] action key called from config entries UI.
133 values: the (intermediate) raw values for config entries sent with the action.
134 """
135
136 def validate_user_token(token):
137 if not isinstance(token, str):
138 return False
139 valid = re.findall(r"[a-zA-Z0-9=/+]{32,}==$", token)
140 return bool(valid)
141
142 # Check for valid app token (1st with regex and then API check) otherwise display a config field
143 default_app_token_valid = False
144 async with (
145 mass.http_session.get(
146 "https://api.music.apple.com/v1/test",
147 headers={"Authorization": f"Bearer {MUSIC_APP_TOKEN}"},
148 ssl=True,
149 timeout=10,
150 ) as response,
151 ):
152 if response.status == 200:
153 values[CONF_MUSIC_APP_TOKEN] = f"{MUSIC_APP_TOKEN}"
154 default_app_token_valid = True
155
156 # Action is to launch MusicKit flow
157 if action == "CONF_ACTION_AUTH" and default_app_token_valid:
158 callback_method = "POST"
159 async with AuthenticationHelper(mass, values["session_id"], callback_method) as auth_helper:
160 callback_url = auth_helper.callback_url
161 flow_base_path = f"apple_music_auth/{values['session_id']}/"
162 flow_timeout = 600
163 parent_file_path = pathlib.Path(__file__).parent.resolve()
164 base_url = f"{mass.webserver.base_url}/{flow_base_path}"
165 flow_base_url = f"{base_url}index.html"
166
167 async def serve_mk_auth_page(request: web.Request) -> web.Response:
168 auth_html_path = parent_file_path.joinpath("musickit_auth/musickit_wrapper.html")
169 return web.FileResponse(
170 auth_html_path,
171 headers={"content-type": "text/html"},
172 )
173
174 async def serve_mk_auth_css(request: web.Request) -> web.Response:
175 auth_css_path = parent_file_path.joinpath("musickit_auth/musickit_wrapper.css")
176 return web.FileResponse(
177 auth_css_path,
178 headers={
179 "content-type": "text/css",
180 },
181 )
182
183 async def serve_mk_glue(request: web.Request) -> web.Response:
184 return_html = f"""
185 const return_url='{callback_url}';
186 const base_url='{base_url}';
187 const app_token='{values[CONF_MUSIC_APP_TOKEN]}';
188 const callback_method='{callback_method}';
189 const user_token='{
190 values[CONF_MUSIC_USER_TOKEN]
191 if validate_user_token(values[CONF_MUSIC_USER_TOKEN])
192 else ""
193 }';
194 const user_token_timestamp='{values[CONF_MUSIC_USER_TOKEN_TIMESTAMP]}';
195 const flow_timeout={max([flow_timeout - 10, 60])};
196 const flow_start_time={int(time.time())};
197 const mass_version='{mass.version}';
198 """
199 return web.Response(
200 body=return_html,
201 headers={
202 "content-type": "text/javascript",
203 },
204 )
205
206 mass.webserver.register_dynamic_route(
207 f"/{flow_base_path}index.html", serve_mk_auth_page
208 )
209 mass.webserver.register_dynamic_route(f"/{flow_base_path}index.css", serve_mk_auth_css)
210 mass.webserver.register_dynamic_route(f"/{flow_base_path}index.js", serve_mk_glue)
211
212 try:
213 result = await auth_helper.authenticate(flow_base_url, flow_timeout)
214 values[CONF_MUSIC_USER_TOKEN] = result["music-user-token"]
215 values[CONF_MUSIC_USER_TOKEN_TIMESTAMP] = result["music-user-token-timestamp"]
216 except KeyError:
217 # no music-user-token URL param was found so likely user cancelled the auth
218 pass
219 except Exception as error:
220 raise LoginFailed(f"Failed to authenticate with Apple '{error}'.")
221 finally:
222 mass.webserver.unregister_dynamic_route(f"/{flow_base_path}index.html")
223 mass.webserver.unregister_dynamic_route(f"/{flow_base_path}index.css")
224 mass.webserver.unregister_dynamic_route(f"/{flow_base_path}index.js")
225
226 # ruff: noqa: ARG001
227 return (
228 ConfigEntry(
229 key=CONF_MUSIC_APP_TOKEN,
230 type=ConfigEntryType.SECURE_STRING,
231 label="MusicKit App Token",
232 hidden=default_app_token_valid,
233 required=True,
234 value=values.get(CONF_MUSIC_APP_TOKEN) if values else None,
235 ),
236 ConfigEntry(
237 key=CONF_MUSIC_USER_TOKEN,
238 type=ConfigEntryType.SECURE_STRING,
239 label="Music User Token",
240 required=False,
241 action="CONF_ACTION_AUTH",
242 description="Authenticate with Apple Music to retrieve a valid music user token.",
243 action_label="Authenticate with Apple Music",
244 value=values.get(CONF_MUSIC_USER_TOKEN)
245 if (
246 values
247 and isinstance(values.get(CONF_MUSIC_USER_TOKEN_TIMESTAMP), int)
248 and (
249 values.get(CONF_MUSIC_USER_TOKEN_TIMESTAMP) > (time.time() - (3600 * 24 * 150))
250 )
251 )
252 else None,
253 ),
254 ConfigEntry(
255 key=CONF_MUSIC_USER_MANUAL_TOKEN,
256 type=ConfigEntryType.SECURE_STRING,
257 label="Manual Music User Token",
258 required=False,
259 category="advanced",
260 description=(
261 "Authenticate with a manual Music User Token in case the Authentication flow"
262 " is unsupported (e.g. when using child accounts)."
263 ),
264 help_link="https://www.music-assistant.io/music-providers/apple-music/",
265 value=values.get(CONF_MUSIC_USER_MANUAL_TOKEN),
266 ),
267 ConfigEntry(
268 key=CONF_MUSIC_USER_TOKEN_TIMESTAMP,
269 type=ConfigEntryType.INTEGER,
270 description="Timestamp music user token was updated.",
271 label="Music User Token Timestamp",
272 hidden=True,
273 required=True,
274 default_value=0,
275 value=values.get(CONF_MUSIC_USER_TOKEN_TIMESTAMP) if values else 0,
276 ),
277 )
278
279
280class AppleMusicProvider(MusicProvider):
281 """Implementation of an Apple Music MusicProvider."""
282
283 _music_user_token: str | None = None
284 _music_app_token: str | None = None
285 _storefront: str | None = None
286 _decrypt_client_id: bytes | None = None
287 _decrypt_private_key: bytes | None = None
288 _session_id: str | None = None
289 # rate limiter needs to be specified on provider-level,
290 # so make it an instance attribute
291 throttler = ThrottlerManager(rate_limit=1, period=2, initial_backoff=15)
292
293 async def handle_async_init(self) -> None:
294 """Handle async initialization of the provider."""
295 self._music_user_token = self.config.get_value(
296 CONF_MUSIC_USER_MANUAL_TOKEN
297 ) or self.config.get_value(CONF_MUSIC_USER_TOKEN)
298 self._music_app_token = self.config.get_value(CONF_MUSIC_APP_TOKEN)
299 self._storefront = await self._get_user_storefront()
300 # create random session id to use for decryption keys
301 # to invalidate cached keys on each provider initialization
302 self._session_id = str(uuid())
303 async with aiofiles.open(
304 os.path.join(WIDEVINE_BASE_PATH, DECRYPT_CLIENT_ID_FILENAME), "rb"
305 ) as _file:
306 self._decrypt_client_id = await _file.read()
307 async with aiofiles.open(
308 os.path.join(WIDEVINE_BASE_PATH, DECRYPT_PRIVATE_KEY_FILENAME), "rb"
309 ) as _file:
310 self._decrypt_private_key = await _file.read()
311
312 @use_cache()
313 async def search(
314 self, search_query: str, media_types: list[MediaType] | None, limit: int = 5
315 ) -> SearchResults:
316 """Perform search on musicprovider.
317
318 :param search_query: Search query.
319 :param media_types: A list of media_types to include. All types if None.
320 :param limit: Number of items to return in the search (per type).
321 """
322 endpoint = f"catalog/{self._storefront}/search"
323 # Apple music has a limit of 25 items for the search endpoint
324 limit = min(limit, 25)
325 searchresult = SearchResults()
326 searchtypes = []
327 if MediaType.ARTIST in media_types:
328 searchtypes.append("artists")
329 if MediaType.ALBUM in media_types:
330 searchtypes.append("albums")
331 if MediaType.TRACK in media_types:
332 searchtypes.append("songs")
333 if MediaType.PLAYLIST in media_types:
334 searchtypes.append("playlists")
335 if not searchtypes:
336 return searchresult
337 searchtype = ",".join(searchtypes)
338 search_query = search_query.replace("'", "")
339 response = await self._get_data(endpoint, term=search_query, types=searchtype, limit=limit)
340 if "artists" in response["results"]:
341 searchresult.artists += [
342 self._parse_artist(item) for item in response["results"]["artists"]["data"]
343 ]
344 if "albums" in response["results"]:
345 searchresult.albums += [
346 self._parse_album(item) for item in response["results"]["albums"]["data"]
347 ]
348 if "songs" in response["results"]:
349 searchresult.tracks += [
350 self._parse_track(item) for item in response["results"]["songs"]["data"]
351 ]
352 if "playlists" in response["results"]:
353 searchresult.playlists += [
354 self._parse_playlist(item) for item in response["results"]["playlists"]["data"]
355 ]
356 return searchresult
357
358 async def get_library_artists(self) -> AsyncGenerator[Artist, None]:
359 """Retrieve library artists from spotify."""
360 endpoint = "me/library/artists"
361 for item in await self._get_all_items(endpoint, include="catalog", extend="editorialNotes"):
362 if item and item["id"]:
363 yield self._parse_artist(item)
364
365 async def get_library_albums(self) -> AsyncGenerator[Album, None]:
366 """Retrieve library albums from the provider."""
367 endpoint = "me/library/albums"
368 album_items = await self._get_all_items(
369 endpoint, include="catalog,artists", extend="editorialNotes"
370 )
371 album_catalog_item_ids = [
372 item["id"]
373 for item in album_items
374 if item and item["id"] and not self.is_library_id(item["id"])
375 ]
376 album_library_item_ids = [
377 item["id"]
378 for item in album_items
379 if item and item["id"] and self.is_library_id(item["id"])
380 ]
381 rating_catalog_response = await self._get_ratings(album_catalog_item_ids, MediaType.ALBUM)
382 rating_library_response = await self._get_ratings(album_library_item_ids, MediaType.ALBUM)
383 for item in album_items:
384 if item and item["id"]:
385 is_favourite = (
386 rating_catalog_response.get(item["id"])
387 if not self.is_library_id(item["id"])
388 else rating_library_response.get(item["id"])
389 )
390 album = self._parse_album(item, is_favourite)
391 if album:
392 yield album
393
394 async def get_library_tracks(self) -> AsyncGenerator[Track, None]:
395 """Retrieve library tracks from the provider."""
396 endpoint = "me/library/songs"
397 song_catalog_ids = []
398 library_only_tracks = []
399 for item in await self._get_all_items(endpoint):
400 catalog_id = item.get("attributes", {}).get("playParams", {}).get("catalogId")
401 if not catalog_id:
402 # Track is library-only (private/uploaded), use library ID instead
403 library_only_tracks.append(item)
404 else:
405 song_catalog_ids.append(catalog_id)
406 # Obtain catalog info per 200 songs, the documented limit of 300 results in a 504 timeout
407 max_limit = 200
408 for i in range(0, len(song_catalog_ids), max_limit):
409 catalog_ids = song_catalog_ids[i : i + max_limit]
410 catalog_endpoint = f"catalog/{self._storefront}/songs"
411 response = await self._get_data(
412 catalog_endpoint, ids=",".join(catalog_ids), include="artists,albums"
413 )
414 # Fetch ratings for this batch
415 rating_response = await self._get_ratings(catalog_ids, MediaType.TRACK)
416 for item in response["data"]:
417 is_favourite = rating_response.get(item["id"])
418 track = self._parse_track(item, is_favourite)
419 yield track
420 # Yield library-only tracks using their library metadata
421 library_ids = [item["id"] for item in library_only_tracks if item and item["id"]]
422 library_rating_response = await self._get_ratings(library_ids, MediaType.TRACK)
423 for item in library_only_tracks:
424 is_favourite = library_rating_response.get(item["id"])
425 yield self._parse_track(item, is_favourite)
426
427 async def get_library_playlists(self) -> AsyncGenerator[Playlist, None]:
428 """Retrieve playlists from the provider."""
429 endpoint = "me/library/playlists"
430 playlist_items = await self._get_all_items(endpoint)
431 playlist_library_item_ids = [
432 item["id"]
433 for item in playlist_items
434 if item and item["id"] and self.is_library_id(item["id"])
435 ]
436 rating_library_response = await self._get_ratings(
437 playlist_library_item_ids, MediaType.PLAYLIST
438 )
439 for item in playlist_items:
440 is_favourite = rating_library_response.get(item["id"])
441 # Prefer catalog information over library information in case of public playlists
442 if item["attributes"]["hasCatalog"]:
443 yield await self.get_playlist(
444 item["attributes"]["playParams"]["globalId"], is_favourite
445 )
446 elif item and item["id"]:
447 yield self._parse_playlist(item, is_favourite)
448
449 @use_cache()
450 async def get_artist(self, prov_artist_id) -> Artist:
451 """Get full artist details by id."""
452 endpoint = f"catalog/{self._storefront}/artists/{prov_artist_id}"
453 response = await self._get_data(endpoint, extend="editorialNotes")
454 return self._parse_artist(response["data"][0])
455
456 @use_cache()
457 async def get_album(self, prov_album_id) -> Album:
458 """Get full album details by id."""
459 endpoint = f"catalog/{self._storefront}/albums/{prov_album_id}"
460 response = await self._get_data(endpoint, include="artists")
461 rating_response = await self._get_ratings([prov_album_id], MediaType.ALBUM)
462 is_favourite = rating_response.get(prov_album_id)
463 return self._parse_album(response["data"][0], is_favourite)
464
465 @use_cache()
466 async def get_track(self, prov_track_id) -> Track:
467 """Get full track details by id."""
468 endpoint = f"catalog/{self._storefront}/songs/{prov_track_id}"
469 response = await self._get_data(endpoint, include="artists,albums")
470 rating_response = await self._get_ratings([prov_track_id], MediaType.TRACK)
471 is_favourite = rating_response.get(prov_track_id)
472 return self._parse_track(response["data"][0], is_favourite)
473
474 @use_cache()
475 async def get_playlist(self, prov_playlist_id, is_favourite: bool = False) -> Playlist:
476 """Get full playlist details by id."""
477 if not self.is_library_id(prov_playlist_id):
478 endpoint = f"catalog/{self._storefront}/playlists/{prov_playlist_id}"
479 else:
480 endpoint = f"me/library/playlists/{prov_playlist_id}"
481 endpoint = f"catalog/{self._storefront}/playlists/{prov_playlist_id}"
482 response = await self._get_data(endpoint)
483 return self._parse_playlist(response["data"][0], is_favourite)
484
485 @use_cache()
486 async def get_album_tracks(self, prov_album_id) -> list[Track]:
487 """Get all album tracks for given album id."""
488 endpoint = f"catalog/{self._storefront}/albums/{prov_album_id}/tracks"
489 response = await self._get_data(endpoint, include="artists")
490 # Including albums results in a 504 error, so we need to fetch the album separately
491 album = await self.get_album(prov_album_id)
492 track_ids = [track_obj["id"] for track_obj in response["data"] if "id" in track_obj]
493 rating_response = await self._get_ratings(track_ids, MediaType.TRACK)
494 tracks = []
495 for track_obj in response["data"]:
496 if "id" not in track_obj:
497 continue
498 track = self._parse_track(track_obj, rating_response.get(track_obj["id"]))
499 track.album = album
500 tracks.append(track)
501 return tracks
502
503 @use_cache(3600 * 3) # cache for 3 hours
504 async def get_playlist_tracks(self, prov_playlist_id, page: int = 0) -> list[Track]:
505 """Get all playlist tracks for given playlist id."""
506 if self._is_catalog_id(prov_playlist_id):
507 endpoint = f"catalog/{self._storefront}/playlists/{prov_playlist_id}/tracks"
508 else:
509 endpoint = f"me/library/playlists/{prov_playlist_id}/tracks"
510 result = []
511 page_size = 100
512 offset = page * page_size
513 response = await self._get_data(
514 endpoint, include="artists,catalog", limit=page_size, offset=offset
515 )
516 if not response or "data" not in response:
517 return result
518 playlist_track_ids = [track["id"] for track in response["data"] if track and track["id"]]
519 rating_response = await self._get_ratings(playlist_track_ids, MediaType.TRACK)
520 for index, track in enumerate(response["data"]):
521 if track and track["id"]:
522 is_favourite = rating_response.get(track["id"])
523 parsed_track = self._parse_track(track, is_favourite)
524 parsed_track.position = offset + index + 1
525 result.append(parsed_track)
526 return result
527
528 @use_cache(3600 * 24 * 7) # cache for 7 days
529 async def get_artist_albums(self, prov_artist_id) -> list[Album]:
530 """Get a list of all albums for the given artist."""
531 endpoint = f"catalog/{self._storefront}/artists/{prov_artist_id}/albums"
532 try:
533 response = await self._get_all_items(endpoint)
534 except MediaNotFoundError:
535 # Some artists do not have albums, return empty list
536 self.logger.info("No albums found for artist %s", prov_artist_id)
537 return []
538 album_ids = [album["id"] for album in response if album["id"]]
539 rating_response = await self._get_ratings(album_ids, MediaType.ALBUM)
540 albums = []
541 for album in response:
542 if not album["id"]:
543 continue
544 is_favourite = rating_response.get(album["id"])
545 parsed_album = self._parse_album(album, is_favourite)
546 if parsed_album:
547 albums.append(parsed_album)
548 return albums
549
550 @use_cache(3600 * 24 * 7) # cache for 7 days
551 async def get_artist_toptracks(self, prov_artist_id) -> list[Track]:
552 """Get a list of 10 most popular tracks for the given artist."""
553 endpoint = f"catalog/{self._storefront}/artists/{prov_artist_id}/view/top-songs"
554 try:
555 response = await self._get_data(endpoint)
556 except MediaNotFoundError:
557 # Some artists do not have top tracks, return empty list
558 self.logger.info("No top tracks found for artist %s", prov_artist_id)
559 return []
560 track_ids = [track["id"] for track in response["data"] if track["id"]]
561 rating_response = await self._get_ratings(track_ids, MediaType.TRACK)
562 tracks = []
563 for track in response["data"]:
564 if not track["id"]:
565 continue
566 is_favourite = rating_response.get(track["id"])
567 tracks.append(self._parse_track(track, is_favourite))
568 return tracks
569
570 async def library_add(self, item: MediaItemType) -> None:
571 """Add item to library."""
572 item_type = self._translate_media_type_to_apple_type(item.media_type)
573 kwargs = {
574 f"ids[{item_type}]": item.item_id,
575 }
576 await self._post_data("me/library/", **kwargs)
577
578 async def library_remove(self, prov_item_id, media_type: MediaType) -> None:
579 """Remove item from library."""
580 self.logger.warning(
581 "Deleting items from your library is not yet supported by the Apple Music API. "
582 f"Skipping deletion of {media_type} - {prov_item_id}."
583 )
584
585 async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]):
586 """Add track(s) to playlist."""
587 endpoint = f"me/library/playlists/{prov_playlist_id}/tracks"
588 data = {
589 "data": [
590 {
591 "id": track_id,
592 "type": "library-songs" if self.is_library_id(track_id) else "songs",
593 }
594 for track_id in prov_track_ids
595 ]
596 }
597 await self._post_data(endpoint, data=data)
598
599 async def remove_playlist_tracks(
600 self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
601 ) -> None:
602 """Remove track(s) from playlist."""
603 self.logger.warning(
604 "Removing tracks from playlists is not supported by the Apple Music "
605 "API. Make sure to delete them using the Apple Music app."
606 )
607
608 @use_cache(3600 * 24) # cache for 24 hours
609 async def get_similar_tracks(self, prov_track_id, limit=25) -> list[Track]:
610 """Retrieve a dynamic list of tracks based on the provided item."""
611 # Note, Apple music does not have an official endpoint for similar tracks.
612 # We will use the next-tracks endpoint to get a list of tracks that are similar to the
613 # provided track. However, Apple music only provides 2 tracks at a time, so we will
614 # need to call the endpoint multiple times. Therefore, set a limit to 6 to prevent
615 # flooding the apple music api.
616 limit = 6
617 endpoint = f"me/stations/next-tracks/ra.{prov_track_id}"
618 found_tracks = []
619 while len(found_tracks) < limit:
620 response = await self._post_data(endpoint, include="artists")
621 if not response or "data" not in response:
622 break
623 track_ids = [track["id"] for track in response["data"] if track and track["id"]]
624 rating_response = await self._get_ratings(track_ids, MediaType.TRACK)
625 for track in response["data"]:
626 if track and track["id"]:
627 is_favourite = rating_response.get(track["id"])
628 found_tracks.append(self._parse_track(track, is_favourite))
629 return found_tracks
630
631 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
632 """Return the content details for the given track when it will be streamed."""
633 stream_metadata = await self._fetch_song_stream_metadata(item_id)
634 if self.is_library_id(item_id):
635 # Library items are not encrypted and do not need decryption keys
636 try:
637 stream_url = stream_metadata["assets"][0]["URL"]
638 except (KeyError, IndexError, TypeError) as exc:
639 raise MediaNotFoundError(
640 f"Failed to extract stream URL for library track {item_id}: {exc}"
641 ) from exc
642 return StreamDetails(
643 item_id=item_id,
644 provider=self.instance_id,
645 path=stream_url,
646 stream_type=StreamType.HTTP,
647 audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
648 can_seek=True,
649 allow_seek=True,
650 )
651 # Continue to obtain decryption keys for catalog items
652 license_url = stream_metadata["hls-key-server-url"]
653 stream_url, uri = await self._parse_stream_url_and_uri(stream_metadata["assets"])
654 if not stream_url or not uri:
655 raise MediaNotFoundError("No stream URL found for song.")
656 key_id = base64.b64decode(uri.split(",")[1])
657 return StreamDetails(
658 item_id=item_id,
659 provider=self.instance_id,
660 audio_format=AudioFormat(content_type=ContentType.MP4, codec_type=ContentType.AAC),
661 stream_type=StreamType.ENCRYPTED_HTTP,
662 decryption_key=await self._get_decryption_key(license_url, key_id, uri, item_id),
663 path=stream_url,
664 can_seek=True,
665 allow_seek=True,
666 )
667
668 async def set_favorite(self, prov_item_id: str, media_type: MediaType, favorite: bool) -> None:
669 """Set the favorite status of an item."""
670 data = {
671 "type": "ratings",
672 "attributes": {
673 "value": 1 if favorite else -1,
674 },
675 }
676 item_type = self._translate_media_type_to_apple_type(media_type)
677 if self._is_catalog_id(prov_item_id):
678 endpoint = f"me/ratings/{item_type}/{prov_item_id}"
679 else:
680 endpoint = f"me/ratings/library-{item_type}/{prov_item_id}"
681 await self._put_data(endpoint, data=data)
682
683 def _parse_artist(self, artist_obj: dict[str, Any]) -> Artist:
684 """Parse artist object to generic layout."""
685 relationships = artist_obj.get("relationships", {})
686 if (
687 artist_obj.get("type") == "library-artists"
688 and relationships.get("catalog", {}).get("data", []) != []
689 ):
690 artist_id = relationships["catalog"]["data"][0]["id"]
691 attributes = relationships["catalog"]["data"][0]["attributes"]
692 elif "attributes" in artist_obj:
693 artist_id = artist_obj["id"]
694 attributes = artist_obj["attributes"]
695 else:
696 artist_id = artist_obj["id"]
697 self.logger.debug("No attributes found for artist %s", artist_obj)
698 # No more details available other than the id, return an ItemMapping
699 return ItemMapping(
700 media_type=MediaType.ARTIST,
701 provider=self.instance_id,
702 item_id=artist_id,
703 name=artist_id,
704 )
705 artist = Artist(
706 item_id=artist_id,
707 name=attributes.get("name"),
708 provider=self.domain,
709 provider_mappings={
710 ProviderMapping(
711 item_id=artist_id,
712 provider_domain=self.domain,
713 provider_instance=self.instance_id,
714 url=attributes.get("url"),
715 )
716 },
717 )
718 if artwork := attributes.get("artwork"):
719 artist.metadata.add_image(
720 MediaItemImage(
721 provider=self.instance_id,
722 type=ImageType.THUMB,
723 path=artwork["url"].format(w=artwork["width"], h=artwork["height"]),
724 remotely_accessible=True,
725 )
726 )
727 if genres := attributes.get("genreNames"):
728 artist.metadata.genres = set(genres)
729 if notes := attributes.get("editorialNotes"):
730 artist.metadata.description = notes.get("standard") or notes.get("short")
731 return artist
732
733 def _parse_album(
734 self, album_obj: dict, is_favourite: bool | None = None
735 ) -> Album | ItemMapping | None:
736 """Parse album object to generic layout."""
737 relationships = album_obj.get("relationships", {})
738 response_type = album_obj.get("type")
739 if (
740 response_type == "library-albums"
741 and relationships["catalog"]["data"] != []
742 and "attributes" in relationships["catalog"]["data"][0]
743 ):
744 album_id = relationships.get("catalog", {})["data"][0]["id"]
745 attributes = relationships.get("catalog", {})["data"][0]["attributes"]
746 elif "attributes" in album_obj:
747 album_id = album_obj["id"]
748 attributes = album_obj["attributes"]
749 else:
750 album_id = album_obj["id"]
751 # No more details available other than the id, return an ItemMapping
752 return ItemMapping(
753 media_type=MediaType.ALBUM,
754 provider=self.instance_id,
755 item_id=album_id,
756 name=album_id,
757 )
758 is_available_in_catalog = attributes.get("url") is not None
759 if not is_available_in_catalog:
760 self.logger.debug(
761 "Skipping album %s. Album is not available in the Apple Music catalog.",
762 attributes.get("name"),
763 )
764 return None
765 name, version = parse_title_and_version(attributes["name"])
766 album = Album(
767 item_id=album_id,
768 provider=self.domain,
769 name=name,
770 version=version,
771 provider_mappings={
772 ProviderMapping(
773 item_id=album_id,
774 provider_domain=self.domain,
775 provider_instance=self.instance_id,
776 url=attributes.get("url"),
777 available=attributes.get("playParams", {}).get("id") is not None,
778 )
779 },
780 )
781 if artists := relationships.get("artists"):
782 album.artists = UniqueList([self._parse_artist(artist) for artist in artists["data"]])
783 elif artist_name := attributes.get("artistName"):
784 album.artists = UniqueList(
785 [
786 ItemMapping(
787 media_type=MediaType.ARTIST,
788 provider=self.instance_id,
789 item_id=artist_name,
790 name=artist_name,
791 )
792 ]
793 )
794 if release_date := attributes.get("releaseDate"):
795 album.year = int(release_date.split("-")[0])
796 if genres := attributes.get("genreNames"):
797 album.metadata.genres = set(genres)
798 if artwork := attributes.get("artwork"):
799 album.metadata.add_image(
800 MediaItemImage(
801 provider=self.instance_id,
802 type=ImageType.THUMB,
803 path=artwork["url"].format(w=artwork["width"], h=artwork["height"]),
804 remotely_accessible=True,
805 )
806 )
807 if album_copyright := attributes.get("copyright"):
808 album.metadata.copyright = album_copyright
809 if record_label := attributes.get("recordLabel"):
810 album.metadata.label = record_label
811 if upc := attributes.get("upc"):
812 album.external_ids.add((ExternalID.BARCODE, "0" + upc))
813 if notes := attributes.get("editorialNotes"):
814 album.metadata.description = notes.get("standard") or notes.get("short")
815 if content_rating := attributes.get("contentRating"):
816 album.metadata.explicit = content_rating == "explicit"
817 album_type = AlbumType.ALBUM
818 if attributes.get("isSingle"):
819 album_type = AlbumType.SINGLE
820 elif attributes.get("isCompilation"):
821 album_type = AlbumType.COMPILATION
822 album.album_type = album_type
823
824 # Try inference - override if it finds something more specific
825 # Apple Music doesn't seem to have version field
826 inferred_type = infer_album_type(album.name, "")
827 if inferred_type in (AlbumType.SOUNDTRACK, AlbumType.LIVE):
828 album.album_type = inferred_type
829 album.favorite = is_favourite or False
830 return album
831
832 def _parse_track(
833 self,
834 track_obj: dict[str, Any],
835 is_favourite: bool | None = None,
836 ) -> Track:
837 """Parse track object to generic layout."""
838 relationships = track_obj.get("relationships", {})
839 if (
840 track_obj.get("type") == "library-songs"
841 and relationships.get("catalog", {}).get("data", []) != []
842 ):
843 # Library track with catalog version available
844 track_id = relationships.get("catalog", {})["data"][0]["id"]
845 attributes = relationships.get("catalog", {})["data"][0]["attributes"]
846 elif "attributes" in track_obj:
847 # Catalog track or library-only track
848 track_id = track_obj["id"]
849 attributes = track_obj["attributes"]
850 else:
851 track_id = track_obj["id"]
852 attributes = {}
853 name, version = parse_title_and_version(attributes.get("name", ""))
854 track = Track(
855 item_id=track_id,
856 provider=self.domain,
857 name=name,
858 version=version,
859 duration=attributes.get("durationInMillis", 0) / 1000,
860 provider_mappings={
861 ProviderMapping(
862 item_id=track_id,
863 provider_domain=self.domain,
864 provider_instance=self.instance_id,
865 audio_format=AudioFormat(content_type=ContentType.AAC),
866 url=attributes.get("url"),
867 available=attributes.get("playParams", {}).get("id") is not None,
868 )
869 },
870 )
871 if disc_number := attributes.get("discNumber"):
872 track.disc_number = disc_number
873 if track_number := attributes.get("trackNumber"):
874 track.track_number = track_number
875 # Prefer catalog information over library information for artists.
876 # For compilations it picks the wrong artists
877 if "artists" in relationships:
878 artists = relationships["artists"]
879 track.artists = [self._parse_artist(artist) for artist in artists["data"]]
880 # 'Similar tracks' do not provide full artist details
881 elif artist_name := attributes.get("artistName"):
882 track.artists = [
883 ItemMapping(
884 media_type=MediaType.ARTIST,
885 item_id=artist_name,
886 provider=self.instance_id,
887 name=artist_name,
888 )
889 ]
890 if albums := relationships.get("albums"):
891 if "data" in albums and len(albums["data"]) > 0:
892 track.album = self._parse_album(albums["data"][0])
893 if artwork := attributes.get("artwork"):
894 track.metadata.add_image(
895 MediaItemImage(
896 provider=self.instance_id,
897 type=ImageType.THUMB,
898 path=artwork["url"].format(w=artwork["width"], h=artwork["height"]),
899 remotely_accessible=True,
900 )
901 )
902 if genres := attributes.get("genreNames"):
903 track.metadata.genres = set(genres)
904 if composers := attributes.get("composerName"):
905 track.metadata.performers = set(composers.split(", "))
906 if isrc := attributes.get("isrc"):
907 track.external_ids.add((ExternalID.ISRC, isrc))
908 track.favorite = is_favourite or False
909 return track
910
911 def _parse_playlist(
912 self, playlist_obj: dict[str, Any], is_favourite: bool | None = None
913 ) -> Playlist:
914 """Parse Apple Music playlist object to generic layout."""
915 attributes = playlist_obj["attributes"]
916 playlist_id = attributes["playParams"].get("globalId") or playlist_obj["id"]
917 is_editable = attributes.get("canEdit", False)
918 playlist = Playlist(
919 item_id=playlist_id,
920 provider=self.instance_id,
921 name=attributes.get("name", UNKNOWN_PLAYLIST_NAME),
922 owner=attributes.get("curatorName", "me"),
923 provider_mappings={
924 ProviderMapping(
925 item_id=playlist_id,
926 provider_domain=self.domain,
927 provider_instance=self.instance_id,
928 url=attributes.get("url"),
929 is_unique=is_editable, # user-owned playlists are unique
930 )
931 },
932 is_editable=is_editable,
933 )
934 if artwork := attributes.get("artwork"):
935 url = artwork["url"]
936 if artwork["width"] and artwork["height"]:
937 url = url.format(w=artwork["width"], h=artwork["height"])
938 playlist.metadata.add_image(
939 MediaItemImage(
940 provider=self.instance_id,
941 type=ImageType.THUMB,
942 path=url,
943 remotely_accessible=True,
944 )
945 )
946 if description := attributes.get("description"):
947 playlist.metadata.description = description.get("standard")
948 playlist.favorite = is_favourite or False
949 return playlist
950
951 async def _get_all_items(self, endpoint, key="data", **kwargs) -> list[dict]:
952 """Get all items from a paged list."""
953 limit = 50
954 offset = 0
955 all_items = []
956 while True:
957 kwargs["limit"] = limit
958 kwargs["offset"] = offset
959 result = await self._get_data(endpoint, **kwargs)
960 if key not in result:
961 break
962 all_items += result[key]
963 if not result.get("next"):
964 break
965 offset += limit
966 return all_items
967
968 @throttle_with_retries
969 async def _get_data(self, endpoint, **kwargs) -> dict[str, Any]:
970 """Get data from api."""
971 url = f"https://api.music.apple.com/v1/{endpoint}"
972 headers = {"Authorization": f"Bearer {self._music_app_token}"}
973 headers["Music-User-Token"] = self._music_user_token
974 async with (
975 self.mass.http_session.get(
976 url, headers=headers, params=kwargs, ssl=True, timeout=120
977 ) as response,
978 ):
979 if response.status == 404 and "limit" in kwargs and "offset" in kwargs:
980 return {}
981 # Convert HTTP errors to exceptions
982 if response.status == 404:
983 raise MediaNotFoundError(f"{endpoint} not found")
984 if response.status == 504:
985 # See if we can get more info from the response on occasional timeouts
986 self.logger.debug(
987 "Apple Music API Timeout: url=%s, params=%s, response_headers=%s",
988 url,
989 kwargs,
990 response.headers,
991 )
992 raise ResourceTemporarilyUnavailable("Apple Music API Timeout")
993 if response.status == 429:
994 # Debug this for now to see if the response headers give us info about the
995 # backoff time. There is no documentation on this.
996 self.logger.debug("Apple Music Rate Limiter. Headers: %s", response.headers)
997 raise ResourceTemporarilyUnavailable("Apple Music Rate Limiter")
998 if response.status == 500:
999 raise MusicAssistantError("Unexpected server error when calling Apple Music")
1000 response.raise_for_status()
1001 return await response.json(loads=json_loads)
1002
1003 @throttle_with_retries
1004 async def _delete_data(self, endpoint, data=None, **kwargs) -> None:
1005 """Delete data from api."""
1006 url = f"https://api.music.apple.com/v1/{endpoint}"
1007 headers = {"Authorization": f"Bearer {self._music_app_token}"}
1008 headers["Music-User-Token"] = self._music_user_token
1009 async with (
1010 self.mass.http_session.delete(
1011 url, headers=headers, params=kwargs, json=data, ssl=True, timeout=120
1012 ) as response,
1013 ):
1014 # Convert HTTP errors to exceptions
1015 if response.status == 404:
1016 raise MediaNotFoundError(f"{endpoint} not found")
1017 if response.status == 429:
1018 # Debug this for now to see if the response headers give us info about the
1019 # backoff time. There is no documentation on this.
1020 self.logger.debug("Apple Music Rate Limiter. Headers: %s", response.headers)
1021 raise ResourceTemporarilyUnavailable("Apple Music Rate Limiter")
1022 response.raise_for_status()
1023
1024 async def _put_data(self, endpoint, data=None, **kwargs) -> str:
1025 """Put data on api."""
1026 url = f"https://api.music.apple.com/v1/{endpoint}"
1027 headers = {"Authorization": f"Bearer {self._music_app_token}"}
1028 headers["Music-User-Token"] = self._music_user_token
1029 async with (
1030 self.mass.http_session.put(
1031 url, headers=headers, params=kwargs, json=data, ssl=True, timeout=120
1032 ) as response,
1033 ):
1034 # Convert HTTP errors to exceptions
1035 if response.status == 404:
1036 raise MediaNotFoundError(f"{endpoint} not found")
1037 if response.status == 429:
1038 # Debug this for now to see if the response headers give us info about the
1039 # backoff time. There is no documentation on this.
1040 self.logger.debug("Apple Music Rate Limiter. Headers: %s", response.headers)
1041 raise ResourceTemporarilyUnavailable("Apple Music Rate Limiter")
1042 response.raise_for_status()
1043 if response.content_length:
1044 return await response.json(loads=json_loads)
1045 return {}
1046
1047 @throttle_with_retries
1048 async def _post_data(self, endpoint, data=None, **kwargs) -> str:
1049 """Post data on api."""
1050 url = f"https://api.music.apple.com/v1/{endpoint}"
1051 headers = {"Authorization": f"Bearer {self._music_app_token}"}
1052 headers["Music-User-Token"] = self._music_user_token
1053 async with (
1054 self.mass.http_session.post(
1055 url, headers=headers, params=kwargs, json=data, ssl=True, timeout=120
1056 ) as response,
1057 ):
1058 # Convert HTTP errors to exceptions
1059 if response.status == 404:
1060 raise MediaNotFoundError(f"{endpoint} not found")
1061 if response.status == 429:
1062 # Debug this for now to see if the response headers give us info about the
1063 # backoff time. There is no documentation on this.
1064 self.logger.debug("Apple Music Rate Limiter. Headers: %s", response.headers)
1065 raise ResourceTemporarilyUnavailable("Apple Music Rate Limiter")
1066 response.raise_for_status()
1067 return await response.json(loads=json_loads)
1068
1069 async def _get_user_storefront(self) -> str:
1070 """Get the user's storefront."""
1071 locale = self.mass.metadata.locale.replace("_", "-")
1072 language = locale.split("-")[0]
1073 result = await self._get_data("me/storefront", l=language)
1074 return result["data"][0]["id"]
1075
1076 async def _get_ratings(self, item_ids: list[str], media_type: MediaType) -> dict[str, bool]:
1077 """Get ratings (aka favorites) for a list of item ids."""
1078 if media_type == MediaType.ARTIST:
1079 raise NotImplementedError(
1080 "Ratings are not available for artist in the Apple Music API."
1081 )
1082 if len(item_ids) == 0:
1083 return {}
1084 apple_type = self._translate_media_type_to_apple_type(media_type)
1085 endpoint = apple_type if not self.is_library_id(item_ids[0]) else f"library-{apple_type}"
1086 # Apple Music limits to 200 ids per request
1087 max_ids_per_request = 200
1088 results = {}
1089 for i in range(0, len(item_ids), max_ids_per_request):
1090 batch_ids = item_ids[i : i + max_ids_per_request]
1091 response = await self._get_data(
1092 f"me/ratings/{endpoint}",
1093 ids=",".join(batch_ids),
1094 )
1095 results.update(
1096 {
1097 item["id"]: bool(item["attributes"].get("value", False) == 1)
1098 for item in response.get("data", [])
1099 }
1100 )
1101 return results
1102
1103 def _translate_media_type_to_apple_type(self, media_type: MediaType) -> str:
1104 """Translate MediaType to Apple Music endpoint string."""
1105 match media_type:
1106 case MediaType.ARTIST:
1107 return "artists"
1108 case MediaType.ALBUM:
1109 return "albums"
1110 case MediaType.TRACK:
1111 return "songs"
1112 case MediaType.PLAYLIST:
1113 return "playlists"
1114 raise MusicAssistantError(f"Unsupported media type: {media_type}")
1115
1116 def is_library_id(self, library_id) -> bool:
1117 """Check a library ID matches known format."""
1118 if not isinstance(library_id, str):
1119 return False
1120 valid = re.findall(r"^(?:[a|i|l|p]{1}\.|pl\.u\-)[a-zA-Z0-9]+$", library_id)
1121 return bool(valid)
1122
1123 def _is_catalog_id(self, catalog_id: str) -> bool:
1124 """Check if input is a catalog id, or a library id."""
1125 return catalog_id.isnumeric() or catalog_id.startswith("pl.")
1126
1127 async def _fetch_song_stream_metadata(self, song_id: str) -> str:
1128 """Get the stream URL for a song from Apple Music."""
1129 playback_url = "https://play.music.apple.com/WebObjects/MZPlay.woa/wa/webPlayback"
1130 data = {}
1131 self.logger.debug("_fetch_song_stream_metadata: Check if Library ID: %s", song_id)
1132 if self.is_library_id(song_id):
1133 data["universalLibraryId"] = song_id
1134 data["isLibrary"] = True
1135 else:
1136 data["salableAdamId"] = song_id
1137 for retry in (True, False):
1138 try:
1139 async with self.mass.http_session.post(
1140 playback_url, headers=self._get_decryption_headers(), json=data, ssl=True
1141 ) as response:
1142 response.raise_for_status()
1143 content = await response.json(loads=json_loads)
1144 if content.get("failureType"):
1145 message = content.get("failureMessage")
1146 raise MediaNotFoundError(f"Failed to get song stream metadata: {message}")
1147 return content["songList"][0]
1148 except (MediaNotFoundError, ClientError) as exc:
1149 if retry:
1150 self.logger.warning("Failed to get song stream metadata: %s", exc)
1151 continue
1152 raise
1153 raise MediaNotFoundError(f"Failed to get song stream metadata for {song_id}")
1154
1155 async def _parse_stream_url_and_uri(self, stream_assets: list[dict]) -> str:
1156 """Parse the Stream URL and Key URI from the song."""
1157 ctrp256_urls = [asset["URL"] for asset in stream_assets if asset["flavor"] == "28:ctrp256"]
1158 if len(ctrp256_urls) == 0:
1159 raise MediaNotFoundError("No ctrp256 URL found for song.")
1160 playlist_url = ctrp256_urls[0]
1161 playlist_items = await fetch_playlist(self.mass, ctrp256_urls[0], raise_on_hls=False)
1162 # Apple returns a HLS (substream) playlist but instead of chunks,
1163 # each item is just the whole file. So we simply grab the first playlist item.
1164 playlist_item = playlist_items[0]
1165 # path is relative, stitch it together
1166 base_path = playlist_url.rsplit("/", 1)[0]
1167 track_url = base_path + "/" + playlist_items[0].path
1168 key = playlist_item.key
1169 return (track_url, key)
1170
1171 def _get_decryption_headers(self):
1172 """Get headers for decryption requests."""
1173 return {
1174 "authorization": f"Bearer {self._music_app_token}",
1175 "media-user-token": self._music_user_token,
1176 "connection": "keep-alive",
1177 "accept": "application/json",
1178 "origin": "https://music.apple.com",
1179 "referer": "https://music.apple.com/",
1180 "accept-encoding": "gzip, deflate, br",
1181 "content-type": "application/json;charset=utf-8",
1182 "user-agent": (
1183 "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko)"
1184 " Chrome/110.0.0.0 Safari/537.36"
1185 ),
1186 }
1187
1188 async def _get_decryption_key(
1189 self, license_url: str, key_id: bytes, uri: str, item_id: str
1190 ) -> str:
1191 """Get the decryption key for a song."""
1192 if decryption_key := await self.mass.cache.get(
1193 key=item_id,
1194 provider=self.instance_id,
1195 category=CACHE_CATEGORY_DECRYPT_KEY,
1196 checksum=self._session_id,
1197 ):
1198 self.logger.debug("Decryption key for %s found in cache.", item_id)
1199 return decryption_key
1200 pssh = self._get_pssh(key_id)
1201 device = Device(
1202 client_id=self._decrypt_client_id,
1203 private_key=self._decrypt_private_key,
1204 type_=DeviceTypes.ANDROID,
1205 security_level=3,
1206 flags={},
1207 )
1208 cdm = Cdm.from_device(device)
1209 session_id = cdm.open()
1210 challenge = cdm.get_license_challenge(session_id, pssh)
1211 track_license = await self._get_license(challenge, license_url, uri, item_id)
1212 cdm.parse_license(session_id, track_license)
1213 key = next(key for key in cdm.get_keys(session_id) if key.type == "CONTENT")
1214 if not key:
1215 raise MediaNotFoundError("Unable to get decryption key for song %s.", item_id)
1216 cdm.close(session_id)
1217 decryption_key = key.key.hex()
1218 self.mass.create_task(
1219 self.mass.cache.set(
1220 key=item_id,
1221 data=decryption_key,
1222 expiration=3600,
1223 provider=self.instance_id,
1224 category=CACHE_CATEGORY_DECRYPT_KEY,
1225 checksum=self._session_id,
1226 )
1227 )
1228 return decryption_key
1229
1230 def _get_pssh(self, key_id: bytes) -> PSSH:
1231 """Get the PSSH for a song."""
1232 pssh_data = WidevinePsshData()
1233 pssh_data.algorithm = 1
1234 pssh_data.key_ids.append(key_id)
1235 init_data = base64.b64encode(pssh_data.SerializeToString()).decode("utf-8")
1236 return PSSH.new(system_id=PSSH.SystemId.Widevine, init_data=init_data)
1237
1238 async def _get_license(self, challenge: bytes, license_url: str, uri: str, item_id: str) -> str:
1239 """Get the license for a song based on the challenge."""
1240 challenge_b64 = base64.b64encode(challenge).decode("utf-8")
1241 data = {
1242 "challenge": challenge_b64,
1243 "key-system": "com.widevine.alpha",
1244 "uri": uri,
1245 "adamId": item_id,
1246 "isLibrary": False,
1247 "user-initiated": True,
1248 }
1249 async with self.mass.http_session.post(
1250 license_url, data=json.dumps(data), headers=self._get_decryption_headers(), ssl=False
1251 ) as response:
1252 response.raise_for_status()
1253 content = await response.json(loads=json_loads)
1254 track_license = content.get("license")
1255 if not track_license:
1256 raise MediaNotFoundError("No license found for song %s.", item_id)
1257 return track_license
1258