AWSTemplateFormatVersion: "2010-09-09"
Description: "This is the template for creating a solution to analyze Amazon Connect Lex logs using Amazon Athena as described in the blog"
Parameters:
    ProjectName:
        Type: "String"
        Default: "test123"
        Description: Enter an all-lowercase single-word name which is between 3 and 42 chars long. An S3 bucket will be created with this name to store all collected data.  
        MinLength : 3
        MaxLength : 42
        AllowedPattern : "^[a-z0-9]*$"
    BotName:
        Type: "String"
        Default: "SampleBot"
        Description: Enter the name of the Lex Bot that will be analyzed in combination with other Amazon Connect records.  
    CreateANewGlueDatabaseAndS3BucketForTheProject:
        Type: "String"
        Default: "Use existing AWS Glue database and Amazon S3 data collection bucket created for this project"
        AllowedValues:
          - "Create a new AWS Glue database and Amazon S3 data collection bucket for this project"
          - "Use existing AWS Glue database and Amazon S3 data collection bucket created for this project"
        Description: Choose whether to create a AWS Glue database and an Amazon S3 bucket for this project. If there is an existing bucket named '{ProjectName}' and a database named '{ProjectName}', then choose to reuse them. 

Conditions:
    CreateCommonGlueDatabaseAndDestinationBucket: !Equals 
        - !Ref CreateANewGlueDatabaseAndS3BucketForTheProject
        - Create a new AWS Glue database and Amazon S3 data collection bucket for this project
        
