"""Track utilities.""" from typing import Dict, List from bulk_metadata_ingester_common.constants.file import CSV_VALUE_SEPARATOR from bulk_metadata_ingester_common.constants.roles import ( BULK_PERFORMER_TYPE_TO_GRAPH_PERFORMER_TYPE ) from bulk_metadata_ingester_common.models.bulk_release import ( BulkRelease, BulkTrack ) from bulk_metadata_ingester_common.utils.catalog_ingestion import \ save_catalog_ingestion_action from bulk_metadata_ingester_common.utils.error import graphql_execute from config import graphql_gateway from constants import queries from constants.exceptions import ( DuplicateParticipantException, TrackPerformerTypeException ) from constants.languages import LANGUAGE_CODE_MAP from constants.ownership import OWNERSHIP_TYPE_MAPPING from constants.roles import ( BULKTRACK_FIELD_REQUIRED_GENRES_SUBGENRES, BULKTRACK_FIELD_TO_TRACK_ARTIST_TYPE, ORCHARD_ROLES, TRACK_PERFORMER_NAME, TRACK_PERFORMER_ROLE, TRACK_PERFORMER_TYPE ) from ddex_ingester_common.constants.catalog_ingestion import ( INSERT_ACTION) from ddex_ingester_common.helpers.metadata import get_country_id from ddex_ingester_common.lambda_exceptions import SetTrackMetadataException def create_tracks( event: Dict, release: BulkRelease, payload: Dict, logger: object) -> Dict: """Create a track with the specified payload. Args: payload (Dict): A dictionary containing the data for the track to be created. logger (object): An instance of a logging object. Returns: Dict: A dictionary containing the response data from the server. """ result = graphql_execute( graphql_gateway, queries.SAVE_TRACKS, {'data': payload}, logger ) if result: payload_tracks = payload['data']['create']['tracks'] for track in result['data']['saveTracks']: for t in payload_tracks: if t['isrc'] == track.get('isrc'): track_name = t.get('trackName') # track_name = t.get('trackName') for t in payload_tracks if t['isrc'] == track.get('isrc') # noqa: E501 track_data = { 'isrc': track.get('isrc'), 'tuid': track.get('tuid'), 'track_sequence_number': track.get('trackNumber'), 'track_volume_number': track.get('volumeNumber'), 'track_name': track_name } save_catalog_ingestion_action( event, release, track_data, INSERT_ACTION ) return result['data']['saveTracks'] def delete_tracks(product_id: str, tuids: List[str], logger: object): """Delete tracks with the specified tuids. Args: product_id (str): The ID of the product. tuids (List[str]): A list of track IDs to be deleted. logger (object): An instance of a logging object. Returns: None """ if not product_id or not tuids: logger.info('Skipped deleting tracks') return payload = { 'delete': { 'productId': str(product_id), 'tracks': [str(t) for t in tuids] } } graphql_execute( graphql_gateway, queries.SAVE_TRACKS, {'data': payload}, logger ) def format_create_track_data( participants: Dict, upc: str, track: BulkTrack) -> Dict: """Format the create track GraphQL payload and checks for null values. Args: participants (Dict): A dictionary containing information about the artists(contributors) and performers. upc (str): The UPC of the track. track (BulkTrack): The object containing information about the track. Returns: Dict: A dictionary containing the formatted data for the create track GraphQL payload. Raises: SetTrackMetadataException: If the track language is not recognized. """ try: # Try to split the field, if multiple artists publishers = track.publishers.split(CSV_VALUE_SEPARATOR) # make list, and strip publishers = [{'name': tp.strip()} for tp in publishers if tp.strip()] except (AttributeError, ValueError, TypeError): # Garbage - move along publishers = [] # TODO: This should be moved / checked pre-flight explicit = 'Y' if track.explicit.lower() == 'yes' else 'N' track_audio_language = LANGUAGE_CODE_MAP[track.track_audio_language] body = { 'upc': upc, 'isrc': track.isrc, 'volumeNumber': track.volume, 'trackNumber': track.track_no, 'trackName': track.track_name, 'metaLanguageCode': track_audio_language, 'explicit': explicit, 'pInfo': track.track_p_info, 'lyrics': sanitize_lyrics(track.track_lyrics), 'recordingCountryId': get_country_id( track.country_of_recording), 'originalRightsHolderCountryId': get_country_id( track.nationality_of_original_copyright_owner), 'ownershipRights': OWNERSHIP_TYPE_MAPPING.get( track.ownership_for_this_sound_recording), 'participations': participants['artists'], 'performers': participants['performers'], 'publishers': publishers, 'version': track.track_version, # Saved for later use: # 'usPublishingObligation': get_us_publishing_obligation( # track.get('us_publishing_obligation')), # Not in the model, do we get this data? # 'offerType': OFFER_TYPE_MAP.get(track.get('offer_type')), } # Remove keys with empty values body = {key: value for key, value in body.items() if value is not None} return body def add_track_artist( label_participant_uuid: str, performer: Dict, track_performers: List[Dict]) -> List[Dict]: """Add an artist to the artists list for saveTracks GraphQL mutation. Args: label_participant_uuid (str): The UUID of the label participant. performer (Dict): A dictionary containing information about a performer. track_performers (List[Dict]): A list of dictionaries containing nformation about the performers. Returns: None: But a list of dictionaries containing all added performers persists. """ role = performer['role'] new_artist = { 'labelParticipantUuid': label_participant_uuid, 'role': role, } # Check for duplicates if new_artist not in track_performers: track_performers.append(new_artist) def add_track_performer( performer: Dict, track_performers: List[Dict], role_map: Dict) -> List[Dict]: """Add performer to list of performers for saveTracks mutation. Args: performer (dict): Dictionary containing performer info, including name and role. track_performers (List[Dict]): List of previously added performers. role_map (Dict): Maps DDEX role to Orchard role type and roleId. Returns: List of dictionaries representing all performers, updated. """ role = performer['role'] name = performer['name'] role_type = performer['type'] if role in role_map: orchard_role_id = role_map[role] new_performer = { 'name': name, 'roleId': int(orchard_role_id), 'type': role_type } # Check for duplicates if new_performer not in track_performers: track_performers.append(new_performer) else: msg = f'"{role}" is not a valid track performer role.' raise SetTrackMetadataException(msg) def get_orchard_tracks(upc: str, logger: object) -> List[Dict]: """Retrieve existing tracks in the orchard for a certain UPC. Args: upc (str): The UPC code to retrieve tracks for. logger (object): The logger object to use for logging. Returns: List[Dict]: A list of dictionaries representing the tracks found for the given UPC. If no tracks are found, an empty list is returned. """ result = graphql_execute( graphql_gateway, queries.GET_ORCHARD_TRACKS, {'upc': upc}, logger ) if result['data']['productByUpc']: return result['data']['productByUpc']['tracks'] else: return [] def get_all_roles( track: BulkTrack, genre: str, subgenre: str, logger: object) -> Dict: """Return a dictionary with every role along with the participant's name. Args: track (BulkTrack): An object containing information about the track. genre (str): A string representing the genre of the track. subgenre (str): A string representing the subgenre of the track. logger (object): An object used for logging messages. Returns: Dict: A dictionary containing tall the roles in the track, with updated information about the artists and performers. """ # Log message logger.info(f'Getting all roles for: {track.isrc} - "{track.track_name}"') track_performer_list = [] track_contributor_list = [] # Handle track-level artists for field in BULKTRACK_FIELD_TO_TRACK_ARTIST_TYPE: # If field is empty, move along. if not getattr(track, field): continue # Check if role is special case / classical if field in BULKTRACK_FIELD_REQUIRED_GENRES_SUBGENRES: # Get genre list required_genres = \ BULKTRACK_FIELD_REQUIRED_GENRES_SUBGENRES[field].keys() # Normalize case for comparison required_genres = [g.lower() for g in required_genres] # Get subgenre list required_subgenres = \ BULKTRACK_FIELD_REQUIRED_GENRES_SUBGENRES[field][genre] # Normalize case for comparison required_subgenres = [sg.lower() for sg in required_subgenres] try: # Try to split the field, if multiple artists artists = getattr(track, field).split(CSV_VALUE_SEPARATOR) # make list, and strip artists = [a.strip() for a in artists if a.strip()] except (AttributeError, ValueError, TypeError): # Garbage - move along continue # If empty set, move along if not artists: continue # Format for payload for artist in artists: track_performer = { 'role': BULKTRACK_FIELD_TO_TRACK_ARTIST_TYPE[field], 'name': artist } # Add to payload track_performer_list.append(track_performer) # Handle track performers for i in range(1, 5): performer_type = getattr(track, TRACK_PERFORMER_TYPE.format(i)) performer_role = getattr(track, TRACK_PERFORMER_ROLE.format(i)) performer_name = getattr(track, TRACK_PERFORMER_NAME.format(i)) # If blank, then move along if not performer_type or not performer_role or not performer_name: continue # Format for payload contributing_performer = { # 'type': performer_type, 'role': performer_role, 'name': performer_name } # TODO: This validation should be moved to pre-flight # Map bulk performer type to graph performer type if performer_type in BULK_PERFORMER_TYPE_TO_GRAPH_PERFORMER_TYPE: contributing_performer['type'] = \ BULK_PERFORMER_TYPE_TO_GRAPH_PERFORMER_TYPE[ performer_type ] # Add to payload track_contributor_list.append(contributing_performer) else: msg = f'Unrecognized performer type: {performer_type}' raise TrackPerformerTypeException(msg) # Return payloads all_roles = { 'artists': track_performer_list, 'performers': track_contributor_list } return all_roles def get_track_participants( track: BulkTrack, all_participants: List[Dict], all_roles: Dict, logger: object) -> Dict: """Return the participants payload for a saveTracks GraphQL mutation. Args: track (BulkTrack): An object containing information about the track. all_participants (List[Dict]): A list of dictionaries containing information about all participants. all_roles (Dict): A dictionary containing information about all roles. logger (object): An object used for logging messages. Returns: Dict: A dictionary payload containing the formatted artists and performer info. """ # Log message logger.info( f'Getting all participants for: {track.isrc} - "{track.track_name}"') track_artists = [] track_performers = [] # Loop through artists for artist in all_roles['artists']: context_participant = \ get_context_participant(all_participants, artist['name']) label_participant_uuid = context_participant['label_participant_uuid'] add_track_artist( label_participant_uuid, artist, track_artists, ) # Loop through performers for performer in all_roles['performers']: # for role in contributor['roles']: context_participant = \ get_context_participant(all_participants, performer['name']) label_participant_uuid = context_participant['label_participant_uuid'] add_track_performer( performer, track_performers, ORCHARD_ROLES ) return { 'artists': track_artists, 'performers': track_performers } def sanitize_lyrics(lyrics: str) -> str: """Remove problem strings from the lyrics. Args: lyrics (str): The string containing the lyrics. Returns: str: The sanitized version of the lyrics. """ # This method is to avoid GenericLFI_BODY WAF rejections # See https://docs.aws.amazon.com/waf/latest/developerguide/aws-managed-rule-groups-baseline.html#aws-managed-rule-groups-baseline-crs # noqa # If we find nothing, pass back original string. sanitized_lyrics = lyrics if lyrics and '../' in lyrics: sanitized_lyrics = lyrics.replace('../', '').strip() return sanitized_lyrics # From ddex_ingester_common def get_context_participant( all_participants: List[Dict], name_to_find: str) -> Dict: """Return context participant whose name matches the passed name. Args: all_participants (List[Dict]): The list of all participants. name_to_find (str): The name of the participant to find. Returns: Dict: The matching participant. Raises: DuplicateParticipantException: If multiple participants with the same name are found. """ matching_participants = [] for participant in all_participants: if participant['name'] == name_to_find: matching_participants.append(participant) if not matching_participants: return None elif len(matching_participants) > 1: raise DuplicateParticipantException( f'Found multiple participants with the same name: ' f'{matching_participants}') else: return matching_participants[0]