"""Lambda function module.""" from time import sleep from typing import Dict import uuid import config from config import graphql_gateway from constants import queries, sql_queries from constants.constants import ( MUSIC_ALBUM, MUSIC_ALBUM_MID_FRONT_TIER, MUSIC_TRACK, MUSIC_TRACK_FRONT_TIER, ) from ddex_ingester_common.constants.catalog_ingestion import ( INSERT_ACTION, PRODUCT_ENTITY_TYPE, SUCCESS, UPDATE_ACTION, ) from ddex_ingester_common.constants.ddex_providers import SME, \ SME_ANALYTICS_PROVIDER from ddex_ingester_common.constants.format_mapping import FORMAT_MAPPING from ddex_ingester_common.helpers.catalog_ingestion import ( CatalogIngestionAction, ) from ddex_ingester_common.helpers.rds import run_rds_query from ddex_ingester_common.helpers.s3_ddex import load_ddex_json from ddex_ingester_common.lambda_exceptions import SetProductException 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 import ( is_release_correction) from ddex_ingester_common.release_correction.release_corrections_s3 import ( load_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 from release_correction import ( diff_for_release_corrections, update_product_nfd, ) from utils import ( get_participations, get_title_localizations, ) logger = logging_utils.get_logger(config.app_logger) def handler(event, context): """Set product handler.""" logger.info(f'Triggered set_product: {event}') context = StateMachineSchema().load(event) s3_data = S3Schema().load(load_ddex_json(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, context.message_id, context.message_thread_id, context.execution_name ) graphql_gateway.set_headers( { 'Orchard-User-Id': config.OA_USER, 'Correlation-Id': correlation_id, } ) try: graphql_result = check_for_product(context) if not graphql_result: try: create_product(context, s3_data) except graphql.GraphQLError as err: product_code_collision = \ get_message_error_code(err) == 'product_code_used' if context.ddex_provider != SME and product_code_collision: handle_product_code_collision(context, s3_data, create=True) else: raise else: if context.ddex_provider != SME: check_remapped_product_code(s3_data, graphql_result) context.product.status = graphql_result.get('status') # 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): s3_release_corrections = load_rc_json(event) diff_for_release_corrections( context, s3_data, graphql_result, s3_release_corrections) # 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_product(context, s3_data) else: gql_nfd = graphql_result.get('notForDistribution') if context.product.not_for_distribution != gql_nfd: update_product_nfd(context) except graphql.GraphQLError as err: message = get_message_error_code(err) if context.ddex_provider == SME: raise SetProductException(message) from err elif message == 'product_code_used': handle_product_code_collision(context, s3_data, create=False) else: raise return StateMachineSchema().dump(context) def check_for_product(context: StateMachineContext) -> Dict: """Check for existence of a product.""" upc = context.product.upc if not upc: logger.info('No UPC found, skipped check_for_product') return {} 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 create_product(context: StateMachineContext, s3_data: S3Context) -> Dict: """Create a product from given params.""" release_type = FORMAT_MAPPING.get(context.product.release_type) if not release_type: raise ValueError(f'Unknown format: {context.product.release_type}') participations = get_participations( s3_data.product.display_artists, s3_data.label_participants, s3_data.product.genres, context ) payload = { 'data': { 'productName': s3_data.product.product_name, 'productCode': s3_data.product.catalog_number, 'productHighlights': 'abc', 'projectId': context.project_id, 'accountId': context.product.vendor_id, 'subaccountId': context.product.subaccount_id, 'upc': context.product.upc, 'metaLanguage': s3_data.product.metadata_language, 'deliveredVersion': s3_data.product.product_version, 'participations': participations, 'pLine': s3_data.product.p_line, 'cLine': s3_data.product.c_line, 'genreId': context.product.genre_id, 'subgenreId': context.product.subgenre_id, 'specialInstructions': sanitize_special_instructions( context.product.special_instructions), 'productLocalizations': get_title_localizations(s3_data.product), 'format': release_type, 'imprint': s3_data.product.imprint, 'notForDistribution': context.product.not_for_distribution, 'version': s3_data.product.product_version_notes, 'vendorReleaseIdentifier': s3_data.product.proprietary_id, 'manufacturerUpc': ( s3_data.product.manufacturer_upc if s3_data.product.manufacturer_upc else context.product.upc), } } logger.info(f'Running create product with payload:\n{payload}') result = graphql_gateway.execute( queries.CREATE_PRODUCT, payload )['data']['createProduct'] if result: product_id = result['productId'] context.product.product_id = product_id if not context.product.upc: context.product.upc = result['upc'] # check if upc is placeholder upc # if yes , store it in db to be reused in # replace_with_placeholder_upc lambda if context.ddex_provider == SME_ANALYTICS_PROVIDER and \ context.product.display_upc is not None and \ context.product.display_upc != context.product.upc: update_display_upc(product_id, context.product.display_upc) store_placeholder_upc(context, context.product.display_upc, context.product.upc) save_catalog_ingestion_action( context, s3_data, INSERT_ACTION ) # Set default values for product and track pricing # This is not done in set_pricing as we only want to do this once set_default_pricing(product_id) return result def update_product( context: StateMachineContext, s3_data: S3Context) -> Dict: """Update existing product.""" participations = get_participations( s3_data.product.display_artists, s3_data.label_participants, s3_data.product.genres, context ) release_type = FORMAT_MAPPING.get(context.product.release_type) if not release_type: raise ValueError(f'Unknown format: {context.product.release_type}') payload = { 'data': { 'productName': s3_data.product.product_name, 'productCode': s3_data.product.catalog_number, 'productId': context.product.product_id, 'metaLanguage': s3_data.product.metadata_language, 'deliveredVersion': s3_data.product.product_version, 'participations': participations, 'pLine': s3_data.product.p_line, 'genreId': context.product.genre_id, 'subgenreId': context.product.subgenre_id, 'specialInstructions': sanitize_special_instructions( context.product.special_instructions), 'productLocalizations': get_title_localizations(s3_data.product), 'format': release_type, 'imprint': s3_data.product.imprint, 'notForDistribution': context.product.not_for_distribution, 'vendorReleaseIdentifier': s3_data.product.proprietary_id, 'manufacturerUpc': ( s3_data.product.manufacturer_upc if s3_data.product.manufacturer_upc else context.product.upc), } } c_line = s3_data.product.c_line if c_line: payload['data']['cLine'] = c_line product_version_notes = s3_data.product.product_version_notes if product_version_notes: payload['data']['version'] = product_version_notes logger.info(f'Running update product with payload:\n{payload}') result = graphql_gateway.execute( queries.UPDATE_PRODUCT, payload )['data']['updateProduct'] save_catalog_ingestion_action( context, s3_data, UPDATE_ACTION ) return result def sanitize_special_instructions(special_instructions: str) -> str: """Sanitizes special instructions 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_instructions = special_instructions if special_instructions and '../' in special_instructions: sanitized_instructions = special_instructions.replace('../', '').strip() return sanitized_instructions def format_update_pricing_payload( product_id: int, pricing_family: int, orchard_pricing_tier_id: int) -> Dict: """Format pricing payload. Args: product_id (int): Product ID pricing_family (str): Pricing Family orchard_pricing_tier_id (int): Orchard pricing tier ID Returns: dict """ payload = { 'data': { 'productId': product_id, 'orchardPricingTier': orchard_pricing_tier_id, 'pricingFamily': pricing_family } } return payload def set_default_pricing(product_id: int): """Set default pricing for product and track.""" payload = format_update_pricing_payload( product_id=product_id, pricing_family=MUSIC_ALBUM, orchard_pricing_tier_id=MUSIC_ALBUM_MID_FRONT_TIER) logger.info( f'Running update product pricing tier with payload:\n{payload}') graphql_gateway.execute( queries.UPDATE_PRODUCT_PRICING_TIER, payload) payload = format_update_pricing_payload( product_id=product_id, pricing_family=MUSIC_TRACK, orchard_pricing_tier_id=MUSIC_TRACK_FRONT_TIER) logger.info( f'Running update track pricing tier with payload:\n{payload}') graphql_gateway.execute( queries.UPDATE_PRODUCT_PRICING_TIER, payload) def save_catalog_ingestion_action( context: StateMachineContext, s3_data: S3Context, action: str): """Save catalog ingestion action to S3.""" action_string = 'updated' if action == UPDATE_ACTION else 'created' message = ( f'Product with release_id {context.product.product_id} ' f'{action_string} successfully.') config.catalog_ingestion_session.add( CatalogIngestionAction( state_machine_name=context.state_machine_name, state_machine_execution_name=context.execution_name, action=action, entity_type=PRODUCT_ENTITY_TYPE, result=SUCCESS, time_created=context.execution_start_time, project_code=s3_data.project.project_code, project_id=context.project_id, project_name=s3_data.project.name, upc=context.product.upc, release_id=context.product.product_id, release_name=s3_data.product.product_name, vendor_catalog_number=s3_data.product.catalog_number, message=message, ) ) config.catalog_ingestion_session.save() def get_message_error_code(err: graphql.GraphQLError) -> str: """Retrieve error message body from GraphQL error.""" error_body = err.get_response_body() logger.info(f'Reading error message from: {error_body}') if not isinstance(error_body, dict): return error_body error_message = error_body.get('message') if not isinstance(error_message, dict): return error_message code = error_message.get('product_code') if not isinstance(code, dict): return code return code.get('error_code') def handle_product_code_collision(context, s3_data, create=False): """Find a product code that is not in use and update/create the product. https://theorchard.atlassian.net/browse/SWITCH-3588 """ logger.info('Starting product code collision logic') i = 1 max_iter = 300 product_code_found = False original_catalog_number = s3_data.product.catalog_number while i <= max_iter and not product_code_found: new_catalog_number = original_catalog_number + '-' + str(i) s3_data.product.catalog_number = new_catalog_number i += 1 try: logger.info(f'Trying with product code {new_catalog_number}') if create: create_product(context, s3_data) else: update_product(context, s3_data) product_code_found = True logger.info(f'Successful operation with code {new_catalog_number}') except graphql.GraphQLError as err: if get_message_error_code(err) != 'product_code_used': raise # Wait after each retry to avoid causing a load spike sleep(0.1) if not product_code_found: raise SetProductException('Could not find a new product code') def check_remapped_product_code(s3_data: 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_data.product.catalog_number if gql_product_code and (ddex_product_code + '-') in gql_product_code: s3_data.product.catalog_number = gql_product_code def store_placeholder_upc(context, upc, placeholder_upc): """Store upc to placeholder upc mapping.""" query_args = ( upc, placeholder_upc, context.product.grid, context.key ) run_rds_query( logger, config.RDS_HOST, config.RDS_DB_NAME, config.RDS_USER, config.RDS_PASSWORD, sql_queries.INSERT_PLACEHOLDER_UPC, query_args, ) def update_display_upc(product_id, display_upc): """Update products display_upc. Unsafe! Use only for ddex product creations with placeholder_upc approach """ logger.info( f'Updating display upc to {display_upc} for product {product_id}') update_upc_response = config. \ ows_client.put('ows-product-digital', path=f'/product/{product_id}/display_upc', json={'display_upc': display_upc}) if update_upc_response.status_code != 200: raise Exception('Unable to update display_upc for the product' f'error message : {update_upc_response.text}')