"""Lambda function module.""" import re import uuid from time import sleep from typing import Dict, List, NamedTuple, Optional from ddex_ingester_common.constants.ddex_providers import ( RELEASE_CORRECTION_PROVIDERS, SME) from ddex_ingester_common.constants.roles import \ DDEX_VIDEO_ROLE_TO_ORCHARD_ARTIST_TYPE from ddex_ingester_common.constants.status import (IN_CONTENT, TRANSFER_TO_CONTENT) from ddex_ingester_common.helpers.s3_ddex import load_ddex_json from ddex_ingester_common.logging import utils as logging_utils from ddex_ingester_common.models.s3.body import Body as S3Context from ddex_ingester_common.models.state_machine.body import \ Body as StateMachineContext from ddex_ingester_common.release_correction.release_correction_diffs import \ ReleaseCorrectionDiffDetail from ddex_ingester_common.release_correction.release_corrections_s3 import ( create_blank_rc_json, write_rc_json) from ddex_ingester_common.schemas.s3_schema import S3Schema from ddex_ingester_common.schemas.state_machine_schema import \ StateMachineSchema from lambdacommon.graphql import graphql import config from config import graphql_gateway from constants import queries from constants.regex_strings import (SAMPLE_VEVO_STRING, SPECIAL_INSTRUCTION_PATTERN) from constants.video import (DEFAULT_VIDEO_CHANNEL, PARENTAL_ADVISORY, RC_JSON_NAME, TYPE_OF_VIDEO) logger = logging_utils.get_logger(config.app_logger) def handler(event, context): """Lambda entry point.""" logger.info(f'Triggered ddex-ingester-set-video-metadata: {event}') s3_context = S3Schema().load(load_ddex_json(event)) context = StateMachineSchema().load(event) correlation_id = context.correlation_id or str(uuid.uuid4()) context.correlation_id = correlation_id 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, } ) graphql_result = check_for_product(context) if context.ddex_provider != SME: check_remapped_product_code(s3_context, graphql_result) if context.product.status in [IN_CONTENT, TRANSFER_TO_CONTENT] \ and context.ddex_provider in RELEASE_CORRECTION_PROVIDERS: check_for_video_updates( context, s3_context, graphql_result) return StateMachineSchema().dump(context) try: update_video_metadata(context, s3_context) except graphql.GraphQLError as err: product_code_collision = 'productCodeInUse' in get_error_message(err) if context.ddex_provider != SME and product_code_collision: handle_product_code_collision(context, s3_context) else: raise return StateMachineSchema().dump(context) def check_for_product(context: StateMachineContext) -> Dict: """Check for existence of a product.""" upc = context.product.upc logger.info( f'Running get product with upc: {upc}') payload = { 'upc': upc } result = graphql_gateway.execute( queries.GET_PRODUCT_BY_UPC, payload )['data']['productByUpc'] if result: context.product.product_id = result['productId'] return result def check_for_video_updates( context: StateMachineContext, s3_context: S3Context, gql_result: Dict): """Check for any fields that were attempted to be updated.""" logger.info( 'Checking for differences between Orchard and SME DDEX.') # noqa _, valid_video_channel = handle_channel_selection( context.product.product_id, context.video.current_channel, s3_context.product.artist_profile_page ) special_instructions = get_special_instructions( context.product.upc, s3_context.product.artist_profile_page, valid_video_channel, context.product.special_instructions, context.ddex_provider ) fields = { 'productId': ReleaseCorrectionValuePair( old=gql_result.get('productId'), new=context.product.product_id), 'vendorId': ReleaseCorrectionValuePair( old=gql_result.get('vendorId'), new=context.product.vendor_id), 'subaccountId': ReleaseCorrectionValuePair( old=gql_result.get('subaccountId'), new=context.product.subaccount_id), 'videoTitle': ReleaseCorrectionValuePair( old=gql_result.get('productName'), new=s3_context.video.video_name), 'deliveredVersion': ReleaseCorrectionValuePair( # version for Video products, deliveredVersion for Audio products old=gql_result.get('version'), new=s3_context.video.video_version), 'genreId': ReleaseCorrectionValuePair( old=gql_result.get('genreId'), new=context.product.genre_id), 'subgenreId': ReleaseCorrectionValuePair( old=gql_result.get('subgenreId'), new=context.product.subgenre_id), 'notForDistribution': ReleaseCorrectionValuePair( old=gql_result.get('notForDistribution'), new=context.product.not_for_distribution), 'productCode': ReleaseCorrectionValuePair( old=gql_result.get('productCode'), new=s3_context.video.product_code), 'imprint': ReleaseCorrectionValuePair( old=gql_result.get('imprint'), new=s3_context.product.imprint), 'specialInstructions': ReleaseCorrectionValuePair( old=gql_result.get('specialInstructions'), new=special_instructions), 'lyrics': ReleaseCorrectionValuePair( old=gql_result.get('videoProduct').get('lyrics'), new=s3_context.video.lyrics), 'languageOfVideoContent': ReleaseCorrectionValuePair( old=gql_result.get('videoProduct').get('languageOfVideoContent'), new=s3_context.video.content_language), 'languageOfVideoTitle': ReleaseCorrectionValuePair( old=gql_result.get('videoProduct').get('languageOfVideoTitle'), new=s3_context.video.title_language), 'description': ReleaseCorrectionValuePair( old=gql_result.get('videoProduct').get('description'), new=s3_context.video.description), 'parentalAdvisory': ReleaseCorrectionValuePair( old=PARENTAL_ADVISORY.get( gql_result.get('videoProduct').get('parentalAdvisory'), gql_result.get('videoProduct').get('parentalAdvisory') ), # noqa new=s3_context.video.parental_advisory), 'keywords': ReleaseCorrectionValuePair( old=set(gql_result.get('videoProduct').get('keywords') or []), new=set(s3_context.video.keywords or [])), 'typeOfVideo': ReleaseCorrectionValuePair( old=TYPE_OF_VIDEO.get( gql_result.get('videoProduct').get('typeOfVideo'), gql_result.get('videoProduct').get('typeOfVideo') ), # noqa new=s3_context.video.video_type), 'plineYear': ReleaseCorrectionValuePair( old=gql_result.get('videoProduct').get('pLine').get('year'), new=s3_context.video.p_line.year), 'plineCopyrightHolder': ReleaseCorrectionValuePair( old=gql_result.get('videoProduct').get('pLine').get('holder'), new=s3_context.video.p_line.line), 'clineYear': ReleaseCorrectionValuePair( old=gql_result.get('videoProduct').get('cLine').get('year'), new=s3_context.video.c_line.year), 'clineCopyrightHolder': ReleaseCorrectionValuePair( old=gql_result.get('videoProduct').get('cLine').get('holder'), new=s3_context.video.c_line.line), } update_diff_details = [] for key in fields.keys(): if fields[key].new and fields[key].old != fields[key].new: current_values = fields[key].old new_values = fields[key].new logger.info(f'Found difference for Video field {key} ' f'Current field values: {current_values} ' f'Update attempt field values: {new_values}') update_diff_details.append(ReleaseCorrectionDiffDetail( field_name=key, isrc=None, old=current_values, new=new_values)) if update_diff_details: event_details = {'bucket': context.bucket, 'key': context.key} s3_release_corrections = {'changes': update_diff_details} create_blank_rc_json(event_details, RC_JSON_NAME) write_rc_json(event_details, s3_release_corrections, RC_JSON_NAME) def update_video_metadata( context: StateMachineContext, s3_context: S3Context) -> Dict: """Call the graphql mutation to update video metadata.""" channel_selection, valid_video_channel = handle_channel_selection( context.product.product_id, context.video.current_channel, s3_context.product.artist_profile_page, ) special_instructions = get_special_instructions( context.product.upc, s3_context.product.artist_profile_page, valid_video_channel, context.product.special_instructions, context.ddex_provider ) payload = { 'update': { 'productId': str(context.product.product_id), 'accountId': context.product.vendor_id, 'subaccountId': context.product.subaccount_id, 'videoTitle': s3_context.video.video_name, 'version': s3_context.video.video_version, 'typeOfVideo': s3_context.video.video_type, 'languageOfVideoTitle': s3_context.video.title_language, 'languageOfVideoContent': s3_context.video.content_language, 'imprint': s3_context.video.imprint, 'description': s3_context.video.description, 'lyrics': s3_context.video.lyrics, 'parentalAdvisory': s3_context.video.parental_advisory, 'specialInstructions': special_instructions, 'associatedTrackId': context.video.associated_track_tuid, 'genreId': context.product.genre_id, 'subgenreId': context.product.subgenre_id, 'plineYear': s3_context.video.p_line.year, 'plineCopyrightHolder': s3_context.video.p_line.line, 'clineYear': s3_context.video.c_line.year, 'clineCopyrightHolder': s3_context.video.c_line.line, 'productCode': s3_context.video.product_code, 'isrc': s3_context.video.isrc, 'primaryArtists': format_primary_artists(context, s3_context), 'keywords': s3_context.video.keywords, 'notForDistribution': context.product.not_for_distribution, 'vendorReleaseIdentifier': s3_context.product.proprietary_id, 'manufacturerUpc': ( s3_context.product.manufacturer_upc if s3_context.product.manufacturer_upc else context.product.upc), 'vendorCatalogNumber': s3_context.product.catalog_number, } } if channel_selection: payload['update']['channelSelection'] = channel_selection logger.info(f'Updating video metadata with payload: {payload}') return graphql_gateway.execute( queries.SAVE_VIDEO_SINGLE_PRODUCT, {'data': payload} )['data']['saveVideoSingleProduct'] def format_primary_artists( context: StateMachineContext, s3_context: S3Context) -> List[Dict]: """Format the primaryArtists field of the update video payload. Event though the field is called primary artists, it also receives secondary contributors. """ display_artists = s3_context.video.display_artists or [] resource_contributors = s3_context.video.resource_contributors or [] primary_artists = [] for s3_participant in display_artists + resource_contributors: context_participant = retrieve_label_participants( s3_context.label_participants, s3_participant) for role in s3_participant.roles: if role in DDEX_VIDEO_ROLE_TO_ORCHARD_ARTIST_TYPE: primary_artists.append({ 'artistName': s3_participant.name, 'artistType': DDEX_VIDEO_ROLE_TO_ORCHARD_ARTIST_TYPE[role], 'artistId': int(context_participant.artist_id), 'vendorId': context.product.vendor_id, 'subaccountId': context.product.subaccount_id, }) return primary_artists def handle_channel_selection( product_id: int, current_channel: str, new_channel: Optional[str]) -> (str, bool): """Handle setting video channel on a video product. Returns the selected channel and if the received video_channel is a valid channel for this product. """ # Don't overwrite a valid channel with a null channel if not new_channel and current_channel \ and current_channel != DEFAULT_VIDEO_CHANNEL: logger.info('Null video channel, skipping check of supported channels') return None, False logger.info(f'Checking if video channel "{new_channel}" is supported.') supported_video_channels = graphql_gateway.execute( queries.GET_VIDEO_CHANNELS, {'productId': product_id} )['data']['getAvailableVideoChannelsByProductId'] logger.info( f'Supported video channels for Product: {supported_video_channels}' ) valid_video_channel = False if new_channel: for channel in supported_video_channels: valid_video_channel = new_channel in [ channel.get('channelName'), channel.get('channelId') ] if valid_video_channel: break channel_selection = DEFAULT_VIDEO_CHANNEL if valid_video_channel: channel_selection = new_channel return channel_selection, valid_video_channel def get_special_instructions( upc: str, new_channel: Optional[str], valid_video_channel: bool, context_special_instructions: Optional[str], ddex_provider: str) -> str: """Get special instructions for a video product. Will add to the special instructions if video channel is not valid. """ if valid_video_channel or not new_channel: return context_special_instructions special_instructions = graphql_gateway.execute( queries.GET_PRODUCT_BY_UPC, {'upc': upc} )['data']['productByUpc'].get('specialInstructions') logger.info( f'Special instructions found: {special_instructions}' ) # Compile pattern we care about pattern = re.compile(SPECIAL_INSTRUCTION_PATTERN) updated_string = '' if special_instructions and pattern.search(special_instructions): # String already exists, update with latest channel name. updated_string = pattern.sub( rf'\1{new_channel}\3\4', special_instructions) # noqa: E501 elif special_instructions: # Add newline and our string instruction = pattern.sub(rf'\1{new_channel}\3', SAMPLE_VEVO_STRING) updated_string = special_instructions + '\n' + instruction else: # Null string - just add ours updated_string = pattern.sub(rf'\1{new_channel}\3', SAMPLE_VEVO_STRING) updated_string.replace( 'was received from SME but not available', f'was received from {ddex_provider} but not available' ) if context_special_instructions and updated_string: # We saw instances where both instructions were the same # The absense of this second if effectively doubled the instructions. if context_special_instructions not in updated_string: return context_special_instructions + '\n' + updated_string return updated_string or context_special_instructions def get_error_message(err: graphql.GraphQLError) -> str: """Retrieve error message body from GraphQL error.""" logger.info(f'Reading error message from: {str(err)}') return str(err) def handle_product_code_collision(context, s3_context): """Find a product code that is not in use and update the product. https://theorchard.atlassian.net/browse/SWITCH-3588 """ logger.info('Starting video product code collision logic') i = 1 max_iter = 300 product_code_found = False original_product_code = s3_context.video.product_code while i <= max_iter and not product_code_found: new_product_code = original_product_code + '-' + str(i) s3_context.video.product_code = new_product_code i += 1 try: logger.info(f'Trying with product code {new_product_code}') update_video_metadata(context, s3_context) product_code_found = True logger.info(f'Successful operation with code {new_product_code}') except graphql.GraphQLError as err: if 'productCodeInUse' not in get_error_message(err): raise # Wait after each retry to avoid causing a load spike sleep(0.1) if not product_code_found: raise Exception('Could not find a new product code') def check_remapped_product_code(s3_context: S3Context, graphql_result: Dict): """If we added "...-1" to a product code before we should use the new code. Product code is not used by other Lambdas so no need to write to S3. """ gql_product_code = graphql_result.get('productCode') ddex_product_code = s3_context.video.product_code if gql_product_code and (ddex_product_code + '-') in gql_product_code: s3_context.video.product_code = gql_product_code def retrieve_label_participants(label_participants, participants): """Retrieve project artist from artist list in S3 data.""" if label_participants and participants: for label_participant in label_participants: if label_participant.name == participants.name: return label_participant class ReleaseCorrectionValuePair(NamedTuple): """Contains the current and incoming values for a DDEX field.""" new: any old: any