"""set-track-metadata.""" import json import re from typing import Dict, List, NamedTuple import uuid import config from config import graphql_gateway from constants import queries from constants.constants import ( ARTIST_NAME, ARTIST_ROLE, PARTICIPANTS_FIELD, PERFORMERS_FIELD, PLACEHOLDER_TRACK_NAME_CONSTANTS, PLACEHOLDER_TRACK_NAME_PATTERNS, WRITERS_FIELD ) from constants.errors import KNOWN_GRAPHQL_ERRORS, \ OWS_PRODUCT_USER_IS_FORBIDDEN from constants.offer_type import OFFER_TYPE_MAP from constants.role_mappings import ( ARTIST_ROLE_MAP, WRITER_ROLE_MAP, ) from ddex_ingester_common.constants.ddex_providers import SME, \ SME_ANALYTICS_PROVIDER, SOM_LIVRE_VENDOR_ID from ddex_ingester_common.constants.ownership import OWNERSHIP_TYPE_MAPPING from ddex_ingester_common.constants.roles import ( DDEX_ARTIST_ROLE_REQUIRED_GENRES, DDEX_ARTIST_ROLE_TO_ORCHARD_ARTIST_TYPE, DDEX_ARTIST_ROLE_TO_ORCHARD_PERFORMER_ROLE, DDEX_ARTIST_ROLE_TO_ORCHARD_WRITER, DDEX_RESOURCE_CONTRIBUTOR_ROLE_TO_ORCHARD_PARTICIPANT_ROLE, DDEX_RESOURCE_CONTRIBUTOR_ROLE_TO_ORCHARD_PERFORMER_ROLE, ) from ddex_ingester_common.constants.swb_deal_types import FOR_DISTRIBUTION from ddex_ingester_common.helpers.list import have_common_elements from ddex_ingester_common.helpers.metadata import ( get_country_id, get_language_id, get_us_publishing_obligation ) from ddex_ingester_common.helpers.s3_ddex import ( get_s3_track, load_ddex_json ) from ddex_ingester_common.lambda_exceptions import \ OwsProductUserIsForbiddenException, SetTrackMetadataException from ddex_ingester_common.logging import utils as logging_utils from ddex_ingester_common.models.s3.genre import Genre as S3Genre from ddex_ingester_common.models.s3.label_participant import LabelParticipant from ddex_ingester_common.models.s3.localized_participant import ( LocalizedParticipant as S3LocalizedParticipant) from ddex_ingester_common.models.s3.participant import ( Participant as S3Participant) from ddex_ingester_common.models.s3.track import ( Track as S3Track) from ddex_ingester_common.models.state_machine.body import ( Body as StateMachineContext) from ddex_ingester_common.release_correction.release_correction import ( is_release_correction) from ddex_ingester_common.release_correction.release_correction_constants import ( # noqa RELEASE_CORRECTION_TRACK_ARTIST_FIELDS as RC_ARTIST_FIELDS, RELEASE_CORRECTION_TRACK_PARTICIPANT_FIELDS as RC_PARTICIPANT_FIELDS, RELEASE_CORRECTION_TRACK_PERFORMER_FIELDS as RC_PERFORMER_FIELDS, RELEASE_CORRECTION_TRACK_WRITER_FIELDS as RC_WRITER_FIELDS, RELEASE_CORRECTION_TRACK_FIELDS, ) from ddex_ingester_common.release_correction.release_correction_diffs import ( ReleaseCorrectionDiffDetail) from ddex_ingester_common.release_correction.release_corrections_s3 import ( load_rc_json, write_rc_json ) from ddex_ingester_common.schemas.s3_schema import S3Schema, TrackSchema from ddex_ingester_common.schemas.state_machine_schema import ( StateMachineSchema, TrackSchema as StateMachineTrackSchema ) from lambdacommon.graphql import graphql from marshmallow.utils import get_value logger = logging_utils.get_logger(config.app_logger) def handler(event, context): """Lambda entry point.""" # Due to being inside an iterator the event is more than just the context event_context = event.get('context') s3_context = S3Schema().load(load_ddex_json(event_context)) context = StateMachineSchema().load(event_context) correlation_id = context.correlation_id or str(uuid.uuid4()) context.correlation_id = correlation_id state_machine_track = StateMachineTrackSchema().load(event.get('track')) s3_track = get_s3_track(s3_context, state_machine_track) logging_utils.update_logger_correlation_id(logger, correlation_id) logging_utils.update_logger_with_message_ids( logger, s3_context.message_id, s3_context.message_thread_id, s3_context.execution_name ) graphql_gateway.set_headers( { 'Orchard-User-Id': config.OA_USER, 'Correlation-Id': correlation_id, } ) unmapped_roles = get_unique_roles(s3_track) try: # Do not replace meaningful track names by placeholder names (Track 1) if (context.ddex_provider == SME_ANALYTICS_PROVIDER and track_name_is_placeholder(s3_track.track_name)): track_name = get_track_name(state_machine_track.tuid) if track_name: s3_track.track_name = track_name # We dont create release corrections for dummies but # if it's a complete Sony product we need to check for invalid # updates with diff_for_release_corrections() even if NFD!=N if is_release_correction(context, check_nfd=False): graphql_result = get_track_data(state_machine_track.tuid) if graphql_result: logger.info(f'Starting error correction for track {state_machine_track.tuid}') # noqa diff_for_release_corrections( s3_context.product.genres, state_machine_track.tuid, context, s3_track, graphql_result, unmapped_roles, event_context, s3_context.label_participants ) # If it's a complete Sony product with NFD=N we dont update it # as it needs to go through release correction if not is_release_correction(context, check_nfd=True): update_track_metadata( context, s3_context.label_participants, context.product.upc, state_machine_track.tuid, s3_context.product.genres, s3_track, unmapped_roles, correlation_id ) update_label_participants(context, s3_context.label_participants, s3_track, unmapped_roles) except graphql.GraphQLError as err: if OWS_PRODUCT_USER_IS_FORBIDDEN in str(err): logger.info(f'Found known GraphQL error: {str(err)}') raise OwsProductUserIsForbiddenException(OWS_PRODUCT_USER_IS_FORBIDDEN) # noqa # probably this is redundant we handle swear words in prep ddex for known_error in KNOWN_GRAPHQL_ERRORS: if known_error in str(err): logger.info(f'Found known GraphQL error: {str(err)}') graphql_response = err.response if type(graphql_response) is not dict: graphql_response = eval(graphql_response) # Add track ISRC to the error dict # so it can be added to the email when the error is processed graphql_response['ISRC'] = s3_track.isrc raise SetTrackMetadataException(graphql_response) from err raise SetTrackMetadataException(f'Graphql error: {str(err)}') from err except Exception as exp: logger.error(f'Error updating track metadata.\n{exp}') raise SetTrackMetadataException( f'Error updating track metadata.\n{exp}') from exp if unmapped_roles: message = '' for role in unmapped_roles: message += '\n' message += ( f'on ISRC: "{s3_track.isrc}" participant ' f'"{unmapped_roles[role]}" has unmapped role "{role}"' ) logger.warning('Some roles were not mapped:' + message) # TODO: Change this handler to return full context once the lambda is moved # in the state machine to be sequential before the handle audio asset steps return StateMachineSchema().dump(context) def get_unique_roles(track: S3Track) -> Dict: """Get a dict with every unique role along with the participant's name.""" unique_roles = {} for artist in track.display_artists: for role in artist.roles: unique_roles[role] = artist.name for contributor in track.resource_contributors: for role in contributor.roles: unique_roles[role] = contributor.name return unique_roles def get_track_data(tuid: int) -> Dict: """Retrieve the data of a track.""" result = graphql_gateway.execute( queries.GET_TRACK_BY_TUID, {'tuid': tuid} )['data']['track'] logger.info(f'Ran get track with tuid {tuid} and received {result}') return result def update_track_metadata( context: StateMachineContext, label_participants, upc: str, tuid: int, genres: List[S3Genre], track: S3Track, unmapped_roles: Dict, correlation_id: str): """Update the metadata of a given track.""" try: logger.info( f'Updating track metadata for track: {TrackSchema().dump(track)}') payload = format_track_update_data( context, label_participants, upc, tuid, genres, track, unmapped_roles) logger.info(f'Payload for GraphQL mutation: {payload}') # Set headers for Som Livre so mechadmin lookup can happen downstream if context.product.vendor_id == SOM_LIVRE_VENDOR_ID: graphql_gateway.set_headers( { 'Grass-Account-Type': config.SOM_LIVRE_ACCOUNT_TYPE, 'Grass-Account-Id': SOM_LIVRE_VENDOR_ID, 'Correlation-Id': correlation_id, } ) graphql_gateway.execute( queries.UPDATE_TRACK_METADATA, {'data': payload} ) # Set headers for Som Livre back to original values if context.product.vendor_id == SOM_LIVRE_VENDOR_ID: graphql_gateway.set_headers( { 'Orchard-User-Id': config.OA_USER, 'Correlation-Id': correlation_id, } ) except graphql.GraphQLError as err: logger.warning( f'Failed to update track metadata for track error: ' f'{str(err)}') if 'HTTP Error 403: Forbidden' in str(err) \ and context.ddex_provider == SME_ANALYTICS_PROVIDER: logger.info( f'Failed to update track metadata for track: ' f'{TrackSchema().dump(track)}') track.lyrics = '' logger.info( f'Updating track metadata for track withgout : ' f'{TrackSchema().dump(track)}') payload = format_track_update_data( context, label_participants, upc, tuid, genres, track, unmapped_roles) logger.info(f'Payload for GraphQL mutation: {payload}') graphql_gateway.execute( queries.UPDATE_TRACK_METADATA, {'data': payload} ) else: raise def format_track_update_data( context: StateMachineContext, label_participants: List[LabelParticipant], upc: str, tuid: int, genres: List[S3Genre], track: S3Track, unmapped_roles: Dict): """Format track update GraphQL payload and check for null values.""" publishers = get_publishers(track) participants = get_track_participants( context, label_participants, track, genres, unmapped_roles) body = { 'upc': upc, 'isrc': track.isrc, 'trackName': track.track_name, 'version': track.track_version, 'metaLanguageCode': track.lyrics_language, 'explicit': track.explicit, 'pInfo': track.p_line, 'lyrics': sanitize_lyrics(track.lyrics), 'recordingCountryId': get_country_id(track.recording_country_code), 'originalRightsHolderCountryId': get_country_id(track.copyright_owner_country), 'ownershipRights': OWNERSHIP_TYPE_MAPPING.get(track.ownership_rights), 'usPublishingObligation': get_us_publishing_obligation(track.us_publishing_obligation), 'participations': participants['participations'], 'performers': participants['performers'], 'publishers': publishers, 'localizations': get_localizations(track), 'previewStartTime': get_preview_start_time(track), 'offerType': OFFER_TYPE_MAP.get(track.offer_type), } # Remove keys with empty values body = {key: value for key, value in body.items() if value is not None} # Hardcode third_party_publisher for Som Livre tracks if publishers and context.product.vendor_id == SOM_LIVRE_VENDOR_ID: body['thirdPartyPublisher'] = 'Y' return { 'update': { 'tracks': [tuid], 'body': body } } def get_publishers(track: S3Track) -> List[Dict]: """Get publishers payload for saveTracks GraphQL mutation.""" if not track.publishers: return None pubs_seen = set() payload = [] for publisher_name in track.publishers: if publisher_name not in pubs_seen: pubs_seen.add(publisher_name) payload.append({ 'name': publisher_name }) return payload def get_localizations(track: S3Track) -> List[Dict]: """Get localizations payload for saveTracks GraphQL mutation. Args: track (S3Track): Track to create payload for Returns: list """ localizations = [] if track.localized_titles: for localized_title in track.localized_titles: if not get_language_id(localized_title.language_code): raise LookupError( 'Language code ' f'{localized_title.language_code} not mapped.') localization = { 'languageId': get_language_id(localized_title.language_code), 'trackName': localized_title.title } if localized_title.version: localization['version'] = localized_title.version localizations.append(localization) return localizations def get_preview_start_time(track: S3Track) -> int: """Get previewStartTime payload for saveTracks GraphQL mutation.""" preview_start_time = track.preview_start_time # (not 0) evaluates to True but 0 is a valid start time if not preview_start_time and preview_start_time != 0: return None if preview_start_time < 0: return None return preview_start_time * 1000 def get_track_participants( context: StateMachineContext, label_participants: List[LabelParticipant], track: S3Track, genres: List[S3Genre], unmapped_roles: Dict, artist_name: bool = False) -> Dict: """Get participants payload for saveTracks GraphQL mutation.""" # AWAL always maps every role regardless of genre skip_genre_check = context.ddex_provider != SME track_participations = [] track_performers = [] for artist in track.display_artists: context_participant = retrieve_label_participants(label_participants, artist) for role in artist.roles: add_artist( artist.localized_names, context_participant, role, track_participations, genres, unmapped_roles, artist_name=artist_name, skip_genre_check=skip_genre_check, ) add_writer( context_participant, role, track_participations, unmapped_roles, artist_name=artist_name ) add_performer( artist.name, role, track_performers, unmapped_roles, DDEX_ARTIST_ROLE_TO_ORCHARD_PERFORMER_ROLE ) for contributor in track.resource_contributors: for role in contributor.roles: context_participant = retrieve_label_participants( label_participants, contributor) # We need to make sure that resource contributors with 'artist' roles are also added to the Tracks # noqa add_artist( artist.localized_names, context_participant, role, track_participations, genres, unmapped_roles, artist_name=artist_name, skip_genre_check=skip_genre_check, ) add_writer( context_participant, role, track_participations, unmapped_roles, artist_name=artist_name ) add_performer( contributor.name, role, track_performers, unmapped_roles, DDEX_RESOURCE_CONTRIBUTOR_ROLE_TO_ORCHARD_PERFORMER_ROLE ) return { 'participations': track_participations, 'performers': track_performers, } def add_artist( localized_names: List[S3LocalizedParticipant], artist, role: str, participations: List[Dict], genres: List[S3Genre], unmapped_roles: Dict, artist_name: bool = False, skip_genre_check: bool = False) -> List[Dict]: """Add artist to artists list for saveTracks GraphQL mutation.""" if is_track_role(role, genres, skip_genre_check=skip_genre_check): unmapped_roles.pop(role, None) new_artist = { 'labelParticipantUuid': artist.label_participant_uuid, 'role': DDEX_ARTIST_ROLE_TO_ORCHARD_ARTIST_TYPE[role], } if artist_name: new_artist[ARTIST_NAME] = artist.name if localized_names: localizations = [] for localized_name in localized_names: localizations.append({ 'languageId': get_language_id( localized_name.language_code), 'name': localized_name.name }) new_artist['localizations'] = localizations # Check for duplicates if new_artist not in participations: participations.append(new_artist) def is_track_role( role: str, genres: List[S3Genre], skip_genre_check: bool = False) -> bool: """Check if a role is an artist role considering the product genres.""" # Check if it's an artist role if role not in DDEX_ARTIST_ROLE_TO_ORCHARD_ARTIST_TYPE: return False if skip_genre_check: return True # Check if this role only applies to certain genres if role not in DDEX_ARTIST_ROLE_REQUIRED_GENRES: return True required_genres = DDEX_ARTIST_ROLE_REQUIRED_GENRES[role] genre_names = [genre.genre for genre in genres] # If any product genre matches one of the required genres the roles applies return have_common_elements(genre_names, required_genres) def is_valid_artist_field( field: str, genres: List[S3Genre]) -> bool: """Check if a release correction field is valid for the product genres.""" role = field.title() # Check if this role only applies to certain genres if role not in DDEX_ARTIST_ROLE_REQUIRED_GENRES: return True required_genres = DDEX_ARTIST_ROLE_REQUIRED_GENRES[role] genre_names = [genre.genre for genre in genres] # If any product genre matches one of the required genres the roles applies return have_common_elements(genre_names, required_genres) def add_writer( artist, role: str, writers: List[Dict], unmapped_roles: Dict, artist_name: bool = False) -> List[Dict]: """Add writer to writers list for saveTracks GraphQL mutation.""" if role in DDEX_ARTIST_ROLE_TO_ORCHARD_WRITER: unmapped_roles.pop(role, None) new_writer = { 'labelParticipantUuid': artist.label_participant_uuid, 'role': DDEX_ARTIST_ROLE_TO_ORCHARD_WRITER[role], } if artist_name: new_writer[ARTIST_NAME] = artist.name # Check for duplicates if new_writer not in writers: writers.append(new_writer) def add_performer( name: str, role: str, performers: List[Dict], unmapped_roles: Dict, role_map: Dict) -> List[Dict]: """Add performer to performers list for saveTracks GraphQL mutation. Args: name: Performer name role: Performer role performers: Previously added performers unmapped_roles: All roles that haven't been mapped role_map : Maps a DDEX role to its Orchard role type and roleId """ if role in role_map: unmapped_roles.pop(role, None) orchard_role = role_map[role] new_performer = { 'name': name, 'roleId': orchard_role['roleId'], 'type': orchard_role['type'] } # Check for duplicates if new_performer not in performers: performers.append(new_performer) def update_label_participants( context: StateMachineContext, label_participants: List[LabelParticipant], track: S3Track, unmapped_roles: Dict): """Update the label participants of a given track. Args: context (object): Context object track (S3Track): Track to update metadata unmapped_roles(dict): All roles that haven't been mapped """ logger.info( f'Updating label participant for track: {TrackSchema().dump(track)}') participations = get_label_participations(label_participants, track, unmapped_roles) if not participations: logger.info('Skipped updating label participants') return payload = { 'isrc': track.isrc, 'vendorId': context.product.vendor_id, 'subaccountId': context.product.subaccount_id, 'participations': participations } # Remove keys with empty values payload = {key: value for key, value in payload.items() if value} logger.info(f'Payload for GraphQL mutation: {payload}') graphql_gateway.execute( queries.SET_LABEL_PARTICIPANTS, payload ) def get_label_participations( label_participants: List[LabelParticipant], track: S3Track, unmapped_roles: Dict, starting_sequence_number: int = 1, artist_name: bool = False) -> List[Dict]: """Get participations for setLabelSoundRecordingParticipations mutation. Args: track (S3Track): Track to create payload for unmapped_roles(dict): All roles that haven't been mapped Returns: list """ role_map = DDEX_RESOURCE_CONTRIBUTOR_ROLE_TO_ORCHARD_PARTICIPANT_ROLE ordered_participants = order_participants(track) sequence_number = starting_sequence_number participations = [] participations_no_sequence_number = [] for participant in ordered_participants: label_participant = retrieve_label_participants(label_participants, participant) label_participant_id = label_participant.label_participant_id for role in participant.roles: if role in role_map: unmapped_roles.pop(role, None) new_participation = { 'labelParticipantId': label_participant_id, 'participationRoleName': role_map[role]['role_name'] } # Check for duplicates if new_participation not in participations_no_sequence_number: participations_no_sequence_number.append( {**new_participation}) new_participation['sequenceNumber'] = sequence_number sequence_number += 1 if artist_name: new_participation['name'] = participant.name participations.append(new_participation) return participations def order_participants(track: S3Track) -> List[S3Participant]: """Order display artists and resource contributors by sequence number. This function returns a list with a track's display_artists ordered by their sequence_number followed by the resource_contributors ordered by their sequence_number. This order is required for the participations field of the setLabelSoundRecordingParticipations mutation. """ get_sequence_number = lambda x: x.sequence_number # noqa participants = [] if track.display_artists: participants.extend(sorted( track.display_artists, key=get_sequence_number )) if track.resource_contributors: participants.extend(sorted( track.resource_contributors, key=get_sequence_number )) return participants def diff_for_release_corrections( genres: List[S3Genre], tuid: int, context: StateMachineContext, s3_track: S3Track, graphql_result: Dict, unmapped_roles: Dict, event_context: Dict, label_participants: List[LabelParticipant]): """Check for items requiring release correction.""" # Check invalid fields before proceeding. logger.info( 'Checking for differences between Orchard and DDEX to add to release correction') # noqa rc_details = [] event_details = {'bucket': context.bucket, 'key': context.key} # Should check if unsupported fields are updated but this logic # was moved to set_product to avoid stopping the ingest at this point # as that would require reverting the created release correction items fields = { 'trackName': ReleaseCorrectionValuePair( old=graphql_result.get('trackName'), new=s3_track.track_name), 'trackVersion': ReleaseCorrectionValuePair( old=graphql_result.get('version'), new=s3_track.track_version), 'explicit': ReleaseCorrectionValuePair( old=graphql_result.get('explicit'), new=s3_track.explicit), 'lyrics': ReleaseCorrectionValuePair( old=graphql_result.get('lyrics'), new=s3_track.lyrics), 'pInfo': ReleaseCorrectionValuePair( old=graphql_result.get('pInfo'), new=s3_track.p_line), 'metaLanguageCode': ReleaseCorrectionValuePair( old=graphql_result.get('metaLanguageCode'), new=s3_track.lyrics_language), } graphql_artists = graphql_result.get('primaryArtists', []) graphql_writers = graphql_result.get('writers', []) graphql_performers = graphql_result.get('performers', []) all_s3_participations = get_track_participants( context, label_participants, s3_track, genres, unmapped_roles, artist_name=True ) s3_participations = all_s3_participations['participations'] s3_performers = all_s3_participations['performers'] s3_label_participants = get_label_participations( label_participants, s3_track, unmapped_roles, starting_sequence_number=0, artist_name=True, ) formatted_s3_artists, formatted_s3_writers = \ format_s3_participations(s3_participations) formatted_s3_label_participants = \ format_s3_label_participants(s3_label_participants) formatted_gql_artists = \ format_graphql_artists(graphql_artists) formatted_gql_writers = \ format_graphql_writers(graphql_writers) formatted_gql_label_participants = \ format_graphql_label_participants(graphql_result) artist_diffs = compare_list_of_dicts( formatted_s3_artists, formatted_gql_artists) logger.info( f'Compared S3 artists: {formatted_s3_artists} ' f'to GraphQL artists: {formatted_gql_artists}. ' f'Result: {artist_diffs}') writer_diffs = compare_list_of_dicts( formatted_s3_writers, formatted_gql_writers) logger.info( f'Compared S3 writers: {formatted_s3_writers} ' f'to GraphQL writers: {formatted_gql_writers}. ' f'Result: {writer_diffs}') # Comparison does not need formatting as they both use (name, roleId, type) performer_diffs = compare_list_of_dicts( s3_performers, graphql_performers) logger.info( f'Compared S3 performers: {s3_performers} ' f'to GraphQL performers: {graphql_performers}. ' f'Result: {performer_diffs}') s3_label_participants_no_seq_number = [p.copy() for p in s3_label_participants] # noqa for p in s3_label_participants_no_seq_number: p.pop('sequenceNumber', None) label_participant_diffs = compare_list_of_dicts( s3_label_participants_no_seq_number, formatted_gql_label_participants) logger.info( f'Compared S3 label participants: {s3_label_participants_no_seq_number} ' # noqa f'to GraphQL label participants: {formatted_gql_label_participants}. ' f'Result: {label_participant_diffs}') if writer_diffs: for p in formatted_s3_writers: rc_details.append(create_release_correction_detail( WRITERS_FIELD, s3_track.isrc, None, [p.get(ARTIST_NAME)] )) if not formatted_s3_writers: rc_details.append(create_release_correction_detail( WRITERS_FIELD, s3_track.isrc, None, [None] )) if performer_diffs: formatted_s3_performers = format_s3_performers(s3_performers) for p in formatted_s3_performers: rc_details.append(create_release_correction_detail( PERFORMERS_FIELD, s3_track.isrc, None, [p] )) if not formatted_s3_performers: rc_details.append(create_release_correction_detail( PERFORMERS_FIELD, s3_track.isrc, None, [None] )) # Performers are updated on release correction if context.product.not_for_distribution == FOR_DISTRIBUTION: payload = format_track_update_data( context, label_participants, context.product.upc, tuid, genres, s3_track, unmapped_roles ) payload['update']['body'] = { 'upc': payload['update']['body']['upc'], 'performers': payload['update']['body']['performers'] } logger.info(f'Updating track metadata with payload: {payload}') graphql_gateway.execute( queries.UPDATE_TRACK_METADATA, {'data': payload} ) if label_participant_diffs: formatted_s3_label_participants = \ format_s3_label_participants(s3_label_participants) for p in formatted_s3_label_participants: rc_details.append(create_release_correction_detail( PARTICIPANTS_FIELD, s3_track.isrc, None, [p] )) if not formatted_s3_label_participants: rc_details.append(create_release_correction_detail( PARTICIPANTS_FIELD, s3_track.isrc, None, [None] )) # Label participants are updated on release correction if context.product.not_for_distribution == FOR_DISTRIBUTION: update_label_participants(context, label_participants, s3_track, unmapped_roles) if artist_diffs: mapped_roles = set() for p in formatted_s3_artists: new_value = { ARTIST_ROLE: p[ARTIST_ROLE], ARTIST_NAME: p[ARTIST_NAME] } rc_details.append(create_release_correction_detail( new_value[ARTIST_ROLE], s3_track.isrc, None, [new_value] )) mapped_roles.add(p[ARTIST_ROLE]) invalid_roles = get_genre_invalid_roles(genres) # Clear roles that have no artists in the DDEX for role in RC_ARTIST_FIELDS: if role not in mapped_roles and role not in invalid_roles: mapped_roles.add(role) rc_details.append(create_release_correction_detail( role, s3_track.isrc, None, [None])) elif includes_genre_updates(context): """ In situations where a genreId or subgenreId has changed for this product, we also need to ensure that the 'featuring' artist(s) are included for every track. We ignore this for participation_diffs because all main and featuring artists are already updated by default. This is a fallback. This is a workaround to compensate for a bug in ows-tracks where the featuring artists are not reliably pulled for a product, resulting in a failed product submit. """ for p in formatted_s3_artists: if p[ARTIST_ROLE] == 'featuring': new_value = { ARTIST_ROLE: p[ARTIST_ROLE], ARTIST_NAME: p[ARTIST_NAME] } rc_details.append(create_release_correction_detail( new_value[ARTIST_ROLE], s3_track.isrc, None, [new_value])) for key in fields.keys(): if fields[key].new and fields[key].old != fields[key].new: rc_details.append(create_release_correction_detail( key, s3_track.isrc, fields[key].old, fields[key].new)) if rc_details: if context.error_correction and \ context.error_correction.release_correction_id: update_release_correction( context, tuid, s3_track, rc_details, genres) rc_details = remove_processed_details(rc_details) s3_release_corrections = load_rc_json(event_context) s3_release_corrections['changes'].extend(rc_details) logger.info(f'Writing RC details back to S3: {s3_release_corrections}') write_rc_json(event_details, s3_release_corrections) def create_release_correction_detail( field_name: str, isrc: str, old_value: any, new_value: any) -> ReleaseCorrectionDiffDetail: """Create ReleaseCorrectionDiffDetail using given values.""" logger.info(f'Creating Release Correction Detail for field: {field_name}' f' with old value: {old_value}' f' and new value: {new_value}') rc_track_field = RELEASE_CORRECTION_TRACK_FIELDS[field_name] return ReleaseCorrectionDiffDetail( field_name=field_name, isrc=isrc, old=old_value, new=new_value, db_table_name=rc_track_field.db_table_name, db_field_name=rc_track_field.db_field_name, email_customer=rc_track_field.email_customer, accept_update=rc_track_field.accept_update) def format_s3_participations(participations: List[Dict]) -> List: """Format S3 participations list for comparison.""" formatted_s3 = { 'artists': [], 'writers': [], } for p in participations: role = p.get('role') if role not in ARTIST_ROLE_MAP and role not in WRITER_ROLE_MAP: raise ValueError(f'Cannot map participant role: {role}') if ARTIST_ROLE_MAP.get(role): part = { ARTIST_NAME: p.get(ARTIST_NAME), ARTIST_ROLE: ARTIST_ROLE_MAP[role] } formatted_s3['artists'].append(part) if WRITER_ROLE_MAP.get(role): part = { ARTIST_NAME: p.get(ARTIST_NAME), ARTIST_ROLE: WRITER_ROLE_MAP[role] } formatted_s3['writers'].append(part) return formatted_s3['artists'], formatted_s3['writers'] def format_s3_performers(performers: List[Dict]) -> List: """Format S3 performers list for release correction.""" formatted_s3 = [] for p in performers: formatted_s3.append({ 'birth_name': p['name'], 'performer_role_id': p['roleId'], 'type': p['type'], }) return formatted_s3 def format_s3_label_participants(participants: List[Dict]) -> List: """Format S3 label participants list for release correction.""" formatted_s3 = [] for p in participants: formatted_s3.append({ 'participation': { 'id': p['labelParticipantId'], 'name': p['name'], }, 'participation_role_name': p['participationRoleName'], 'sequence_number': p['sequenceNumber'], }) return formatted_s3 def format_graphql_artists(artists: List[Dict]) -> List: """Format GraphQL artists list for comparison.""" formatted_gql = [] for p in artists: part = { ARTIST_ROLE: p['artistType'], ARTIST_NAME: p['artistName'] } formatted_gql.append(part) return formatted_gql def format_graphql_writers(writers: List[Dict]) -> List: """Format GraphQL writers list for comparison.""" formatted_gql = [] for p in writers: part = { ARTIST_ROLE: p['type'] or 'writer', ARTIST_NAME: p['name'] } formatted_gql.append(part) return formatted_gql def format_graphql_label_participants(graphql_response: List[Dict]) -> List: """Format GraphQL label participants list for comparison.""" formatted_gql = [] participants = get_value( graphql_response, 'labelSoundRecording.participations', []) for p in participants: part = { 'labelParticipantId': get_value(p, 'participant.id'), 'participationRoleName': get_value(p, 'role.name'), 'name': get_value(p, 'participant.name'), } formatted_gql.append(part) return formatted_gql def compare_list_of_dicts( s3_participations: List[Dict], graphql_participations: List[Dict]) -> List: """Compare product participations for release correction.""" list_a_differences = [ item for item in s3_participations if item not in graphql_participations ] list_b_differences = [ item for item in graphql_participations if item not in s3_participations ] return [*list_a_differences, *list_b_differences] def includes_genre_updates(context: StateMachineContext) -> bool: """Check if genre fields have been updated.""" graphql_response = get_release_corrections(context.product.upc) if not graphql_response: return False GENRE_ID = 'genre_id' RELEASE_SUBGENRE = 'release_subgenre' if graphql_response.get('status') == 'active': for item in graphql_response.get('items', []): if item.get('fieldName') in (GENRE_ID, RELEASE_SUBGENRE): return True return False def get_release_corrections(upc: str) -> Dict: """Retrieve release corrections from GraphQL.""" response = graphql_gateway.execute( queries.GET_PRODUCT_RELEASE_CORRECTIONS, {'upc': upc} )['data']['productByUpc']['releaseCorrection'] logger.info(f'Release Correction response: {response}') return response def update_release_correction( context: StateMachineContext, tuid: int, s3_track: S3Track, rc_details: List[ReleaseCorrectionDiffDetail], genres: List[S3Genre]): """Update the release correction for a track with GraphQL.""" if not rc_details: return rc_updates = [] artist_fields = {key: [] for key in RC_ARTIST_FIELDS.keys()} artist_fields.update({key: [] for key in RC_WRITER_FIELDS.keys()}) artist_fields.update({key: [] for key in RC_PERFORMER_FIELDS.keys()}) artist_fields.update({key: [] for key in RC_PARTICIPANT_FIELDS.keys()}) for detail in rc_details: if detail.accept_update: logger.info( f'Adding RC Detail for {detail.field_name}. ' f'Field data: {detail.new}') # Artists require special formatting, so aggregate the data if detail.field_name in RC_ARTIST_FIELDS: # Ignore classical roles on non classical tracks if is_valid_artist_field(detail.field_name, genres): # Artists are serialized as a list of dicts artist_fields[detail.field_name].append(*detail.new) elif detail.field_name in RC_WRITER_FIELDS: # Writers are serialized as a list of names artist_fields[detail.field_name].append(*detail.new) elif detail.field_name in RC_PERFORMER_FIELDS: # Performers are serialized as a list of dicts artist_fields[detail.field_name].append(*detail.new) elif detail.field_name in RC_PARTICIPANT_FIELDS: # Label participants are serialized as a list of dicts artist_fields[detail.field_name].append(*detail.new) else: rc_updates.append({ 'tableName': detail.db_table_name, 'fieldName': detail.db_field_name, 'keyValue': json.dumps(detail.new), 'keyId': tuid, }) table_field_names = set() # If we have any, parse that Artist data before adding to updates. for key, value in artist_fields.items(): if value: rc_map = RC_ARTIST_FIELDS if key in RC_WRITER_FIELDS: rc_map = RC_WRITER_FIELDS elif key in RC_PERFORMER_FIELDS: rc_map = RC_PERFORMER_FIELDS elif key in RC_PARTICIPANT_FIELDS: rc_map = RC_PARTICIPANT_FIELDS # A new value of None means we should clear the field if value == [None]: value = [] table_name = rc_map[key].db_table_name field_name = rc_map[key].db_field_name name_tuple = (table_name, field_name) if name_tuple not in table_field_names: correction_object = { 'tableName': table_name, 'fieldName': field_name, 'keyValue': json.dumps(value), 'keyId': tuid, } rc_updates.append(correction_object) table_field_names.add(name_tuple) # Merge duplicate table and field names if present # Fields composer and performer map to the same values else: for rc_update in rc_updates: rc_update_table_name = rc_update['tableName'] rc_update_field_name = rc_update['fieldName'] if rc_update_field_name == field_name and \ rc_update_table_name == table_name: rc_update_value = rc_update['keyValue'] merged_value = json.loads(rc_update_value) merged_value.extend(value) rc_update['keyValue'] = json.dumps(merged_value) if rc_updates: product_id = context.product.product_id payload = { 'productId': product_id, 'releaseCorrectionId': context.error_correction.release_correction_id, # noqa 'corrections': rc_updates } logger.info(f'Creating Release Correction Details for {product_id}' f' With details: {payload}') response = graphql_gateway.execute( queries.CREATE_PRODUCT_CORRECTION_DETAIL, payload )['data']['createProductCorrectionDetail'] logger.info(f'RC Creation response: {response}') def remove_processed_details(rc_details: List) -> List: """Remove all processed Release Correction details.""" not_processed = [] for rcd in rc_details: # Assume all acceptable updates have been processed. if not rcd.accept_update: not_processed.append(rcd) return not_processed def get_genre_invalid_roles(genres: List[S3Genre]) -> set: """Get roles that do not apply to the product genres.""" roles = set() for role in DDEX_ARTIST_ROLE_REQUIRED_GENRES: required_genres = DDEX_ARTIST_ROLE_REQUIRED_GENRES[role] genre_names = [genre.genre for genre in genres] # Applies if any product genre matches one of the required genres if not have_common_elements(genre_names, required_genres): roles.add(role.lower()) return roles def sanitize_lyrics(lyrics: str) -> str: """Sanitizes lyrics to remove problem strings.""" # 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 def retrieve_label_participants(label_participants, artist): """Retrieve project artist from artist list in S3 data.""" if label_participants and artist: for label_participant in label_participants: if label_participant.name == artist.name: return label_participant def track_name_is_placeholder(track_name): """Check if track name is placeholder.""" if track_name in PLACEHOLDER_TRACK_NAME_CONSTANTS: return True for placeholder in PLACEHOLDER_TRACK_NAME_PATTERNS: if re.match(placeholder, track_name): return True return False def get_track_name(tuid: int) -> str: """Retrieve the data of a track.""" result = graphql_gateway.execute( queries.GET_TRACK_NAME_BY_TUID, {'tuid': tuid} )['data']['track'] logger.info(f'Ran get_track_name with tuid {tuid} and received {result}') return result['trackName'] class ReleaseCorrectionUpdateException(Exception): """Release correction update exception.""" class ReleaseCorrectionValuePair(NamedTuple): """Contains the current and incoming values for a DDEX field.""" new: any old: any