Resources:
    DataDestinationBucket:
        Type: "AWS::S3::Bucket"
        Condition: CreateCommonGlueDatabaseAndDestinationBucket
        DeletionPolicy: Retain
        Properties:
            BucketName: !Sub "${ProjectName}"

    GlueDatabase:
        Type: "AWS::Glue::Database"
        Condition: CreateCommonGlueDatabaseAndDestinationBucket
        DependsOn: DataDestinationBucket
        DeletionPolicy: Retain
        Properties:
            DatabaseInput:
                Name: !Sub "${ProjectName}"
            CatalogId: !Ref "AWS::AccountId"

    DbAndBucketCreationWaitHandle: 
        Condition: CreateCommonGlueDatabaseAndDestinationBucket
        DependsOn: GlueDatabase 
        Type: "AWS::CloudFormation::WaitConditionHandle"

    WaitHandle: 
        Type: "AWS::CloudFormation::WaitConditionHandle"

    DbAndBucketWaitCondition: 
        Type: "AWS::CloudFormation::WaitCondition"
        Properties: 
            Handle: !If [CreateCommonGlueDatabaseAndDestinationBucket, !Ref DbAndBucketCreationWaitHandle, !Ref WaitHandle]
            Timeout: "1"
            Count: 0

    
    GrantPermissionToPassIAMRole:
        Type: "AWS::IAM::ManagedPolicy"
        DependsOn: LogGroupForLexBot
        Properties:
            ManagedPolicyName: !Sub "${ProjectName}-LexPass-${BotName}"
            Path: "/"
            PolicyDocument: !Sub |
                {
                    "Version": "2012-10-17",
                    "Statement": [
                        {
                            "Effect": "Allow",
                            "Action": "iam:PassRole",
                            "Resource": "arn:aws:iam::${AWS::AccountId}:role/${ProjectName}-LexLogRole-${BotName}",
                            "Condition": {
                                "StringEquals": {
                                    "iam:PassedToService": "lex.amazonaws.com"
                                },
                                "StringLike": {
                                    "iam:AssociatedResourceARN": "arn:aws:lex:${AWS::Region}:${AWS::AccountId}:bot:${BotName}:*"
                                }
                            }
                        }
                    ]
                }

    LexPolicyToWriteLexLogs:
        Type: "AWS::IAM::ManagedPolicy"
        DependsOn: LogGroupForLexBot
        Properties:
            ManagedPolicyName: !Sub "${ProjectName}-LexCW-${BotName}"
            Path: "/"
            PolicyDocument: !Sub |
                {
                    "Version": "2012-10-17",
                    "Statement": [
                        {
                            "Sid": "VisualEditor0",
                            "Effect": "Allow",
                            "Action": [
                                "logs:CreateLogStream",
                                "logs:PutLogEvents"
                            ],
                            "Resource": "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws-lex/${ProjectName}/${BotName}:*"
                        }
                    ]
                }

    BotRoleForLexLogs:
        Type: "AWS::IAM::Role"
        Properties:
            Path: "/"
            RoleName: !Sub "${ProjectName}-LexLogRole-${BotName}"
            AssumeRolePolicyDocument: |
              {
                "Version": "2012-10-17",
                "Statement": [
                  {
                    "Effect": "Allow",
                    "Principal": {
                      "Service": "lex.amazonaws.com"
                    },
                    "Action": "sts:AssumeRole"
                  }
                ]
              }
            MaxSessionDuration: 3600
            ManagedPolicyArns:
              - !Ref GrantPermissionToPassIAMRole
              - !Ref LexPolicyToWriteLexLogs


    LogGroupForLexBot:
        Type: "AWS::Logs::LogGroup"
        DependsOn: KinesisFirehoseDeliveryStreamForLex
        Properties:
            LogGroupName: !Sub "/aws-lex/${ProjectName}/${BotName}"      

    LogSubscriptionRole:
        Type: "AWS::IAM::Role"
        DependsOn: LogGroupForLexBot
        Properties:
            Path: "/"
            RoleName: !Sub "${ProjectName}-LexLogSubs-${BotName}"
            AssumeRolePolicyDocument: !Sub "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Service\":\"logs.${AWS::Region}.amazonaws.com\"},\"Action\":\"sts:AssumeRole\"}]}"
            MaxSessionDuration: 3600
            ManagedPolicyArns:
              - !Ref PutLexLog2DeliveryStreamPolicy
            Description: ""

    LogsSubscriptionFilter:
        Type: "AWS::Logs::SubscriptionFilter"
        DependsOn: KinesisFirehoseDeliveryStreamForLex
        Properties:
            LogGroupName: !Sub "/aws-lex/${ProjectName}/${BotName}"
            FilterPattern: ""
            DestinationArn: !Sub "arn:aws:firehose:${AWS::Region}:${AWS::AccountId}:deliverystream/${KinesisFirehoseDeliveryStreamForLex}"
            RoleArn: !GetAtt LogSubscriptionRole.Arn

    PutLexLog2DeliveryStreamPolicy:
        Type: "AWS::IAM::ManagedPolicy"
        DependsOn: KinesisFirehoseDeliveryStreamForLex
        Properties:
            ManagedPolicyName: !Sub "${ProjectName}-Lexlogs2KFH-${BotName}"
            Path: "/"
            PolicyDocument: !Sub |
                {
                    "Version": "2012-10-17",
                    "Statement": [
                        {
                            "Sid": "VisualEditor0",
                            "Effect": "Allow",
                            "Action": [
                                "firehose:PutRecord",
                                "firehose:PutRecordBatch"
                            ],
                            "Resource": "arn:aws:firehose:${AWS::Region}:${AWS::AccountId}:deliverystream/${KinesisFirehoseDeliveryStreamForLex}"
                        },
                        {
                            "Action": "iam:PassRole",
                            "Resource": "arn:aws:iam::${AWS::AccountId}:role/${ProjectName}-LexLogSubs-${BotName}",
                            "Effect": "Allow"
                        }
                    ]
                }

 
    FirehoseTransformationLambdaForLexBasicPolicy:
        Type: "AWS::IAM::ManagedPolicy"
        DependsOn: DbAndBucketWaitCondition
        Properties:
            ManagedPolicyName: !Sub "${ProjectName}-LambdaLex-${BotName}"
            Path: "/"
            PolicyDocument: !Sub |
                {
                    "Version": "2012-10-17",
                    "Statement": [
                        {
                            "Action": [
                                "logs:CreateLogStream",
                                "logs:CreateLogGroup",
                                "logs:PutLogEvents"
                            ],
                            "Resource": "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/*",
                            "Effect": "Allow",
                            "Sid": "WriteCloudWatchLogs"
                        }
                    ]
                }

    FirehoseTransformationLambdaRoleForLex:
        Type: "AWS::IAM::Role"
        DependsOn: 
            - GlueTableForLex
        Properties:
            Path: "/service-role/"
            RoleName: !Sub "${ProjectName}-LambdaLex-${BotName}"
            AssumeRolePolicyDocument: !Sub |
                {
                    "Version":"2012-10-17",
                    "Statement":[
                        {
                            "Effect":"Allow",
                            "Principal":{
                                "Service":"lambda.amazonaws.com"
                                },
                            "Action":"sts:AssumeRole"
                        }
                    ]
                }
            MaxSessionDuration: 3600
            ManagedPolicyArns:
              - !Ref FirehoseTransformationLambdaForLexBasicPolicy

    TransformationLambdaFunctionForLex:
        Type: "AWS::Lambda::Function"
        Properties:
            Description: "Kinesis firehose uses this processor for custom transformation logic."
            FunctionName: !Sub "${ProjectName}-TransformLex-${BotName}"
            Handler: "index.lambda_handler"
            Code:
                ZipFile: |
                    import base64
                    import json
                    import gzip
                    from io import BytesIO
                    import boto3

                    def transformLexLogKeyValueArray(parent, nameOfKeyValueArray):
                        if nameOfKeyValueArray in parent:
                            parent[nameOfKeyValueArray] = [ {"key":key,"value":value} for key, value in parent[nameOfKeyValueArray].items()]


                    def transformV1Log(message):
                        v2message = { 
                            'bot':{}, 
                            'sessionState': {
                                'dialogAction':{},
                                'intent':{} 
                            } 
                        }
                        if 'messageVersion' in message:
                            v2message['message-version']=message['messageVersion']
                        if 'botName' in message:
                            v2message['bot']['name']=message['botName']
                        if 'botAlias' in message:
                            v2message['bot']['aliasName']=message['botAlias']
                        if 'botVersion' in message:
                            v2message['bot']['version']=message['botVersion']
                        if 'locale' in message:
                            v2message['bot']['localeId']=message['locale']       
                        if 'inputTranscript' in message:
                            v2message['inputTranscript']=message['inputTranscript']
                        if 'intent' in message:
                            v2message['sessionState']['intent']['name']=message['intent']
                        if 'nluIntentConfidence' in message:
                            v2message['nluconfidence']=message['nluIntentConfidence']
                        if 'slots' in message:
                            v2message['sessionState']['intent'] ['slots'] = [{ 'key': key, 'interpretedValue':value} for key, value in message['slots'].items()]
                        ## ols user id becomes v2 session id
                        if 'userId' in message:
                            v2message['sessionId']=message['userId']
                        if 'developerOverride' in message:
                            v2message['developerOverride']=message['developerOverride']
                        if 'missedUtterance' in message:
                            v2message['missedUtterance']=message['missedUtterance']
                        if 'requestId' in message:
                            v2message['requestId']=message['requestId']
                        if 'inputDialogMode' in message:
                            v2message['inputMode']=message['inputDialogMode']
                        if 'dialogState' in message:
                            v2message['sessionState']['dialogAction']['type']=message['dialogState']
                        if 'slotToElicit' in message:
                            v2message['sessionState']['dialogAction']['slotToElicit']=message['slotToElicit']
                        if 's3PathForAudio' in message:
                            v2message['audioProperties']= {
                                "s3Path": message['s3PathForAudio'],
                            }
                        if 'timestamp' in message:
                            v2message['timestamp']=message['timestamp']
                        if 'kendraResponse' in message:
                            v2message['sessionState']['intent']['kendraResponse']=message['kendraResponse']
                        if 'requestAttributes' in message:
                            v2message['requestAttributes']=[ {"key":key,"value":value} for key, value in message['requestAttributes'].items()]
                        if 'sessionAttributes' in message:
                            v2message['sessionState']['sessionAttributes']=[ {"key":key,"value":value} for key, value in message['sessionAttributes'].items()]
                        if 'botResponse' in message:
                            v2message['messages']= [
                                {
                                    "content": message['botResponse']
                                }
                            ]
                        ## Fix json issue at the sentiment and bring the data to the root level
                        if 'sentimentResponse' in message:
                            sentimentText= message['sentimentResponse']['sentimentScore']    
                            sentimentText= sentimentText.replace("Positive",'"Positive"')
                            sentimentText= sentimentText.replace("Negative",'"Negative"')
                            sentimentText= sentimentText.replace("Neutral",'"Neutral"')
                            sentimentText= sentimentText.replace("Mixed",'"Mixed"')
                            sentimentResults=json.loads(sentimentText)
                            v2message['sentimentpositive']=sentimentResults['Positive']
                            v2message['sentimentnegative']=sentimentResults['Negative'] 
                            v2message['sentimentneutral']=sentimentResults['Neutral'] 
                            v2message['sentimentmixed']=sentimentResults['Mixed'] 
                            v2message['sentiment']=message['sentimentResponse']['sentimentLabel']
                        ## print(json.dumps(v2message))
                        return v2message

                    def transformV2Log(message):
                        if 'sessionState' in message:
                            ## session attributes are tarnsformed to be database friendly
                            transformLexLogKeyValueArray(message['sessionState'], 'sessionAttributes')            
                            if 'intent' in message['sessionState']:
                                ## intent has the slots that needs to be transformed to be database friendly
                                intent= message['sessionState']['intent'] 
                                if 'slots' in intent:
                                    newSlots=[]
                                    for key, value in intent['slots'].items():
                                        if value: 
                                            if 'value' in value:
                                                newSlot = { 'key': key }
                                                if 'shape' in value:
                                                    newSlot['shape'] = value['shape']
                                                slotValue= value['value']
                                                if 'originalValue' in slotValue:
                                                    newSlot['originalValue'] = slotValue['originalValue']
                                                if 'interpretedValue' in slotValue:
                                                    newSlot['interpretedValue'] = slotValue['interpretedValue']
                                                if 'resolvedValues' in slotValue:
                                                    newSlot['resolvedValues'] = slotValue['resolvedValues']
                                                newSlots.append(newSlot)
                                            if 'values' in value:
                                                for slot in value['values']:
                                                    newSlot = { 'key': key }
                                                    if 'shape' in slot:
                                                        newSlot['shape'] = slot['shape']
                                                    slotValue= slot['value']
                                                    if 'originalValue' in slotValue:
                                                        newSlot['originalValue'] = slotValue['originalValue']
                                                    if 'interpretedValue' in slotValue:
                                                        newSlot['interpretedValue'] = slotValue['interpretedValue']
                                                    if 'resolvedValues' in slotValue:
                                                        newSlot['resolvedValues'] = slotValue['resolvedValues']
                                                    newSlots.append(newSlot)
                                        else: 
                                            newSlots.append({ 'key': key , 'value': None })
                                    intent['slots'] = newSlots
                        ## we only get the nlu and sentiment information from interpretations
                        if 'interpretations' in message:
                            if len (message['interpretations'])>0:
                                interpretation = message['interpretations'][0]
                                if 'nluConfidence' in interpretation:
                                    message['nluConfidence'] = interpretation['nluConfidence']
                                if 'sentimentResponse' in interpretation:
                                    sentimentResponse=interpretation['sentimentResponse']
                                    if 'sentimentScore' in sentimentResponse:
                                        sentimentResults = sentimentResponse['sentimentScore']
                                        message['sentimentpositive']=sentimentResults['positive']
                                        message['sentimentnegative']=sentimentResults['negative'] 
                                        message['sentimentneutral']=sentimentResults['neutral'] 
                                        message['sentimentmixed']=sentimentResults['mixed'] 
                                    if 'sentiment' in sentimentResponse:
                                        message['sentiment']=sentimentResponse['sentiment']    
                            #del message['interpretations']
                        ## transform requestAttributes into database friendly format 
                        if 'requestAttributes' in message:
                            requestAttributes = message['requestAttributes']
                            if requestAttributes:
                                if (type(requestAttributes) is dict):
                                    message['requestAttributes'] = [ {"key":key,"value":value} for key, value in requestAttributes.items()]
                                else:    
                                    message['requestAttributes'] = [ {"key":key,"value":value} for key, value in json.loads(requestAttributes).items()]
                        ## print (json.dumps(message, indent=2))
                        return message
                    def transformLogEvent(log_event):
                        ## print(log_event['message'] + '\n')
                        message = json.loads(log_event['message'])
                        if ('bot' in message and 'id' in message['bot']):
                            return json.dumps(transformV2Log(message)) + '\n'
                        return json.dumps(transformV1Log(message)) + '\n'



                    def processRecords(records):
                        for r in records:
                            data = base64.b64decode(r['data'])
                            striodata = BytesIO(data)
                            with gzip.GzipFile(fileobj=striodata, mode='r') as f:
                                data = json.loads(f.read())

                            recId = r['recordId']
                            """
                            CONTROL_MESSAGE are sent by CWL to check if the subscription is reachable.
                            They do not contain actual data.
                            """
                            if data['messageType'] == 'CONTROL_MESSAGE':
                                yield {
                                    'result': 'Dropped',
                                    'recordId': recId
                                }
                            elif data['messageType'] == 'DATA_MESSAGE':
                                joinedData = ''.join([transformLogEvent(e) for e in data['logEvents']])
                                dataBytes = joinedData.encode("utf-8")
                                encodedData = base64.b64encode(dataBytes)
                                if len(encodedData) <= 6000000:
                                    yield {
                                        'data': encodedData,
                                        'result': 'Ok',
                                        'recordId': recId
                                    }
                                else:
                                    yield {
                                        'result': 'ProcessingFailed',
                                        'recordId': recId
                                    }
                            else:
                                yield {
                                    'result': 'ProcessingFailed',
                                    'recordId': recId
                                }


                    def putRecordsToFirehoseStream(streamName, records, client, attemptsMade, maxAttempts):
                        failedRecords = []
                        codes = []
                        errMsg = ''
                        # if put_record_batch throws for whatever reason, response['xx'] will error out, adding a check for a valid
                        # response will prevent this
                        response = None
                        try:
                            response = client.put_record_batch(DeliveryStreamName=streamName, Records=records)
                        except Exception as e:
                            failedRecords = records
                            errMsg = str(e)

                        # if there are no failedRecords (put_record_batch succeeded), iterate over the response to gather results
                        if not failedRecords and response and response['FailedPutCount'] > 0:
                            for idx, res in enumerate(response['RequestResponses']):
                                # (if the result does not have a key 'ErrorCode' OR if it does and is empty) => we do not need to re-ingest
                                if 'ErrorCode' not in res or not res['ErrorCode']:
                                    continue

                                codes.append(res['ErrorCode'])
                                failedRecords.append(records[idx])

                            errMsg = 'Individual error codes: ' + ','.join(codes)

                        if len(failedRecords) > 0:
                            if attemptsMade + 1 < maxAttempts:
                                print('Some records failed while calling PutRecordBatch to Firehose stream, retrying. %s' % (errMsg))
                                putRecordsToFirehoseStream(streamName, failedRecords, client, attemptsMade + 1, maxAttempts)
                            else:
                                raise RuntimeError('Could not put records after %s attempts. %s' % (str(maxAttempts), errMsg))


                    def putRecordsToKinesisStream(streamName, records, client, attemptsMade, maxAttempts):
                        failedRecords = []
                        codes = []
                        errMsg = ''
                        # if put_records throws for whatever reason, response['xx'] will error out, adding a check for a valid
                        # response will prevent this
                        response = None
                        try:
                            response = client.put_records(StreamName=streamName, Records=records)
                        except Exception as e:
                            failedRecords = records
                            errMsg = str(e)

                        # if there are no failedRecords (put_record_batch succeeded), iterate over the response to gather results
                        if not failedRecords and response and response['FailedRecordCount'] > 0:
                            for idx, res in enumerate(response['Records']):
                                # (if the result does not have a key 'ErrorCode' OR if it does and is empty) => we do not need to re-ingest
                                if 'ErrorCode' not in res or not res['ErrorCode']:
                                    continue

                                codes.append(res['ErrorCode'])
                                failedRecords.append(records[idx])

                            errMsg = 'Individual error codes: ' + ','.join(codes)

                        if len(failedRecords) > 0:
                            if attemptsMade + 1 < maxAttempts:
                                print('Some records failed while calling PutRecords to Kinesis stream, retrying. %s' % (errMsg))
                                putRecordsToKinesisStream(streamName, failedRecords, client, attemptsMade + 1, maxAttempts)
                            else:
                                raise RuntimeError('Could not put records after %s attempts. %s' % (str(maxAttempts), errMsg))


                    def createReingestionRecord(isSas, originalRecord):
                        if isSas:
                            return {'data': base64.b64decode(originalRecord['data']), 'partitionKey': originalRecord['kinesisRecordMetadata']['partitionKey']}
                        else:
                            return {'data': base64.b64decode(originalRecord['data'])}


                    def getReingestionRecord(isSas, reIngestionRecord):
                        if isSas:
                            return {'Data': reIngestionRecord['data'], 'PartitionKey': reIngestionRecord['partitionKey']}
                        else:
                            return {'Data': reIngestionRecord['data']}


                    def lambda_handler(event, context):
                        isSas = 'sourceKinesisStreamArn' in event
                        streamARN = event['sourceKinesisStreamArn'] if isSas else event['deliveryStreamArn']
                        region = streamARN.split(':')[3]
                        streamName = streamARN.split('/')[1]
                        records = list(processRecords(event['records']))
                        projectedSize = 0
                        dataByRecordId = {rec['recordId']: createReingestionRecord(isSas, rec) for rec in event['records']}
                        putRecordBatches = []
                        recordsToReingest = []
                        totalRecordsToBeReingested = 0

                        for idx, rec in enumerate(records):
                            if rec['result'] != 'Ok':
                                continue
                            projectedSize += len(rec['data']) + len(rec['recordId'])
                            # 6000000 instead of 6291456 to leave ample headroom for the stuff we didn't account for
                            if projectedSize > 6000000:
                                totalRecordsToBeReingested += 1
                                recordsToReingest.append(
                                    getReingestionRecord(isSas, dataByRecordId[rec['recordId']])
                                )
                                records[idx]['result'] = 'Dropped'
                                del(records[idx]['data'])

                            # split out the record batches into multiple groups, 500 records at max per group
                            if len(recordsToReingest) == 500:
                                putRecordBatches.append(recordsToReingest)
                                recordsToReingest = []

                        if len(recordsToReingest) > 0:
                            # add the last batch
                            putRecordBatches.append(recordsToReingest)

                        # iterate and call putRecordBatch for each group
                        recordsReingestedSoFar = 0
                        if len(putRecordBatches) > 0:
                            client = boto3.client('kinesis', region_name=region) if isSas else boto3.client('firehose', region_name=region)
                            for recordBatch in putRecordBatches:
                                if isSas:
                                    putRecordsToKinesisStream(streamName, recordBatch, client, attemptsMade=0, maxAttempts=20)
                                else:
                                    putRecordsToFirehoseStream(streamName, recordBatch, client, attemptsMade=0, maxAttempts=20)
                                recordsReingestedSoFar += len(recordBatch)
                                print('Reingested %d/%d records out of %d' % (recordsReingestedSoFar, totalRecordsToBeReingested, len(event['records'])))
                        else:
                            print('No records to be reingested')

                        return {"records": records}

            MemorySize: 128
            Role: !GetAtt FirehoseTransformationLambdaRoleForLex.Arn
            Runtime: "python3.9"
            Timeout: 60
            TracingConfig:
                Mode: "PassThrough"
   


    KinesisFirehoseServicePolicyForLex:
        Type: "AWS::IAM::ManagedPolicy"
        DependsOn: GlueTableForLex
        Properties:
            ManagedPolicyName: !Sub "${ProjectName}-KFH-Lex-${BotName}"
            Path: "/service-role/"
            PolicyDocument: !Sub |
                {
                    "Version": "2012-10-17",
                    "Statement": [
                        {
                            "Sid": "",
                            "Effect": "Allow",
                            "Action": [
                                "glue:GetTable",
                                "glue:GetTableVersion",
                                "glue:GetTableVersions"
                            ],
                            "Resource": [
                                "arn:aws:glue:${AWS::Region}:${AWS::AccountId}:catalog",
                                "arn:aws:glue:${AWS::Region}:${AWS::AccountId}:database/${ProjectName}",
                                "arn:aws:glue:${AWS::Region}:${AWS::AccountId}:table/${ProjectName}/${GlueTableForLex}"
                            ]
                        },
                        {
                            "Sid": "",
                            "Effect": "Allow",
                            "Action": [
                                "s3:AbortMultipartUpload",
                                "s3:GetBucketLocation",
                                "s3:GetObject",
                                "s3:ListBucket",
                                "s3:ListBucketMultipartUploads",
                                "s3:PutObject"
                            ],
                            "Resource": [
                                "arn:aws:s3:::${ProjectName}",
                                "arn:aws:s3:::${ProjectName}/*"
                            ]
                        },
                        {
                            "Sid": "",
                            "Effect": "Allow",
                            "Action": [
                                "lambda:InvokeFunction",
                                "lambda:GetFunctionConfiguration"
                            ],
                            "Resource": "arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:${TransformationLambdaFunctionForLex}"
                        },
                        {
                            "Sid": "",
                            "Effect": "Allow",
                            "Action": [
                                "logs:PutLogEvents"
                            ],
                            "Resource": [
                                "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/kinesisfirehose/${ProjectName}-Lex:log-stream:*"
                            ]
                        }                    ]
                }



    KinesisFirehoseServiceRoleForLex:
        Type: "AWS::IAM::Role"
        Properties:
            Path: "/service-role/"
            RoleName: !Sub "${ProjectName}-KFH-Lex-${BotName}"
            AssumeRolePolicyDocument: !Sub |
                {
                    "Version":"2012-10-17",
                    "Statement":[
                        {
                            "Effect":"Allow",
                            "Principal":{"Service":"firehose.amazonaws.com"},
                            "Action":"sts:AssumeRole"
                        }
                    ]
                }
            MaxSessionDuration: 3600
            ManagedPolicyArns:
              - !Ref KinesisFirehoseServicePolicyForLex

    KinesisFirehoseDeliveryStreamForLex:
        Type: "AWS::KinesisFirehose::DeliveryStream"
        DependsOn: 
            - TransformationLambdaFunctionForLex
        Properties:
            DeliveryStreamName: !Sub "${ProjectName}-Lex-${BotName}"
            DeliveryStreamType: "DirectPut"
            ExtendedS3DestinationConfiguration:
                BucketARN: !Sub "arn:aws:s3:::${ProjectName}"
                BufferingHints:
                    SizeInMBs: 128
                    IntervalInSeconds: 60
                CloudWatchLoggingOptions:
                    Enabled: true
                    LogGroupName: !Sub "/aws/kinesisfirehose/${ProjectName}-Lex"
                    LogStreamName: "DestinationDelivery"
                CompressionFormat: "UNCOMPRESSED"
                DataFormatConversionConfiguration:
                    SchemaConfiguration:
                        RoleARN: !GetAtt KinesisFirehoseServiceRoleForLex.Arn
                        DatabaseName: !Sub "${ProjectName}"
                        TableName: !Ref  GlueTableForLex
                        Region: !Ref AWS::Region
                        VersionId: "LATEST"
                    InputFormatConfiguration:
                        Deserializer:
                            OpenXJsonSerDe: {}
                    OutputFormatConfiguration:
                        Serializer:
                            ParquetSerDe: {}
                    Enabled: true
                EncryptionConfiguration:
                    NoEncryptionConfig: "NoEncryption"
                Prefix: !Sub "lex-base/${BotName}/year=!{timestamp:YYYY}/month=!{timestamp:MM}/day=!{timestamp:dd}/"
                ErrorOutputPrefix: !Sub "lex-erroroutputbase/${BotName}/!{timestamp:yyy/MM/dd}/!{firehose:random-string}/!{firehose:error-output-type}/"
                RoleARN: !GetAtt KinesisFirehoseServiceRoleForLex.Arn
                ProcessingConfiguration:
                    Enabled: true
                    Processors:
                      -
                        Type: "Lambda"
                        Parameters:
                          -
                            ParameterName: "LambdaArn"
                            ParameterValue: !GetAtt TransformationLambdaFunctionForLex.Arn 
                          -
                            ParameterName: "NumberOfRetries"
                            ParameterValue: "3"
                          -
                            ParameterName: "RoleArn"
                            ParameterValue: !GetAtt KinesisFirehoseServiceRoleForLex.Arn
                          -
                            ParameterName: "BufferSizeInMBs"
                            ParameterValue: "3"
                          -
                            ParameterName: "BufferIntervalInSeconds"
                            ParameterValue: "60"
                S3BackupMode: "Disabled"

    GlueTableForLex:
        Type: "AWS::Glue::Table"
        DependsOn: 
        - "DbAndBucketWaitCondition"
        Properties:
            DatabaseName: !Sub "${ProjectName}"
            CatalogId: !Ref "AWS::AccountId"
            TableInput:
                Owner: "hadoop"
                TableType: "EXTERNAL_TABLE"
                Parameters:
                    EXTERNAL: "TRUE"
                    "parquet.compression": "SNAPPY"
                    "projection.year.type": "integer"
                    "projection.year.range": "2022,2024"
                    "projection.month.type": "integer"
                    "projection.month.range": "01,12"
                    "projection.month.digits": "2"
                    "projection.day.type": "integer"
                    "projection.day.range": "01,31"
                    "projection.day.digits": "2"
                    "projection.enabled": "true"
                    transient_lastDdlTime: "1628198798"
                StorageDescriptor:
                    Columns:
                      -
                        Name: "messageversion"
                        Type: "string"
                      -
                        Name: "bot"
                        Type: "struct<name:string,version:string,id:string,aliasname:string,aliasid:string,localeid:string>"
                      -
                        Name: "messages"
                        Type: "array<struct<contenttype:string,content:string,imageresponsecard:struct<title:string,subtitle:string,imageurl:string,buttonslist:array<struct<text:string,value:string>>>>>"
                      -
                        Name: "sessionstate"
                        Type: "struct<dialogaction:struct<type:string,slottoelicit:string>,intent:struct<name:string,slots:array<struct<key:string,value:struct<shape:string,interpretedvalue:string,originalvalue:string,resolvedvalues:array<string>>>>,kendraresponse:string,state:string,confirmationstate:string>,originatingrequestid:string,sessionattributes:array<struct<key:string,value:string>>>"
                      -
                        Name: "sessionid"
                        Type: "string"
                      -
                        Name: "inputtranscript"
                        Type: "string"
                      -
                        Name: "missedutterance"
                        Type: "string"
                      -
                        Name: "requestid"
                        Type: "string"
                      -
                        Name: "timestamp"
                        Type: "string"
                      -
                        Name: "developeroverride"
                        Type: "string"
                      -
                        Name: "inputmode"
                        Type: "string"
                      -
                        Name: "requestattributes"
                        Type: "array<struct<key:string,value:string>>"
                      -
                        Name: "audioproperties"
                        Type: "struct<contenttype:string,s3path:string,duration:struct<total:string,voice:string,silence:string>>"
                      -
                        Name: "bargein"
                        Type: "string"
                      -
                        Name: "responsereason"
                        Type: "string"
                      -
                        Name: "operationname"
                        Type: "string"
                      -
                        Name: "nluconfidence"
                        Type: "string"
                      -
                        Name: "sentimentpositive"
                        Type: "double"
                      -
                        Name: "sentimentnegative"
                        Type: "double"
                      -
                        Name: "sentimentneutral"
                        Type: "double"
                      -
                        Name: "sentimentmixed"
                        Type: "double"
                      -
                        Name: "sentiment"
                        Type: "string"
                    Location: !Sub "s3://${ProjectName}/lex-base/${BotName}" 
                    InputFormat: "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat"
                    OutputFormat: "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat"
                    Compressed: false
                    NumberOfBuckets: -1
                    SerdeInfo:
                        SerializationLibrary: "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe"
                        Parameters:
                            "serialization.format": "1"
                    Parameters: {}
                    SkewedInfo:
                        SkewedColumnValueLocationMaps: {}
                    StoredAsSubDirectories: false
                PartitionKeys:
                  -
                    Name: "year"
                    Type: "int"
                  -
                    Name: "month"
                    Type: "int"
                  -
                    Name: "day"
                    Type: "int"
                Retention: 0
                Name: !GetAtt StringTransformer.result



    StringTransformer:
      Type: 'AWS::CloudFormation::CustomResource'
      Version: '1.0'
      Properties:
        ServiceToken: !GetAtt TransformStringFunction.Arn 
        InputString: !Sub "connect_lex_${BotName}"
        Operation: "Lower"

    TransformStringExecutionBasicPolicy:
        Type: "AWS::IAM::ManagedPolicy"
        DependsOn: DbAndBucketWaitCondition
        Properties:
            ManagedPolicyName: !Sub "${ProjectName}-${BotName}-TransformString"
            Path: "/"
            PolicyDocument: !Sub |
                {
                    "Version": "2012-10-17",
                    "Statement": [
                        {
                            "Action": [
                                "logs:CreateLogStream",
                                "logs:CreateLogGroup",
                                "logs:PutLogEvents"
                            ],
                            "Resource": "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/*",
                            "Effect": "Allow",
                            "Sid": "WriteCloudWatchLogs"
                        }
                    ]
                }
    TransformStringExecutionRole:
        Type: "AWS::IAM::Role"
        Properties:
            Path: "/service-role/"
            RoleName: !Sub "${ProjectName}-${BotName}-TransformString"
            AssumeRolePolicyDocument: !Sub |
                {
                    "Version":"2012-10-17",
                    "Statement":[
                        {
                            "Effect":"Allow",
                            "Principal":{
                                "Service":"lambda.amazonaws.com"
                                },
                            "Action":"sts:AssumeRole"
                        }
                    ]
                }
            MaxSessionDuration: 3600
            ManagedPolicyArns:
              - !Ref TransformStringExecutionBasicPolicy

    TransformStringFunction:
      Type: AWS::Lambda::Function
      Properties:
        Description: "CloudFormation uses this function for string transformation logic."
        FunctionName: !Sub "${ProjectName}-${BotName}-TransformString"
        Handler: "index.lambda_handler"      
        Code:
          ZipFile: |
            import json
            import logging
            from urllib.request import urlopen, Request
            LOG = logging.getLogger()
            LOG.setLevel(logging.INFO)

            import traceback
            def send_response(event, context, response_status, response_data):
                response_body = {'Status': response_status,
                                'Reason': 'See the details in CloudWatch Log Stream: ' + context.log_stream_name,
                                'PhysicalResourceId': context.log_stream_name,
                                'StackId': event['StackId'],
                                'RequestId': event['RequestId'],
                                'LogicalResourceId': event['LogicalResourceId'],
                                'Data': response_data}
                LOG.info(f'RESPONSE BODY: {json.dumps(response_body)}')
                try:
                    data = json.dumps(response_body)
                    req = Request(url = event['ResponseURL'], data = bytes(data.encode("utf-8")), method = "PUT")
                    # Add the appropriate header.
                    req.add_header("Content-type", "application/json; charset=UTF-8")
                    response = urlopen(req)
                    LOG.info("Response:")
                    LOG.info(response)
                    return
                except Exception as e:
                    traceback.print_exc()
                    LOG.error('Exception while sending back the request to CFN', e)


            def handler(event, context):
                LOG.info(f'REQUEST RECEIVED: {event}')
                
                # For Delete requests, immediately send a SUCCESS response.
                if event['RequestType'] == 'Delete':
                    LOG.info('Entering Delete')
                    send_response(event, context, "SUCCESS", {})
                    return

                response_data = {}
                response_status = "SUCCESS"
                eventParams = event["ResourceProperties"]
                try:
                    operation = eventParams["Operation"]
                    inputString = eventParams["InputString"]
                    no_param_string_funcs = ["Upper", "Lower", "Capitalize", "Title", "SwapCase"]
                    if operation in no_param_string_funcs:
                        response_data["result"]  = getattr(inputString, operation.lower())()
                    elif operation == "Strip":
                        chars = None
                        if "Chars" in eventParams:
                            chars = eventParams["Chars"]
                        response_data["result"] = inputString.strip(chars)
                    elif operation == "Replace":
                        old = eventParams["Old"]
                        new = eventParams["New"]
                        response_data["result"]  = inputString.replace(old, new)
                    elif operation == "MaxLength":
                        length = int(eventParams["Length"])
                        if len(inputString) <= length:
                            response_data["result"]  = inputString
                        elif "StripFrom" in eventParams:
                            if eventParams["StripFrom"] == "Left":
                                response_data["result"]  = inputString[len(inputString)-length:]
                            elif eventParams["StripFrom"] != "Right":
                                response_status = "failure"
                        else:
                            response_data["result"]  = inputString[:length]
                    else:
                        response_status = "failure"
                except Exception as e:
                    traceback.print_exc()
                    response_status = "failure"
                    response_data["message"] = str(e)
                send_response(event, context, response_status, response_data)
        Handler: index.handler
        Runtime: python3.9
        Role: !GetAtt TransformStringExecutionRole.Arn


    AthenaNamedQueryForLexWithCtrAverageSentimentAndCtrDataPerContact:
        Type: "AWS::Athena::NamedQuery"
        DependsOn: "DbAndBucketWaitCondition"
        Properties:
            Name: !Sub "${ProjectName} ${BotName} AverageSentimentAndCtrDataPerContact"
            Description: "This query analyzes the lex records in combination with the CTR records to return CTR data and average sentiment per contact"
            Database: !Sub "${ProjectName}"
            QueryString: !Sub |
                CREATE OR REPLACE VIEW ${ProjectName}_${BotName}_Lex_with_Ctr_AverageSentimentAndCtrDataPerContact AS
                WITH
                  lex_ctr_map_data AS (
                   SELECT DISTINCT sessionid, s_att.value as "ContactId"
                   FROM
                     "${ProjectName}"."${StringTransformer.result}", UNNEST(sessionstate.sessionattributes) t (s_att)
                   WHERE s_att.key = 'ContactId'
                ), 
                  lex_sentiment_data AS (
                   SELECT sessionid, AVG(sentimentpositive) sentimentpositive, AVG(sentimentnegative) sentimentnegative, AVG(sentimentneutral) sentimentneutral, AVG(sentimentmixed) sentimentmixed
                   FROM
                     "${ProjectName}"."${StringTransformer.result}"
                     GROUP BY sessionid
                ) 
                SELECT
                  ctr_data.contactid
                , ctr_data.initiationtimestamp
                , ctr_data.customerendpoint.address
                , ctr_data.customerendpoint.type
                , ctr_data.agent.username agent_username
                , ctr_data.queue.name queue_name
                , ctr_data.disconnectreason
                , lex_ctr_map_data.sessionid
                , lex_sentiment_data.sentimentpositive
                , lex_sentiment_data.sentimentnegative
                , lex_sentiment_data.sentimentneutral
                , lex_sentiment_data.sentimentmixed
                FROM
                  "${ProjectName}"."connect_ctr" as ctr_data
                  JOIN lex_ctr_map_data
                  ON ctr_data.ContactId = lex_ctr_map_data.ContactId
                  JOIN lex_sentiment_data
                  ON lex_ctr_map_data.sessionid = lex_sentiment_data.sessionid


Outputs:
    LexWithCtrAverageSentimentAndCtrDataPerContact:
        Description: This query analyzes the lex records in combination with the CTR records to return CTR data and average sentiment per contact
        Value: !Join
        - ''
        - - !Sub 'https://console.aws.amazon.com/athena/home?force&region=${AWS::Region}'
          - '#query/saved/'
          - !Ref AthenaNamedQueryForLexWithCtrAverageSentimentAndCtrDataPerContact




 