"""Graphql gateway client.""" from collections import defaultdict from copy import deepcopy from gql import Client from gql import gql as gql_query from gql.transport.requests import RequestsHTTPTransport import config from src import constants from src import mock_graphql_gateway from src import utils def _replace_handles_from_graph_with_user_input( event: dict, global_participant_data: dict) -> dict: """Replace social handles founded in the knowlegde graph with user input.""" user_input_urls = { platform + '_url': event.get(platform + '_url') for platform in constants.PLATFORMS} copy_of_global_participant_data = deepcopy(global_participant_data) for key, value in global_participant_data.items(): if user_input_urls.get(key) and user_input_urls.get(key) != value: # replace platform url with user input copy_of_global_participant_data[key] = user_input_urls.get(key) # replace followers and mothly_listeners with None platform = key.rstrip('_url') copy_of_global_participant_data[platform + '_followers'] = None if platform == 'spotify': copy_of_global_participant_data['spotify_monthly_listeners'] = None copy_of_global_participant_data['spotify_id'] = utils.get_spotify_id( user_input_urls.get(key)) return copy_of_global_participant_data def _create_list_of_label_participants(result: dict, main_key: str) -> list: """Create a list of LabelParicipant uuids for the GlobalParticipants.""" label_participants = [] for gp in result.get(main_key, {}): for lp in gp.get('labelParticipants', {}).get('labelParticipants', {}): label_participants.append(lp.get('uuid')) label_participants = ','.join(label_participants) or None return label_participants def _merge_and_dedup(result: dict, main_key: str) -> dict: """Merge and dedup all social platform stats.""" # merge and dedup all social platform stats social_stats = {platform: defaultdict(dict) for platform in constants.PLATFORMS} for gp in result.get(main_key, {}): for platform_data in gp.get( 'publicParticipantChartmetric', {}).get('accountStatsV2', {}): platform = platform_data.get('platform') url = platform_data.get('url') followers = platform_data.get('followers', 0) or 0 monthly_listeners = platform_data.get('monthlyListeners', 0) or 0 previous_stats = social_stats.get(platform, {}) or {} possibly_existing_followers = previous_stats.get('followers', 0) or 0 possibly_existing_monthly_listeners = previous_stats.get( 'monthly_listeners', 0) or 0 if not possibly_existing_followers \ or possibly_existing_followers < followers: social_stats[platform].update( {'followers': followers}) social_stats[platform].update({'url': url}) if platform == 'spotify': if not possibly_existing_monthly_listeners \ or possibly_existing_monthly_listeners < monthly_listeners: social_stats[platform].update( {'monthly_listeners': monthly_listeners}) social_stats[platform].update({'url': url}) return social_stats def _create_global_participant_data_result( social_stats: dict, label_participants: list, spotify_id: str) -> dict: """Create a final result containing all the required data.""" global_participant_data = {} global_participant_data.update({ 'spotify_id': spotify_id, 'label_participants': label_participants }) for platform in constants.PLATFORMS: global_participant_data.update({ f'{platform}_url': social_stats[platform].get('url'), f'{platform}_followers': social_stats[platform].get('followers') }) if platform == 'spotify': global_participant_data.update({ f'{platform}_monthly_listeners': social_stats[platform].get('monthly_listeners') }) return global_participant_data def get_global_participant_data(event: dict) -> dict: """Get global participant data by a social account URL. Args: event (dict): input event of the lambda. Returns: dict: a dict of global participant data (label participants, social handles and followers). """ # by spotify id we can get GPs in a very reliable way spotify_url = event.get('spotify_url') if spotify_url: spotify_id = utils.get_spotify_id(spotify_url) query_params = {'spotifyId': spotify_id} main_key = 'globalParticipantBySpotifyId' result = _make_request( constants.QUERIES.get(f'get_{main_key}_query'), query_params) else: main_key = 'globalParticipantBySocialAccountUrl' urls = [ event.get(f'{platform}_url') for platform in constants.PLATFORMS if platform != 'spotify' and event.get(f'{platform}_url')] if urls: for url in urls: # first try to get by an URL with a trailing slash url = url if url[-1] == '/' else url + '/' result = _make_request( constants.QUERIES.get(f'get_{main_key}_query'), {'url': url}) if result.get(main_key): # use the first non-empty result break # an attempt to strip trailing slash if present url = url.rstrip('/') result = _make_request( constants.QUERIES.get(f'get_{main_key}_query'), {'url': url}) if result.get(main_key): # use the first non-empty result break else: result = {'globalParticipantBySocialAccountUrl': []} label_participants = _create_list_of_label_participants(result, main_key) social_stats = _merge_and_dedup(result, main_key) spotify_id = utils.get_spotify_id(social_stats.get('spotify', {}).get('url')) global_participant_data = _create_global_participant_data_result( social_stats, label_participants, spotify_id) return _replace_handles_from_graph_with_user_input(event, global_participant_data) def _make_request(query_string: str, params: dict) -> dict: """Format and send request to graphql-gateway. Args: query_string (str): graphql formatted request body params (dict): parameters to inject into body Returns: dict: response directly from graphql-gateway """ if config.USE_MOCK_GRAPHQL_RESPONSE: if 'getGlobalParticipantDataBySocialAccountUrl' in query_string: return mock_graphql_gateway.get_global_participant_data( test_data='instagram') else: return mock_graphql_gateway.get_global_participant_data( test_data='spotify') transport = RequestsHTTPTransport( url=config.GRAPHQL_GATEWAY_URL, use_json=True, headers=config.GRAPHQL_HEADERS, verify=True, retries=3 ) client = Client(transport=transport, fetch_schema_from_transport=False) query = gql_query(query_string) return client.execute(query, variable_values=params)