import time from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple, Union import aiohttp import jwt import config from server.apple.constants import US_MARKET, SongIdType from server.apple.exceptions import AppleError from server.apple.utils import format_image_url_with_size, get_image_url, get_isrc from server.core.cache import ListResult, SingleResult, cache_data, cache_request from server.core.cache.decorators import get_cached, set_cached from server.core.constants import CollectionName from server.core.utils import get_id, handle_requests, make_request, request_in_chunks, retry class AppleMusic: """This class is used for Apple Music API""" def __init__(self, session: aiohttp.ClientSession, token_ttl: int = 24): """Init apple music client. Args: session: Instance of aiohttp.ClientSession. token_ttl: Time to live for token in hours. """ self.algorithm = "ES256" self.token = None self.url = config.APPLE_MUSICKIT_BASE_URL self.key_id = config.APPLE_MUSICKIT_KEYID self.team_id = config.APPLE_MUSICKIT_TEAM_ID self.token_ttl = token_ttl self.secret_key = config.APPLE_MUSICKIT_KEY self.token_expired_at = None self.session = session def generate_token(self) -> None: """Generate encrypted token to be used by in API requests.""" headers = {"alg": self.algorithm, "kid": self.key_id} self.token_expired_at = int((datetime.now() + timedelta(hours=self.token_ttl)).timestamp()) payload = {"iss": self.team_id, "iat": int(time.time()), "exp": self.token_expired_at} self.token = jwt.encode(payload, self.secret_key, algorithm=self.algorithm, headers=headers) def _auth_headers(self) -> Dict[str, str]: """Get header for API request""" if not self.token or self.token_expired_at <= int(time.time()): self.generate_token() return {"Authorization": "Bearer {}".format(self.token)} async def _make_request(self, url: str, **kwargs) -> dict: """Make http request to Apple Music API""" url = f"{self.url}/{url}" equivalent_ids = kwargs.pop("equivalent_ids", None) # handle 'filter[equivalents]' which passed by inner name if equivalent_ids: kwargs["filter[equivalents]"] = equivalent_ids params = {k: ",".join(v) if isinstance(v, list) else v for k, v in kwargs.items() if v is not None} localization = params.pop("localization", None) if localization: params["l"] = localization headers = self._auth_headers() headers["Content-Type"] = "application/json" return await make_request(self.session, url, params=params, headers=headers, error_cls=AppleError) @retry() async def _get(self, url: str, **kwargs) -> dict: """Make http request to Apple Music API""" return await self._make_request(url, **kwargs) @cache_request(collection_name=CollectionName.APPLE_CACHE) async def cached_get(self, url: str, **kwargs) -> dict: return await self._get(url, **kwargs) @cache_data(result_handler=SingleResult(CollectionName.APPLE_SONG, result_items_key="data")) async def song( self, song_id: int, storefront: str = US_MARKET, localization: Optional[str] = None, include: Optional[dict] = None, ) -> dict: """Get a catalog Song by ID. Args: song_id: Song ID. storefront: Apple Music Storefront. localization: The localization, specified by a language tag. include: Additional relationships to include in the fetch. Returns: Song data in JSON format. """ url_path = f"v1/catalog/{storefront}/songs/{song_id}" return await self._get(url_path, localization=localization, include=include) @cache_data(result_handler=ListResult(CollectionName.APPLE_SONG, result_items_key="data")) @request_in_chunks(25) async def songs_by_id( self, song_ids: List[str], storefront: str = US_MARKET, localization: Optional[str] = None, include: Optional[dict] = None, ) -> dict: """Get all catalog song data associated with the IDs provided. Args: song_ids: a list of song IDs storefront: Apple Music store front localization: The localization, specified by a language tag. include: Additional relationships to include in the fetch. Returns: A list of catalog song data in JSON format. """ url_path = f"v1/catalog/{storefront}/songs" id_str = ",".join(song_ids) return await self._get(url_path, ids=id_str, localization=localization, include=include) @request_in_chunks(300) async def _songs_by_equivalent_id( self, song_ids: List[str], storefront: str = US_MARKET, localization: Optional[str] = None, include: Optional[dict] = None, ) -> dict: """Get catalog song data for the particular storefront associated with the equivalent IDs provided. Args: song_ids: a list of known song IDs (any storefront) storefront: Apple Music store front (desired storefront to get data in) localization: The localization, specified by a language tag. include: Additional relationships to include in the fetch. Returns: A list of catalog song data in JSON format. """ url_path = f"v1/catalog/{storefront}/songs" id_str = ",".join(song_ids) _result = await self._get(url_path, equivalent_ids=id_str, localization=localization, include=include) result = {} for k, items in result.get("meta", {}).get("filters", {}).get("equivalents", {}).items(): v = items[0] v["known_id"] = k result[k] = v return result async def songs_by_equivalent_id( self, song_ids: List[str], storefront: str = US_MARKET, localization: Optional[str] = None, include: Optional[dict] = None, equivalent_cache_handler=ListResult( collection_name=CollectionName.APPLE_SONG_EQUIVALENT_ID, key_getter=lambda x: x["known_id"] ), meta_cache_handler=ListResult( key_options=( (CollectionName.APPLE_SONG_ISRC, get_isrc), (CollectionName.APPLE_SONG, get_id), ), result_items_key="data", ), ) -> Tuple[dict, dict]: equivalent_songs_key_kwargs = {"storefront": storefront} # get known_id to equivalent_id items map from cache cached_meta, missing_song_ids = await get_cached( id_list=song_ids, collection_name=equivalent_cache_handler.collection_name, key_kwargs=equivalent_songs_key_kwargs, ) cached_equivalent_map = cached_meta.get("data", {}) # known_id to item with id, known_id, type, href attributes equivalent_songs_with_data, songs_data = await handle_requests( ( ( # get equivalent id items & songs meta from api for those known ids, that were not found in cache self._songs_by_equivalent_id, (missing_song_ids,), { "storefront": storefront, "localization": localization, "include": include, }, {}, missing_song_ids, ), ( # get songs meta from api/cache for those known ids, that were found in cache self.songs_by_id, { "song_ids": [s["id"] for s in cached_equivalent_map.values()], "storefront": storefront, "localization": localization, "include": include, }, {}, cached_equivalent_map, ), ) ) new_equivalent_map = equivalent_songs_with_data.pop("meta", {}).get("filters", {}).get("equivalents", {}) await handle_requests( ( ( # save to cache new equivalence mapping set_cached, { "result": list(new_equivalent_map.values()), "result_handler": equivalent_cache_handler, "key_kwargs": equivalent_songs_key_kwargs, }, [], new_equivalent_map, ), ( # save to cache new songs meta set_cached, { "result": equivalent_songs_with_data, "result_handler": meta_cache_handler, "key_kwargs": {"storefront": storefront, "localization": localization, "include": include}, }, {}, equivalent_songs_with_data.get("data"), ), ) ) return {"data": equivalent_songs_with_data.get("data", []) + songs_data.get("data", [])}, { **cached_equivalent_map, **new_equivalent_map, } @cache_data( result_handler=ListResult( key_options=((CollectionName.APPLE_SONG_ISRC, get_isrc), (CollectionName.APPLE_SONG, get_id)), result_items_key="data", ) ) @request_in_chunks(25) async def songs_by_isrc( self, isrc_list: List[str], storefront: str = US_MARKET, localization: Optional[str] = None, include: Optional[dict] = None, ) -> dict: """Get all catalog songs associated with the ISRCs provided. Args: isrc_list: list of ISRCs. storefront: Apple Music store front. localization: The localization, specified by a language tag. include: Additional relationships to include in the fetch. Returns: A list of catalog song data in JSON format. """ url_path = f"v1/catalog/{storefront}/songs" params = {"filter[isrc]": ",".join(isrc_list)} return await self._get(url_path, localization=localization, include=include, **params) async def songs( self, ids: List[str], id_type: SongIdType, storefront: str = US_MARKET, localization: Optional[str] = None, include: Optional[dict] = None, ) -> dict: handler = self.songs_by_id if id_type == SongIdType.ID else self.songs_by_isrc return await handler(ids, storefront=storefront, localization=localization, include=include) async def songs_images( self, ids: List[str], id_type: SongIdType, storefront: str = US_MARKET, localization: Optional[str] = None, include: Optional[dict] = None, image_size: int = None, ) -> List[dict]: """Get catalog song images urls associated with the identifiers provided. Args: ids: A list of song identifiers. id_type: Song identifier type. storefront: Apple Music store front. localization: The localization, specified by a language tag. include: Additional relationships to include in the fetch. image_size: required image size in pixels (used in get_with_size_decorator). Returns: A list of catalog song image urls in JSON format. """ songs_data = await self.songs(ids, id_type, storefront=storefront, localization=localization, include=include) if songs_data: return [ { "id": get_id(song) if id_type == SongIdType.ID else get_isrc(song), "id_type": id_type.name, "image_url": get_image_url(song, image_size), } for song in songs_data.get("data", []) ] return [] @cache_data(result_handler=SingleResult(CollectionName.APPLE_PLAYLIST, result_items_key="data")) async def playlist( self, playlist_id: str, storefront: str = US_MARKET, localization: Optional[str] = None, include: Optional[dict] = None, include_songs: Optional[dict] = None, include_music_videos: Optional[dict] = None, include_library_playlists: Optional[dict] = None, ) -> dict: """Get playlist data associated with the IDs provided. Args: playlist_id: Playlist ID. storefront: Apple Music store front. localization: The localization, specified by a language tag. include: Additional relationships to include in the fetch. include_songs: Additional relationships to include in the related objects of "songs" type. include_music_videos: Additional relationships to include in the related objects of "music-videos" type. include_library_playlists: Additional relationships to include in the related objects of "library-playlists" type. Returns: Playlist data. """ url_path = f"v1/catalog/{storefront}/playlists/{playlist_id}" params = { "include[songs]": include_songs, "include[music-videos]": include_music_videos, "include[library-playlists]": include_library_playlists, } return await self._get(url_path, localization=localization, include=include, **params) @cache_data(result_handler=ListResult(CollectionName.APPLE_PLAYLIST, result_items_key="data")) @request_in_chunks(25) async def playlists( self, ids, storefront: str = US_MARKET, localization: Optional[str] = None, include: Optional[dict] = None ) -> dict: """Get playlist data associated with the IDs provided. Args: ids: list of IDs. storefront: Apple Music store front. localization: The localization, specified by a language tag. include: Additional relationships to include in the fetch. Returns: A list of catalog playlists data. """ url_path = f"v1/catalog/{storefront}/playlists" id_str = ",".join(ids) return await self._get(url_path, ids=id_str, localization=localization, include=include) async def _playlists_images( self, ids: List[str], storefront: str = US_MARKET, localization: Optional[str] = None, include: Optional[dict] = None, ) -> List[dict]: """Get url masks of songs images associated with the identifiers provided. We need it to cache url mask instead of url for particular size. """ playlists_data = await self.playlists(ids, storefront=storefront, localization=localization, include=include) if playlists_data: return [ {"id": playlist["id"], "image_url": get_image_url(playlist)} for playlist in playlists_data.get("data", []) ] return [] async def playlists_images( self, ids: List[str], storefront: str = US_MARKET, localization: Optional[str] = None, include: Optional[dict] = None, image_size: int = None, ) -> List[dict]: """Get catalog song images urls associated with the identifiers provided. Args: ids: A list of playlist IDs. storefront: Apple Music store front. localization: Localization, specified by a language tag. include: Additional relationships to include in the fetch. image_size: Image size in pixels (if None method returns url masks). Returns: A list of catalog playlist image urls in JSON format. """ playlists_data = await self._playlists_images( ids, storefront=storefront, localization=localization, include=include ) if image_size: for playlist in playlists_data: playlist["image_url"] = format_image_url_with_size(playlist["image_url"], image_size) return playlists_data @cache_data(result_handler=SingleResult(CollectionName.APPLE_ALBUM, result_items_key="data")) async def album( self, album_id: str, storefront: str = US_MARKET, localization: Optional[str] = None, include: Optional[dict] = None, ) -> dict: """Get a catalog Album by ID. Args: album_id: Album ID. storefront: Apple Music Storefront. localization: Localization, specified by a language tag. include: Additional relationships to include in the fetch. Returns: Album data in JSON format. """ url_path = f"v1/catalog/{storefront}/albums/{album_id}" return await self._get(url_path, localization=localization, include=include) @cache_data(result_handler=ListResult(CollectionName.APPLE_ALBUM, result_items_key="data")) @request_in_chunks(100) async def albums( self, album_ids: List[str], storefront: str = US_MARKET, localization: Optional[str] = None, include: Optional[dict] = None, ) -> dict: """Get a catalog Album by ID. Args: album_ids: A list of album IDs. storefront: Apple Music Storefront. localization: Localization, specified by a language tag. include: Additional relationships to include in the fetch. Returns: A list of catalog album data in JSON format. """ url_path = f"v1/catalog/{storefront}/albums" return await self._get(url_path, ids=",".join(album_ids), localization=localization, include=include) @cache_data(result_handler=SingleResult(CollectionName.APPLE_STATION, result_items_key="data")) async def station( self, station_id: str, storefront: str = US_MARKET, localization: Optional[str] = None, include: Optional[dict] = None, ) -> dict: """Get a catalog station by ID. Args: station_id: Station ID. storefront: Apple Music Storefront. localization: Localization, specified by a language tag. include: Additional relationships to include in the fetch. Returns: Station data in JSON format. """ url_path = f"v1/catalog/{storefront}/stations/{station_id}" return await self._get(url_path, localization=localization, include=include) @cache_data(result_handler=ListResult(CollectionName.APPLE_STATION, result_items_key="data")) @request_in_chunks(100) async def stations( self, station_ids: List[str], storefront: str = US_MARKET, localization: Optional[str] = None, include: Optional[dict] = None, ) -> dict: """Get catalog stations by list of ID. Args: station_ids: A list of station IDs. storefront: Apple Music Storefront. localization: Localization, specified by a language tag. include: Additional relationships to include in the fetch. Returns: A list of catalog stations data in JSON format. """ url_path = f"v1/catalog/{storefront}/stations" return await self._get(url_path, ids=",".join(station_ids), localization=localization, include=include) @cache_request(collection_name=CollectionName.APPLE_MARKETS) async def storefronts( self, localization: Optional[str] = None, include: Optional[dict] = None, extend: Optional[dict] = None ) -> dict: """Get supported storefronts. Args: localization: Localization, specified by a language tag. include: Additional relationships to include in the fetch. extend: A list of attribute extensions to apply to resources in the response. Returns: A dict of supported storefront code to data. """ url_path = f"v1/storefronts" result = (await self._get(url_path, localization=localization, include=include, extend=extend)).get("data", []) return {s["id"]: s for s in result} @cache_request(collection_name=CollectionName.APPLE_SEARCH) async def search( self, q: str, storefront: str = "", item_type: Union[List[str], str] = "track", localization: Optional[str] = None, offset: int = 0, limit: int = 10, ) -> dict: """Search tracks by query. Args: q: the search query. storefront: Apple Music Storefront. item_type: the type of item to return: 'artists', 'albums', 'tracks' or 'playlists'. localization: Localization, specified by a language tag. offset: Index of the first item to return. limit: Number of items to return. Returns: Search result. """ types = ",".join(item_type) if isinstance(item_type, list) else item_type params = {"term": q, "limit": limit, "offset": offset, "types": types, "localization": localization} return await self._get(f"v1/catalog/{storefront}/search", **params) async def check_health(self): response = await self._make_request("v1/storefronts", ids="us") return bool(response.get("data"))