"""send-sme-ack.""" import os import re import uuid from datetime import datetime from xml.dom.minidom import parseString import boto3 import requests from ddex_ingester_common.constants.ddex_providers import \ SME_ANALYTICS_PROVIDER from ddex_ingester_common.constants.send_acknowledgement import ( SME_ERROR_STATUS, SME_SUCCESS_STATUS) from ddex_ingester_common.helpers.switchboard import generate_switchboard_jwt from ddex_ingester_common.logging import utils as logging_utils from ddex_ingester_common.schemas.state_machine_schema import \ StateMachineSchema from ddex_ingester_common.send_acknowledgement import ddex_scsm from marshmallow.utils import get_value from requests import auth import config from config import SEND_SME_ACK, secrets_manager_client from constants.queries import REGISTER_DDEX_INGESTION_RESULT from constants.status import (INGESTION_FAILURE_STATUS, SWITCHBOARD_ERROR_STATUS, SWITCHBOARD_SUCCESS_STATUS) logger = logging_utils.get_logger(config.app_logger) def handler(event, context): """Lambda entry point.""" grid = None upc = None timestamp = None delivery_location = None message_id = None try: context = StateMachineSchema().load(event.get('context')) 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 ) grid = context.product.grid upc = context.product.upc timestamp = context.execution_start_time message_id = context.message_id delivery_location = context.key except Exception: logger.info('Context not found in event payload.') # It's possible for the context to be double nested inside the event event_context = event.get('context') event_context_2 = event.get('context', {}).get('context') event_context = event_context_2 if event_context_2 else event_context key = get_value( event_context, 'detail.requestParameters.key', None) bucket = get_value( event_context, 'detail.requestParameters.bucketName', None) logger.info(f'Bucket: {bucket} Key: {key}') delivery_location = key s3_client = boto3.client('s3') file = s3_client.get_object(Bucket=bucket, Key=key) try: doc = parseString(file.get('Body').read()) message_id_node = doc.getElementsByTagName('MessageId')[0] message_id = message_id_node.firstChild.data release_list = doc.getElementsByTagName('Release') for release in release_list: if release.getAttribute('IsMainRelease'): release_id = release.getElementsByTagName('ReleaseId')[0] grid_node = release_id.getElementsByTagName('GRid')[0] upc_node = release_id.getElementsByTagName('ICPN')[0] grid = grid_node.firstChild.data upc = upc_node.firstChild.data break except Exception: logger.info(f'{key} is a malformed XML or not an XML file.') if not grid: _, filename = os.path.split(key) matches = re.match('A10[0-9A-Z]{15}', filename) grid = matches.group(0) # Assume failure if we get here and no status is in the event status = event.get('status', INGESTION_FAILURE_STATUS) sme_status = ( SME_ERROR_STATUS if status == INGESTION_FAILURE_STATUS else SME_SUCCESS_STATUS) logger.info( f'Going to call SME with grid: {grid} and status: {sme_status}.') scsm_generator = ddex_scsm.DDEXSupplyChainStatusMessage(grid, sme_status) xml_data = scsm_generator.output_xml() # remove context.ddex_provide == SME_ANALYTICS_PROVIDER after validation if SEND_SME_ACK or context.ddex_provider == SME_ANALYTICS_PROVIDER: response = send_ack(xml_data.decode('UTF-8'), context.ddex_provider) if response.status_code != 200: err_msg = ( 'Call to SME DDEX WS failed. ' f'Response: {response.content}') logger.error(err_msg) raise SystemError(err_msg) logger.info('Call to SME DDEX WS succeeded.') else: logger.info('Call to SME DDEX WS skipped.') if not context.has_switchboard_deal: logger.info('Skipping registering result due to no Switchboard deal') return if upc: logger.info('Going to call Switchboard with DDEX result') register_ddex_result(upc, status, timestamp, message_id, delivery_location) else: err_msg = ( 'No UPC parsed to be able to call Switchboard with DDEX result') logger.error(err_msg) raise SystemError(err_msg) def send_ack(xml_data, ddex_provider): """Send acknowledge email with XML attachment. Args: xml_data (str): ddex_provider (str): Returns: requests.Response """ if ddex_provider == SME_ANALYTICS_PROVIDER: sme_ddexws_user = secrets_manager_client.get_cred( 'SME_ANALYTICS_PROVIDER_USER') sme_ddexws_passwd = secrets_manager_client.get_cred( 'SME_ANALYTICS_PROVIDER_PASSWD') else: sme_ddexws_user = secrets_manager_client.get_cred('SME_DDEXWS_USER') sme_ddexws_passwd = secrets_manager_client.get_cred( 'SME_DDEXWS_PASSWD') session = requests.Session() request = requests.Request( method='post', url=config.SME_DDEXWS_HOST, auth=auth.HTTPBasicAuth( username=sme_ddexws_user, password=sme_ddexws_passwd ), headers={'Content-Type': 'application/xml'}, data=xml_data ).prepare() return session.send(request) def register_ddex_result(upc: str, status: str, timestamp: str, message_id: str, delivery_location: str): """Call register ddex result switchboard graphql mutation.""" # Generate JWT for this request at runtime instead of in Config as this # function may not be called in some scenarios. config.graphql_switchboard.set_headers( {'x-token': generate_switchboard_jwt(config.SWITCHBOARD_ROLE_ARN, config.SWITCHBOARD_SECRET_ARN, config.APPLICATION_NAME)}) status = ( SWITCHBOARD_ERROR_STATUS if status == INGESTION_FAILURE_STATUS else SWITCHBOARD_SUCCESS_STATUS) response = config.graphql_switchboard.execute( REGISTER_DDEX_INGESTION_RESULT, { 'input': { 'upc': upc, 'status': status, 'timestamp': timestamp or datetime.utcnow().isoformat(), 'ddexMessageId': message_id or 'N/A', 'deliveryLocation': delivery_location or 'N/A' } } ) logger.info(f'Switchboard GraphQL response: {response}')