AWSTemplateFormatVersion: "2010-09-09"
Description: "This is the template for creating a solution to analyze Amazon Connect chat 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]*$"
    ConnectBucket:
        Type: "String"
        Default: "amazon-connect-123456"
        Description: Enter the name of the bucket that Amazon Connect instance is using ( typically named as 'amazon-connect-<instanceId>' )
    ConnectInstanceAlias:
        Type: "String"
        Default: "myconnectinstancealias"
        MinLength : 2
        Description: Enter the alias of the Amazon Connect instance 
    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

    EventBridgePolicyForChat:
        Type: "AWS::IAM::ManagedPolicy"
        DependsOn: DbAndBucketWaitCondition
        Properties:
            ManagedPolicyName: !Sub "${ProjectName}-EB-Chat"
            Path: "/service-role/"
            PolicyDocument: !Sub |
                {
                    "Version": "2012-10-17",
                    "Statement": [
                        {
                            "Effect": "Allow",
                            "Action": [
                                "firehose:PutRecord",
                                "firehose:PutRecordBatch"
                            ],
                            "Resource": [
                                "${KinesisFirehoseDeliveryStreamForChat.Arn}"
                            ]
                        }
                    ]
                }


    EventBridgeRoleForChat:
        Type: "AWS::IAM::Role"
        DependsOn: EventBridgePolicyForChat
        Properties:
            Path: "/service-role/"
            RoleName: !Sub "${ProjectName}-EB-Chat"
            AssumeRolePolicyDocument: "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Service\":\"events.amazonaws.com\"},\"Action\":\"sts:AssumeRole\"}]}"
            MaxSessionDuration: 3600
            ManagedPolicyArns:
              - !Ref EventBridgePolicyForChat


    EventsRuleForChat:
        Type: "AWS::Events::Rule"
        Properties:
            Name: !Sub "${ProjectName}_Chat"
            EventPattern: !Sub |
                {
                  "source": ["aws.s3"],
                  "detail-type": ["Object Created"],
                  "detail": {
                    "bucket": {
                      "name": ["${ConnectBucket}"]
                    },
                    "object": {
                      "key": [{ "prefix": "connect/${ConnectInstanceAlias}/ChatTranscripts/" }]
                    }
                  }
                }
            State: "ENABLED"
            Targets:
              -
                Arn: !GetAtt KinesisFirehoseDeliveryStreamForChat.Arn
                Id: "Id6e2fb565-24e4-48af-92ca-26a8ec7e15b0"
                InputTransformer:
                    InputPathsMap:
                        bucketname: "$.detail.bucket.name"
                        key: "$.detail.object.key"
                    InputTemplate: |
                        { "Records":[
                          {
                            "s3": {
                                 "object":{
                                       "key":"<key>"
                                 },
                                "bucket":{
                                       "name":"<bucketname>"
                                 }
                            }
                          }]
                        }
                RoleArn: !GetAtt EventBridgeRoleForChat.Arn
            EventBusName: "default"


    ReadConnectChatTranscriptPolicy:
        Type: "AWS::IAM::ManagedPolicy"
        DependsOn: DbAndBucketWaitCondition
        Properties:
            ManagedPolicyName: !Sub "${ProjectName}-S3Read-Chat"
            Path: "/"
            PolicyDocument: !Sub |
                {
                    "Version": "2012-10-17",
                    "Statement": [
                        {
                            "Sid": "VisualEditor0",
                            "Effect": "Allow",
                            "Action": "s3:GetObject",
                            "Resource": "arn:aws:s3:::${ConnectBucket}/connect/*"
                        }
                    ]
                }

    FirehoseTransformationLambdaForChatBasicPolicy:
        Type: "AWS::IAM::ManagedPolicy"
        DependsOn: DbAndBucketWaitCondition
        Properties:
            ManagedPolicyName: !Sub "${ProjectName}-Lambda-Chat"
            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"
                        }
                    ]
                }

    FirehoseTransformationLambdaRoleForChat:
        Type: "AWS::IAM::Role"
        Properties:
            Path: "/service-role/"
            RoleName: !Sub "${ProjectName}-Lambda-Chat"
            AssumeRolePolicyDocument: !Sub |
                {
                    "Version":"2012-10-17",
                    "Statement":[
                        {
                            "Effect":"Allow",
                            "Principal":{
                                "Service":"lambda.amazonaws.com"
                                },
                            "Action":"sts:AssumeRole"
                        }
                    ]
                }
            MaxSessionDuration: 3600
            ManagedPolicyArns:
              - !Ref FirehoseTransformationLambdaForChatBasicPolicy
              - !Ref ReadConnectChatTranscriptPolicy
              - "arn:aws:iam::aws:policy/ComprehendReadOnly"

    TransformationLambdaFunctionForChat:
        Type: "AWS::Lambda::Function"
        Properties:
            Description: "Kinesis firehose uses this processor for custom transformation logic."
            FunctionName: !Sub "${ProjectName}-TransformChat"
            Handler: "index.lambda_handler"
            Code:
                ZipFile: |
                    import base64
                    import json
                    import boto3
                    import urllib.parse

                    print('Loading function')
                    s3 = boto3.resource('s3') 
                    comprehend = boto3.client('comprehend')

                    class FileToBeIgnored(Exception):
                        """Raised when the input is to be ignored - e.g. we ignore files other than json """
                        def __init__(self, key, message=" will not be processed as is not json or it is in an ignored directory"):
                            self.key = key
                            self.message = message
                            super().__init__(self.message)
                        def __str__(self):
                            return f'{self.key} -> {self.message}'

                    class FileProcessingError(Exception):
                        """Raised when we try to process the file """
                        def __init__(self, key, message="Error during file processing"):
                            self.key = key
                            self.message = message
                            super().__init__(self.message)
                        def __str__(self):
                            return f'{self.key} -> {self.message}'

                    def transformRecord(recordToTransform):
                        ## This method receives the record to transform and returns the transformed record. 
                        chatLog = json.loads(recordToTransform)
                        #print("Chat log before transformation: " + json.dumps(chatLog, indent=2))
                        for chatTranscript in chatLog["Transcript"]:
                            if "Content" in chatTranscript and chatTranscript["ParticipantRole"] != "SYSTEM"  :
                                comprehendSentimentResponse = comprehend.detect_sentiment(Text=chatTranscript["Content"],LanguageCode='en')
                                #print (comprehendSentimentResponse)
                                chatTranscript["Sentiment"] = comprehendSentimentResponse["Sentiment"]
                                chatTranscript["SentimentScorePositive"] = comprehendSentimentResponse["SentimentScore"]["Positive"]
                                chatTranscript["SentimentScoreNegative"] = comprehendSentimentResponse["SentimentScore"]["Negative"]
                                chatTranscript["SentimentScoreNeutral"] = comprehendSentimentResponse["SentimentScore"]["Neutral"]
                                chatTranscript["SentimentScoreMixed"] = comprehendSentimentResponse["SentimentScore"]["Mixed"]
                        #print("Chat log after transformation: " + json.dumps(chatLog, indent=2))    
                        return json.dumps(chatLog)

                    def readRecordFromS3AndTransform(s3Event):
                        ## Get the location of the chat log file from S3 event 
                        key = urllib.parse.unquote_plus(s3Event['s3']['object']['key'], encoding='utf-8')
                        if (key[-4:] != 'json'):
                            raise FileToBeIgnored (key)
                        bucket = s3.Bucket(s3Event['s3']['bucket']['name'])
                        encodedFileContent = bucket.Object(key).get()['Body'].read()
                        ## Get the log content from the file in S3. 
                        try:
                            fileContent = encodedFileContent.decode()
                        except Exception as e: 
                            print (e)
                            raise FileProcessingError(key)
                        return transformRecord(fileContent)

                    def processKFHRecord(eachKFHRecord):
                        ## Each Kinesis Firehose record has the incoming record in 'data' field and an id that needs to be returned with transformed response.
                        recordId = eachKFHRecord['recordId']
                        eachKFHRecordData = json.loads(base64.b64decode(eachKFHRecord['data']))
                        #print("Record from firehose:" + json.dumps(eachKFHRecordData, indent=2))
                        if not ('Records' in eachKFHRecordData):
                            ## Drop the test events that are created at the creation of the S3 Event Notification - there is no data in these records
                            return {'result': 'Dropped','recordId': recordId}
                        else:
                            try:
                                ## For each valid record, get the data, transform the data and return it with 'Ok' result.  
                                transformedRecords = [readRecordFromS3AndTransform(eachS3Record) for eachS3Record in eachKFHRecordData['Records']]
                                encodedData = base64.b64encode(''.join(transformedRecords).encode("utf-8"))
                                return {'data': encodedData,'result': 'Ok','recordId': recordId}
                            except FileToBeIgnored as eFileToBeIgnored:
                                print (eFileToBeIgnored)
                            except FileProcessingError as eFileProcessingError:
                                print (eFileProcessingError)
                            except Exception as e:
                                print ('Unknown Exception')
                                print (e)
                            return {'result': 'Dropped','recordId': recordId}

                    def lambda_handler(event, context):
                        #print (json.dumps(event, indent=2))
                        output = [processKFHRecord(eachKFHRecord) for eachKFHRecord in event['records']]
                        print("{} record(s) processed successfully".format(len(output)))
                        return {'records': output}
            MemorySize: 128
            Role: !GetAtt FirehoseTransformationLambdaRoleForChat.Arn
            Runtime: "python3.9"
            Timeout: 60
            TracingConfig:
                Mode: "PassThrough"

    KinesisFirehoseServicePolicyForChat:
        Type: "AWS::IAM::ManagedPolicy"
        DependsOn: GlueTableForChat
        Properties:
            ManagedPolicyName: !Sub "${ProjectName}-KFH-Chat"
            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}/connect_chat"
                            ]
                        },
                        {
                            "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:${TransformationLambdaFunctionForChat}"
                        },
                        {
                            "Sid": "",
                            "Effect": "Allow",
                            "Action": [
                                "logs:PutLogEvents"
                            ],
                            "Resource": [
                                "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/kinesisfirehose/${ProjectName}-Chat:log-stream:*"
                            ]
                        }
                    ]
                }

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

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

    GlueTableForChat:
        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": "2021,2023"
                    "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: "awsaccountid"
                        Type: "string"
                      -
                        Name: "instanceid"
                        Type: "string"
                      -
                        Name: "initialcontactid"
                        Type: "string"
                      -
                        Name: "contactid"
                        Type: "string"
                      -
                        Name: "participants"
                        Type: "array<struct<participantid:string>>"
                      -
                        Name: "transcript"
                        Type: "array<struct<content:string,contenttype:string,id:string,displayname:string,type:string,participantrole:string,absolutetime:string,participantid:string,sentiment:string,sentimentscorepositive:float,sentimentscorenegative:float,sentimentscoreneutral:float,sentimentscoremixed:float>>"
                    Location: !Sub "s3://${ProjectName}/chat-base"
                    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: "connect_chat"


    ViewForChatPerDate:
       Type: "AWS::Glue::Table"
       DependsOn: "GlueTableForChat"
       Properties:
            DatabaseName: !Sub "${ProjectName}"
            CatalogId: !Ref "AWS::AccountId"
            TableInput: 
                ViewOriginalText: !Join
                  - ''
                  - - '/* Presto View: '
                    - Fn::Base64: 
                        Fn::Sub: |
                          {"originalSql":"SELECT\n  chat_data.year\n, chat_data.month\n, chat_data.day\n, \"avg\"(transcript_data.sentimentscorepositive) positive\n, \"avg\"(transcript_data.sentimentscorenegative) negative\n, \"avg\"(transcript_data.sentimentscoreneutral) neutral\n, \"avg\"(transcript_data.sentimentscoremixed) mixed\nFROM\n  \"connect_chat\" chat_data\n, UNNEST(transcript) t (transcript_data)\nWHERE ((transcript_data.content IS NOT NULL) AND (transcript_data.sentiment IS NOT NULL))\nGROUP BY year, month, day\nORDER BY year ASC, month ASC, day ASC\n","catalog":"awsdatacatalog","schema":"${ProjectName}","columns":[{"name":"year","type":"integer"},{"name":"month","type":"integer"},{"name":"day","type":"integer"},{"name":"positive","type":"real"},{"name":"negative","type":"real"},{"name":"neutral","type":"real"},{"name":"mixed","type":"real"}]}
                    - ' */'
                TableType: "VIRTUAL_VIEW"
                Parameters: 
                   comment: "Presto View"
                   presto_view: "true"
                ViewExpandedText: "/* Presto View */"
                StorageDescriptor:
                    Columns:
                      -
                        Name: "year"
                        Type: "int"
                      -
                        Name: "month"
                        Type: "int"
                      -
                        Name: "day"
                        Type: "int"
                      -
                        Name: "positive"
                        Type: "float"
                      -
                        Name: "negative"
                        Type: "float"
                      -
                        Name: "neutral"
                        Type: "float"
                      -
                        Name: "mixed"
                        Type: "float"
                    Location: ""
                    Compressed: false
                    NumberOfBuckets: 0
                    SerdeInfo: {}
                    StoredAsSubDirectories: false
                Retention: 0
                Name: "avg_chat_sentiment_per_date"

