"""Spotify related helpers.""" import base64 from oto import response import requests from requests import exceptions as requests_exceptions from analytics import config from analytics.connectors import sentry from analytics.consts import models as model_consts metadata_url_template = ( 'https://api.spotify.com/v1/playlists/{playlist_id}' '?fields=images,followers(total),' 'name,owner(display_name,id),external_urls.spotify' ) def _to_base64(s): """Encode string to base64 and return result as string. Args: s (str): string which needs to be encoded. Returns: str: result of encoding. """ return base64.b64encode(s.encode()).decode() def _obtain_access_token(): """Obtain access token for further Spotify API calls. Returns: str: access token. """ auth_token = _to_base64('{}:{}'.format( config.SPOTIFY_CLIENT_ID, config.SPOTIFY_SECRET_ID)) resp = requests.post( 'https://accounts.spotify.com/api/token', data={'grant_type': 'client_credentials'}, headers={ 'Authorization': 'Basic {}'.format(auth_token)}).json() return resp['access_token'] def _auth_headers(access_token): """Form Spotify API headers. Args: access_token (str): access token for Spotify API calls. Returns: dict: headers dictionary. """ return {'Authorization': 'Bearer {}'.format(access_token)} def _spotify_api_playlist_data(playlist_link, response_data): """Extract API response data and generate dict with result. Args: playlist_link (str): Playlist link. response_data (dict): Spotify API response data. Returns: dict: Playlist metadata. """ owner = response_data['owner'] if response_data['images']: image = response_data['images'][0]['url'] else: image = None return { 'playlist_link': playlist_link, 'playlist_title': response_data['name'], 'playlist_author': owner['display_name'] or owner['id'], 'display_link': response_data['external_urls']['spotify'], 'image': image, 'followers': response_data['followers']['total']} def _get_fail_details(playlist_link, api_url, api_response): """Generate failure details. Args: playlist_link (str): Playlist link. api_url (str): Spotify API url. api_response (requests.Response): Response object. Returns: dict: Formatted fail details. """ return { 'playlist_link': playlist_link, 'request_url': api_url, 'retry-after': api_response.headers.get('retry-after'), 'response_reason': api_response.reason, 'response_status_code': api_response.status_code, 'response_text': api_response.text, 'response_url': api_response.url} def _playlist_link_with_status(playlist_link, status_code): """Add status code to playlist link. Args: playlist_link (str): Playlist URL. status_code (int): HTTP status code. Returns: dict: Dict with playlist_link and status_code values. """ return { 'playlist_link': playlist_link, 'status_code': status_code} def _get_playlist_metadata( playlist_link, playlist_id, headers, logger): """Get Spotify playlist metadata by playlist URL. Function handles Spotify API response and generates dict result whether it was successful or failed. Args: playlist_link (str): Spotify playlist link. headers (dict): Required headers for API call. logger (OwsLoggingAdapter): Application logger. Returns: dict: Playlist metadata. Empty dict if requests failed. """ api_url = metadata_url_template.format(playlist_id=playlist_id) api_response_status_code = 500 try: api_response = requests.get(api_url, headers=headers, timeout=3) api_response_status_code = api_response.status_code if api_response_status_code == 200: playlist_metadata = _spotify_api_playlist_data( playlist_link, api_response.json()) playlist_metadata['status_code'] = api_response_status_code return playlist_metadata if api_response_status_code == 404: logger.info( "Playlist wasn't found. " 'Playlist link: {}'.format(playlist_link)) return _playlist_link_with_status( playlist_link, api_response_status_code) fail_details = _get_fail_details( playlist_link, api_url, api_response) if api_response_status_code == 429: sentry.sentry_client.captureMessage( 'Spotify API limit was reached.', extra=fail_details) else: sentry.sentry_client.captureMessage( 'Failed to obtain Spotify playlist metadata.', extra=fail_details) return _playlist_link_with_status( playlist_link, api_response_status_code) except Exception as e: if isinstance(e, requests_exceptions.Timeout): api_response_status_code = 504 sentry.sentry_client.captureException() return _playlist_link_with_status( playlist_link, api_response_status_code) def get_personalized_playlists_metadata( playlists, logger, feed_ids, distributors): """Get metadata for each personalized playlist. Args: playlists (list): list of playlist. logger (OwsLoggingAdapter): Application logger. It is passed explicitly since flask `g` object is not available in spawned threads. feed_ids (list): list of available feed_ids. distributors (list): list of distributors names. Returns: Response: list of dict of playlist_link and related image and followers. """ result = [] if playlists: for playlist in playlists: if playlist['playlist_id'] in model_consts.SPOTIFY_PLAYLISTS: playlist_id = playlist['playlist_id'] else: playlist_id = 'unknown' playlist_data = model_consts.SPOTIFY_PLAYLISTS[playlist_id] playlist_data['playlist_author'] = 'Spotify' playlist_data['playlist_link'] = playlist['playlist_link'] playlist_data['status_code'] = 200 result.append(playlist_data) return response.Response(result) def get_playlists_metadata(playlists, logger, feed_ids, distributors): """Get metadata for each playlist. Args: playlists (list): list of playlist. logger (OwsLoggingAdapter): Application logger. It is passed explicitly since flask `g` object is not available in spawned threads. distributors (list): list of distributors names. Returns: Response: list of dict of playlist_link and related image and followers. """ result = [] if playlists: access_token = _obtain_access_token() headers = _auth_headers(access_token) for playlist in playlists: metadata = _get_playlist_metadata( playlist['playlist_link'], playlist['playlist_id'], headers, logger) result.append(metadata) return response.Response(result)