from time import time from core_images.cached_either import Right, Left from requests.exceptions import HTTPError from core_images.cache import ServiceCacheMixin from core_images.dsp_clients import get_apple_music_client from core_images.consts import ( NO_IMAGE_CACHE_VALUE, SMALL_RESOLUTION, MEDIUM_RESOLUTION, LARGE_RESOLUTION, ) class AppleMusicService(ServiceCacheMixin): """Encapsulates access to the Apple Music API.""" NO_IMAGE_MSG = "No image available" RESOLUTION_MAPPING = { SMALL_RESOLUTION: 160, MEDIUM_RESOLUTION: 320, LARGE_RESOLUTION: 640, } def __init__(self, client): self._client = client self.original_init = time() self.init_apple_music_client() def init_apple_music_client(self): self.current_init = time() def playlist_artwork_url( self, playlist_id, resolution, storefront, tries=0, ): """applemusicpy doesn’t automatically update token when it expires, and doesn’t offer a way to update the token either. """ cache_key = [ "playlist_artwork_url", playlist_id, resolution, storefront, ] url = self.cache_get(cache_key) if url == NO_IMAGE_CACHE_VALUE: return Left(self.NO_IMAGE_MSG, True) elif url: return Right(url, True) try: results = self._client.playlist(playlist_id, storefront) url = self._get_image(results, resolution) if results else None except HTTPError as e: if e.response.status_code == 401 and tries == 0: self.init_apple_music_client() url = self.playlist_artwork_url(playlist_id, resolution, 1) else: self.cache_set(cache_key, NO_IMAGE_CACHE_VALUE) return Left( "Unable to locate an Apple Music playlist with ID %s: %s. " "Using JWT [%s] at (%d, %d, %d)." # noqa % ( playlist_id, e, self._client.token_str, self.original_init, self.current_init, time(), ) ) if not url: self.cache_set(cache_key, NO_IMAGE_CACHE_VALUE) return Left("That Apple Music playlist has no associated artwork.") self.cache_set(cache_key, url) return Right(url) def album_artwork_url(self, album_id, resolution, storefront, tries=0): """applemusicpy doesn’t automatically update token when it expires, and doesn’t offer a way to update the token either. """ cache_key = ["album_artwork_url", album_id, resolution, storefront] url = self.cache_get(cache_key) if url == NO_IMAGE_CACHE_VALUE: return Left(self.NO_IMAGE_MSG, True) elif url: return Right(url, True) try: results = self._client.album(album_id, storefront) url = self._get_image(results, resolution) if results else None except HTTPError as e: if e.response.status_code == 401 and tries == 0: self.init_apple_music_client() url = self.album_artwork_url(album_id, resolution, 1) else: self.cache_set(cache_key, NO_IMAGE_CACHE_VALUE) return Left( "Unable to locate an Apple Music album with ID %s: %s. " "Using JWT [%s] at (%d, %d, %d)." # noqa % ( album_id, e, self._client.token_str, self.original_init, self.current_init, time(), ) ) if not url: self.cache_set(cache_key, NO_IMAGE_CACHE_VALUE) return Left("That Apple Music album has no associated artwork.") self.cache_set(cache_key, url) return Right(url) def track_artwork_url(self, track_id, resolution, storefront, tries=0): """applemusicpy doesn’t automatically update token when it expires, and doesn’t offer a way to update the token either. """ cache_key = ["track_artwork_url", track_id, resolution, storefront] url = self.cache_get(cache_key) if url == NO_IMAGE_CACHE_VALUE: return Left(self.NO_IMAGE_MSG, True) elif url: return Right(url, True) try: results = self._client.song(track_id, storefront) url = self._get_image(results, resolution) if results else None except HTTPError as e: if e.response.status_code == 401 and tries == 0: self.init_apple_music_client() url = self.track_artwork_url( track_id, resolution, storefront, 1, ) else: self.cache_set(cache_key, NO_IMAGE_CACHE_VALUE) return Left( "Unable to locate an Apple Music track with ID %s: %s. " "Using JWT [%s] at (%d, %d, %d)." # noqa % ( track_id, e, self._client.token_str, self.original_init, self.current_init, time(), ) ) if not url: self.cache_set(cache_key, NO_IMAGE_CACHE_VALUE) return Left("That Apple Music track has no associated artwork.") self.cache_set(cache_key, url) return Right(url) def _get_image(self, results, resolution): attributes = results["data"][0]["attributes"] if "artwork" in attributes: size = self.RESOLUTION_MAPPING.get(resolution) url = attributes["artwork"]["url"] return url.replace("{w}", str(size)).replace("{h}", str(size)) def get_apple_music_service(): client = get_apple_music_client() return AppleMusicService( client=client, )