AWSTemplateFormatVersion: '2010-09-09'

Parameters:
  DataQualityDatabaseSuffix:
    Description: Database name suffix for NYC Taxi Dataset
    Type: String
    Default: db_nyc_taxi_data
  DataQualityS3BucketNameSuffix:
    Description: S3 Bucket Names suffix to store the input results
    Type: String
    Default: nyc-taxi-datasets
  DataQualityS3BucketNameSuffixOutputs:
    Description: S3 Bucket Names suffix to store the output results
    Type: String
    Default: nyc-taxi-datasets-outputs
  DataQualityTableSuffix:
    Description: Table name suffix for NYC Taxi Dataset
    Type: String
    Default: tbl_nxc_taxi_data
  SNSTopicNameAlertNotification:
    Type: String
    Description:  SNS Standard Glue Data Quality Blog Alert Notification
    Default: SNSStandardGlueDataQualityBlogAlertNotification
  EmailAddressAlertNotification:
    Type: String
    Default: abc@amazon.com

Resources:

  GlueDataQualityBlogRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Join ['-', ['GlueDataQualityLaunchBlogDemoRole', !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Effect: Allow
          Principal:
            Service:
            - glue.amazonaws.com
            - lambda.amazonaws.com
          Action:
          - sts:AssumeRole
      Path: "/"
      ManagedPolicyArns:
      - arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole
      - arn:aws:iam::aws:policy/AWSLambdaExecute
      Policies:
      - PolicyName: !Join ['-', ['GlueDataQualityLaunchBlogSnsGluePolicy', !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]
        PolicyDocument:
          Version: 2012-10-17
          Statement:
            - Sid: VisualEditor0
              Effect: Allow
              Action:
                - glue:StartDataQualityRuleRecommendationRun
                - glue:ListDataQualityRuleRecommendationRuns
                - glue:SearchTables
                - glue:PublishDataQuality
                - glue:GetDataQualityResult
                - glue:GetTables
                - sns:GetTopicAttributes
                - glue:GetTableVersions
                - glue:GetDataQualityRuleRecommendationRun
                - glue:GetDataQualityRuleset
                - glue:GetDatabases
                - glue:UpdateDataQualityRuleset
                - glue:GetTable
                - glue:GetDatabase
                - glue:GetPartition
                - sns:Publish
                - glue:CancelDataQualityRuleRecommendationRun
                - glue:ListDataQualityRulesetEvaluationRuns
                - glue:CancelDataQualityRulesetEvaluationRun
                - glue:ListDataQualityRulesets
                - sns:Subscribe
                - glue:GetDataQualityRulesetEvaluationRun
                - glue:ListDataQualityResults
                - glue:StartDataQualityRulesetEvaluationRun
              Resource:
                - arn:aws:glue:*:*:table/*db_nyc_taxi_data*/*tbl_nxc_taxi_data*
                - arn:aws:glue:*:*:dataQualityRuleset/*
                - arn:aws:glue:*:*:database/*db_nyc_taxi_data*
                - arn:aws:glue:*:*:catalog
                - arn:aws:sns:*:*:*SNS*
            - Sid: VisualEditor1
              Effect: Allow
              Action:
                - sns:GetSubscriptionAttributes
                - glue:CreateDataQualityRuleset
                - glue:ListCrawlers
                - glue:ListCrawls
                - glue:GetClassifier
              Resource: "*"
      - PolicyName: !Join ['-', ['GlueDataQualityLaunchBlogS3IAMPolicy', !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]
        PolicyDocument:
          Version: 2012-10-17
          Statement:
            - Sid: VisualEditor0
              Effect: Allow
              Action:
                - s3:PutObject
                - s3:GetObject
                - iam:PassRole
                - s3:ListBucket
              Resource:
                - arn:aws:s3:::*nyc-taxi-datasets*
                - arn:aws:iam::*:role/*
            - Sid: VisualEditor1
              Effect: Allow
              Action: s3:ListAllMyBuckets
              Resource: "*"
      - PolicyName: !Join ['-', ['GlueDataQualityLaunchBlogLambdaPolicy', !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]
        PolicyDocument:
          Version: 2012-10-17
          Statement:
            - Sid: VisualEditor0
              Effect: Allow
              Action:
                - lambda:CreateFunction
                - lambda:UpdateFunctionEventInvokeConfig
                - lambda:InvokeFunction
                - lambda:ListVersionsByFunction
                - lambda:GetFunction
                - lambda:PublishLayerVersion
                - lambda:ListFunctionUrlConfigs
                - lambda:InvokeAsync
                - lambda:GetFunctionConfiguration
                - lambda:UpdateFunctionCode
                - lambda:ListFunctionEventInvokeConfigs
                - lambda:ListProvisionedConcurrencyConfigs
                - lambda:PutFunctionEventInvokeConfig
                - lambda:PublishVersion
                - lambda:GetPolicy
              Resource: arn:aws:lambda:*:*:function:*GlueDataQualityBlogAlertLambda*
            - Sid: VisualEditor1
              Effect: Allow
              Action: lambda:ListFunctions
              Resource: "*"

  
  GlueDataQualityDatabase:
    Type: AWS::Glue::Database
    Properties:
      CatalogId:
        Ref: AWS::AccountId
      DatabaseInput:
        Name: !Join ['-', [ !Ref DataQualityDatabaseSuffix , !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]

  GlueDataQualityTable:
    DependsOn: GlueDataQualityDatabase
    Type: AWS::Glue::Table
    Properties:
      CatalogId:
        Ref: AWS::AccountId
      DatabaseName: !Join ['-', [ !Ref DataQualityDatabaseSuffix , !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]
      TableInput:
        Name: !Join ['-', [ !Ref DataQualityTableSuffix , !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]
        TableType: EXTERNAL_TABLE
        Parameters:
          classification: parquet
          compressionType: none
          typeOfData: file
        StorageDescriptor:
          OutputFormat: org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat
          InputFormat: org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat
          Location:
            Fn::Sub: s3://${GlueDataQualityS3Bucket}/
          SerdeInfo:
            SerializationLibrary: org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe
          Columns:
          - Name: vendorid
            Type: bigint
          - Name: tpep_pickup_datetime
            Type: timestamp
          - Name: tpep_dropoff_datetime
            Type: timestamp
          - Name: passenger_count
            Type: double
          - Name: trip_distance
            Type: double
          - Name: ratecodeid
            Type: double
          - Name: store_and_fwd_flag
            Type: string
          - Name: pulocationid
            Type: bigint
          - Name: dolocationid
            Type: bigint
          - Name: payment_type
            Type: bigint
          - Name: fare_amount
            Type: double
          - Name: extra
            Type: double
          - Name: mta_tax
            Type: double
          - Name: tip_amount
            Type: double
          - Name: tolls_amount
            Type: double
          - Name: improvement_surcharge
            Type: double
          - Name: total_amount
            Type: double
          - Name: congestion_surcharge
            Type: double
          - Name: airport_fee
            Type: double
        
  GlueDataQualityS3Bucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Join ['-', [ !Ref DataQualityS3BucketNameSuffix , !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]

  GlueDataQualityResultsS3Bucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName:  !Join ['-', [ !Ref DataQualityS3BucketNameSuffixOutputs , !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]

  ResourceSNSTopicNameAlertNotification: 
   Type: AWS::SNS::Topic
   Properties: 
     Subscription: 
       - Endpoint: !Ref EmailAddressAlertNotification
         Protocol: "email"
     TopicName: !Join ['-', [ !Ref SNSTopicNameAlertNotification , !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]
     
  GlueDataQualityCustomRulesetName:
    Type: AWS::Glue::DataQualityRuleset
    Properties:
      Name: !Join ['-', [ 'custom-ruleset' , !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]
      Description: !Join ['-', [ 'custom-ruleset' , !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]
      TargetTable:        
        DatabaseName: !Ref 'GlueDataQualityDatabase'
        TableName: !Ref 'GlueDataQualityTable'
      Ruleset: 'Rules = [
        RowCount > 100,
        IsComplete "vendorid",
        IsPrimaryKey "vendorid" ]'  

  GlueDataQualityBlogStepsFunctionRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Join ['-', [ 'GlueDataQualityStepsfunctionBlogRole' , !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Effect: Allow
          Principal:
            Service:
            - states.amazonaws.com
            - glue.amazonaws.com
          Action:
          - sts:AssumeRole
      Path: "/"
      ManagedPolicyArns:
       - arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole
      Policies:
      - PolicyName: !Join ['-', [ 'GlueDataQualityStepsfunctionBlogPolicy' , !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]
        PolicyDocument:
          Version: 2012-10-17
          Statement:
            - Sid: VisualEditor0
              Effect: Allow
              Action:
                - glue:SearchTables
                - glue:StartDataQualityRuleRecommendationRun
                - glue:GetTableVersions
                - glue:UpdateTable
                - glue:GetDataQualityRuleRecommendationRun
                - s3:ListBucket
                - glue:UpdateDataQualityRuleset
                - iam:PassRole
                - glue:GetTableVersion
                - glue:CancelDataQualityRuleRecommendationRun
                - glue:UpdateColumnStatisticsForTable
                - glue:GetColumnStatisticsForTable
                - glue:ListDataQualityRulesets
                - glue:ListDataQualityRuleRecommendationRuns
                - glue:PublishDataQuality
                - glue:UpdateDatabase
                - glue:GetDataQualityResult
                - glue:CreateTable
                - glue:GetTables
                - glue:DeleteDataQualityRuleset
                - glue:GetDataQualityRuleset
                - glue:GetDatabases
                - glue:GetTable
                - glue:GetDatabase
                - s3:GetObject
                - s3:PutObject
                - glue:ListDataQualityRulesetEvaluationRuns
                - glue:CancelDataQualityRulesetEvaluationRun
                - glue:GetDataQualityRulesetEvaluationRun
                - glue:ListDataQualityResults
                - glue:StartDataQualityRulesetEvaluationRun
              Resource:
                - arn:aws:iam::*:role/*
                - arn:aws:s3:::*nyc-taxi-datasets*
                - arn:aws:glue:*:*:table/*db_nyc_taxi_data*/*tbl_nxc_taxi_data*
                - arn:aws:glue:*:*:dataQualityRuleset/*
                - arn:aws:glue:*:*:database/*db_nyc_taxi_data*
                - arn:aws:glue:*:*:catalog
            - Sid: VisualEditor1
              Effect: Allow
              Action:
                - s3:ListAllMyBuckets
                - glue:CreateDataQualityRuleset
              Resource: "*"

  StateMachineGlueDataQualityRecommenedBlog:
    Type: 'AWS::StepFunctions::StateMachine'
    Properties:
      DefinitionString: !Sub |
        {
          "Comment": "A description of my state machine ",
          "StartAt": "Ruleset Exists?",
          "States": {
            "Ruleset Exists?": {
              "Type": "Task",
              "Next": "Start DataQualityRuleEvaluation",
              "Parameters": {
                "Name.$": "$$.Execution.Input.ruleset_name"
              },
              "Resource": "arn:aws:states:::aws-sdk:glue:getDataQualityRuleset",
              "Catch": [
                {
                  "ErrorEquals": [
                    "States.TaskFailed"
                  ],
                  "Next": "StartDataQualityRuleRecommendationRun",
                  "ResultPath": "$.error",
                  "Comment": "Glue data quality rule not found for any table"
                },
                {
                  "ErrorEquals": [
                    "States.Timeout"
                  ],
                  "Comment": "Unable to execute the command",
                  "Next": "JobFailed",
                  "ResultPath": "$.error"
                }
              ]
            },
            "StartDataQualityRuleRecommendationRun": {
              "Type": "Task",
              "Next": "WaitForRecommendationRunCompletion",
              "Parameters": {
                "DataSource": {
                  "GlueTable": {
                    "DatabaseName.$": "$$.Execution.Input.database_name",
                    "TableName.$": "$$.Execution.Input.table_name"
                  }
                },
                "Role": "arn:aws:iam::ACCOUNTID:role/GlueDataQualityStepsfunctionBlogRole",
                "CreatedRulesetName.$": "$$.Execution.Input.ruleset_name"
              },
              "Resource": "arn:aws:states:::aws-sdk:glue:startDataQualityRuleRecommendationRun"
            },
            "WaitForRecommendationRunCompletion": {
              "Type": "Wait",
              "Seconds": 30,
              "Next": "GetDataQualityRuleRecommendationRun"
            },
            "GetDataQualityRuleRecommendationRun": {
              "Type": "Task",
              "Next": "CheckCompletion",
              "Parameters": {
                "RunId.$": "$.RunId"
              },
              "Resource": "arn:aws:states:::aws-sdk:glue:getDataQualityRuleRecommendationRun"
            },
            "CheckCompletion": {
              "Type": "Choice",
              "Choices": [
                {
                  "Variable": "$.Status",
                  "StringMatches": "SUCCEEDED",
                  "Next": "Start DataQualityRuleEvaluation"
                },
                {
                  "Variable": "$.Status ",
                  "StringMatches": "RUNNING",
                  "Next": "WaitToChangeTheStatus"
                }
              ],
              "Default": "JobFailed"
            },
            "WaitToChangeTheStatus": {
              "Type": "Wait",
              "Seconds": 30,
              "Next": "GetDataQualityRuleRecommendationRun"
            },
            "JobFailed": {
              "Type": "Fail"
            },
            "Start DataQualityRuleEvaluation": {
              "Type": "Task",
              "Parameters": {
                "DataSource": {
                  "GlueTable": {
                    "DatabaseName.$": "$$.Execution.Input.database_name",
                    "TableName.$": "$$.Execution.Input.table_name"
                  }
                },
                "Role": "arn:aws:iam::ACCOUNTID:role/GlueDataQualityStepsfunctionBlogRole",
                "RulesetNames.$": "States.Array($$.Execution.Input.ruleset_name)",
                "AdditionalRunOptions": {
                  "ResultsS3Prefix.$": "$$.Execution.Input.dq_output_location"
                }
              },
              "Resource": "arn:aws:states:::aws-sdk:glue:startDataQualityRulesetEvaluationRun",
              "Next": "WaitForDataQualityRuleEvaluationCompletion",
              "Retry": [
                {
                  "ErrorEquals": [
                    "States.TaskFailed"
                  ],
                  "BackoffRate": 2,
                  "IntervalSeconds": 1,
                  "MaxAttempts": 2
                }
              ],
              "Catch": [
                {
                  "ErrorEquals": [
                    "States.TaskFailed"
                  ],
                  "Next": "JobFailed"
                }
              ]
            },
            "WaitForDataQualityRuleEvaluationCompletion": {
              "Type": "Wait",
              "Seconds": 30,
              "Next": "Get DQ Rule Evaluation State"
            },
            "Get DQ Rule Evaluation State": {
              "Type": "Task",
              "Next": "ChoiceForJobStatus",
              "Parameters": {
                "RunId.$": "$.RunId"
              },
              "Resource": "arn:aws:states:::aws-sdk:glue:getDataQualityRulesetEvaluationRun"
            },
            "ChoiceForJobStatus": {
              "Type": "Choice",
              "Choices": [
                {
                  "Or": [
                    {
                      "Variable": "$.Status",
                      "StringEquals": "FAILED"
                    },
                    {
                      "Variable": "$.Status",
                      "StringEquals": "SUCCEEDED"
                    }
                  ],
                  "Next": "JobSuccessed"
                }
              ],
              "Default": "Wait for Rule Evaluation Run"
            },
            "Wait for Rule Evaluation Run": {
              "Type": "Wait",
              "Seconds": 30,
              "Next": "Get DQ Rule Evaluation State"
            },
            "JobSuccessed": {
              "Type": "Succeed"
            }
          }
        }

      RoleArn: !GetAtt 'GlueDataQualityBlogStepsFunctionRole.Arn'

  StateMachineGlueDataQualityCustomBlog:
    Type: 'AWS::StepFunctions::StateMachine'
    Properties:
      DefinitionString: !Sub |
        {
          "Comment": "A description of my state machine ",
          "StartAt": "Ruleset Exists?",
          "States": {
            "Ruleset Exists?": {
              "Type": "Task",
              "Next": "Start DataQualityRuleEvaluation",
              "Parameters": {
                "Name.$": "$$.Execution.Input.ruleset_name"
              },
              "Resource": "arn:aws:states:::aws-sdk:glue:getDataQualityRuleset",
              "Catch": [
                {
                  "ErrorEquals": [
                    "States.TaskFailed"
                  ],
                  "Next": "CreateDataQualityRuleset",
                  "ResultPath": "$.error",
                  "Comment": "Glue data quality rule not found for any table"
                },
                {
                  "ErrorEquals": [
                    "States.Timeout"
                  ],
                  "Comment": "Unable to execute the command",
                  "Next": "JobFailed",
                  "ResultPath": "$.error"
                }
              ]
            },
            "CreateDataQualityRuleset": {
              "Type": "Task",
              "Next": "WaitForCustomRuleRunCompletion",
              "Parameters": {
                "Name.$": "$$.Execution.Input.ruleset_name",
                "Ruleset": "Rules = [IsComplete \"vendorid\", IsPrimaryKey \"vendorid\"]",
                "TargetTable": {
                  "DatabaseName.$": "$$.Execution.Input.database_name",
                  "TableName.$": "$$.Execution.Input.table_name"
                }
              },
              "Resource": "arn:aws:states:::aws-sdk:glue:createDataQualityRuleset"
            },
            "WaitForCustomRuleRunCompletion": {
              "Type": "Wait",
              "Seconds": 30,
              "Next": "GetDataQualityRuleset"
            },
            "GetDataQualityRuleset": {
              "Type": "Task",
              "Next": "CheckCompletion",
              "Parameters": {
                "Name.$": "$$.Execution.Input.ruleset_name"
              },
              "Resource": "arn:aws:states:::aws-sdk:glue:getDataQualityRuleset"
            },
            "CheckCompletion": {
              "Type": "Choice",
              "Choices": [
                {
                  "Variable": "$.LastModifiedOn",
                  "IsPresent": true,
                  "Next": "Start DataQualityRuleEvaluation"
                }
              ],
              "Default": "JobFailed"
            },
            "JobFailed": {
              "Type": "Fail"
            },
            "Start DataQualityRuleEvaluation": {
              "Type": "Task",
              "Parameters": {
                "DataSource": {
                  "GlueTable": {
                    "DatabaseName.$": "$$.Execution.Input.database_name",
                    "TableName.$": "$$.Execution.Input.table_name"
                  }
                },
                "Role": "arn:aws:iam::ACCOUNTID:role/GlueDataQualityStepsfunctionBlogRole",
                "RulesetNames.$": "States.Array($$.Execution.Input.ruleset_name)",
                "AdditionalRunOptions": {
                  "ResultsS3Prefix.$": "$$.Execution.Input.dq_output_location"
                }
              },
              "Resource": "arn:aws:states:::aws-sdk:glue:startDataQualityRulesetEvaluationRun",
              "Next": "WaitForDataQualityRuleEvaluationCompletion",
              "Retry": [
                {
                  "ErrorEquals": [
                    "States.TaskFailed"
                  ],
                  "BackoffRate": 2,
                  "IntervalSeconds": 1,
                  "MaxAttempts": 2
                }
              ],
              "Catch": [
                {
                  "ErrorEquals": [
                    "States.TaskFailed"
                  ],
                  "Next": "JobFailed"
                }
              ]
            },
            "WaitForDataQualityRuleEvaluationCompletion": {
              "Type": "Wait",
              "Seconds": 30,
              "Next": "Get DQ Rule Evaluation State"
            },
            "Get DQ Rule Evaluation State": {
              "Type": "Task",
              "Next": "ChoiceForJobStatus",
              "Parameters": {
                "RunId.$": "$.RunId"
              },
              "Resource": "arn:aws:states:::aws-sdk:glue:getDataQualityRulesetEvaluationRun"
            },
            "ChoiceForJobStatus": {
              "Type": "Choice",
              "Choices": [
                {
                  "Or": [
                    {
                      "Variable": "$.Status",
                      "StringEquals": "FAILED"
                    },
                    {
                      "Variable": "$.Status",
                      "StringEquals": "SUCCEEDED"
                    }
                  ],
                  "Next": "JobSuccessed"
                }
              ],
              "Default": "Wait for Rule Evaluation Run"
            },
            "Wait for Rule Evaluation Run": {
              "Type": "Wait",
              "Seconds": 30,
              "Next": "Get DQ Rule Evaluation State"
            },
            "JobSuccessed": {
              "Type": "Succeed"
            }
          }
        }

      RoleArn: !GetAtt 'GlueDataQualityBlogStepsFunctionRole.Arn'
  
  GlueDataQualityBlogAlertLambdaFunction:
    Type: 'AWS::Lambda::Function'
    Properties:
      FunctionName: !Join ['-', [ 'GlueDataQualityBlogAlertLambda' , !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]

      Code:
        ZipFile: |
          import boto3
          import json
          from datetime import datetime

          sns_client = boto3.client('sns')
          s3_client = boto3.client('s3')
          glue_client = boto3.client('glue')

          sns_topic_arn = 'sns_topic_arn'
          s3_bucket = 's3bucket'

          def write_logs(log_metadata):
              try:
                  filename = datetime.now().strftime("%m%d%Y%H%M%S") + ".json"
                  key_opts = {
                      'year': datetime.now().year,
                      'month': "{:02d}".format(datetime.now().month),
                      'day': "{:02d}".format(datetime.now().day),
                      'filename': filename
                  }
                  s3key = "gluedataqualitylogs/year={year}/month={month}/day={day}/{filename}".format(**key_opts)
                  s3_client.put_object(Bucket=s3_bucket, Key=s3key, Body=json.dumps(log_metadata))
              except Exception as e:
                  print(f'Error writing logs to S3: {e}')

          def get_catalog_data_quality_url(database_name, table_name):
              console_url = f"https://console.aws.amazon.com/glue/home?region={boto3.Session().region_name}#/v2/data-catalog/tables/data-quality-runs/{table_name}?database={database_name}"
              return console_url
    
          def get_glue_job_url(job_name):
              console_url = f"https://console.aws.amazon.com/gluestudio/home?region={boto3.Session().region_name}#/editor/job/{job_name}"
              return console_url
            

          def lambda_handler(event, context):
              log_metadata = {}
              message_text = ""
              subject_text = ""

              if event['detail']['context']['contextType'] == 'GLUE_DATA_CATALOG':
                  log_metadata['ruleset_name'] = str(event['detail']['rulesetNames'][0])
                  log_metadata['databaseName'] = str(event['detail']['context']['databaseName'])
                  log_metadata['tableName'] = str(event['detail']['context']['tableName'])
                  log_metadata['runId'] = str(event['detail']['context']['runId'])
                  log_metadata['resultId'] = str(event['detail']['resultId'])
                  log_metadata['state'] = str(event['detail']['state'])
                  log_metadata['score'] = str(event['detail']['score'])
                  log_metadata['numRulesSucceeded'] = str(event['detail']['numRulesSucceeded'])
                  log_metadata['numRulesFailed'] = str(event['detail']['numRulesFailed'])
                  log_metadata['numRulesSkipped'] = str(event['detail']['numRulesSkipped'])

                  message_text += "Glue Data Quality run details:\n"
                  message_text += "ruleset_name: {}\n".format(log_metadata['ruleset_name'])
                  message_text += "glue_database_name: {}\n".format(log_metadata['databaseName'])
                  message_text += "glue_table_name: {}\n".format(log_metadata['tableName'])
                  message_text += "run_id: {}\n".format(log_metadata['runId'])
                  message_text += "result_id: {}\n".format(log_metadata['resultId'])
                  message_text += "state: {}\n".format(log_metadata['state'])
                  message_text += "score: {}\n".format(log_metadata['score'])
                  message_text += "numRulesSucceeded: {}\n".format(log_metadata['numRulesSucceeded'])
                  message_text += "numRulesFailed: {}\n".format(log_metadata['numRulesFailed'])
                  message_text += "numRulesSkipped: {}\n".format(log_metadata['numRulesSkipped'])

                  subject_text = "Glue Data Quality ruleset {} run details".format(log_metadata['ruleset_name'])

                  # Get catalog data quality URL
                  database_name = log_metadata['databaseName']
                  table_name = log_metadata['tableName']
                  catalog_data_quality_url = get_catalog_data_quality_url(database_name, table_name)
                  message_text += f"\n\nCatalog Data Quality URL: {catalog_data_quality_url}"

              else:
                  log_metadata['ruleset_name'] = str(event['detail']['rulesetNames'][0])
                  log_metadata['jobName'] = str(event['detail']['context']['jobName'])
                  log_metadata['jobId'] = str(event['detail']['context']['jobId'])
                  log_metadata['resultId'] = str(event['detail']['resultId'])
                  log_metadata['state'] = str(event['detail']['state'])
                  log_metadata['score'] = str(event['detail']['score'])
                  log_metadata['numRulesSucceeded'] = str(event['detail']['numRulesSucceeded'])
                  log_metadata['numRulesFailed'] = str(event['detail']['numRulesFailed'])
                  log_metadata['numRulesSkipped'] = str(event['detail']['numRulesSkipped'])

                  message_text += "Glue Data Quality run details:\n"
                  message_text += "ruleset_name: {}\n".format(log_metadata['ruleset_name'])
                  message_text += "glue_job_name: {}\n".format(log_metadata['jobName'])
                  message_text += "job_run_id: {}\n".format(log_metadata['jobId'])
                  message_text += "result_id: {}\n".format(log_metadata['resultId'])
                  message_text += "state: {}\n".format(log_metadata['state'])
                  message_text += "score: {}\n".format(log_metadata['score'])
                  message_text += "numRulesSucceeded: {}\n".format(log_metadata['numRulesSucceeded'])
                  message_text += "numRulesFailed: {}\n".format(log_metadata['numRulesFailed'])
                  message_text += "numRulesSkipped: {}\n".format(log_metadata['numRulesSkipped'])

                  subject_text = "Glue Data Quality ruleset {} run details".format(log_metadata['ruleset_name'])

                  # Get Glue job URL
                  job_name = log_metadata['jobName']
                  job_url = get_glue_job_url(job_name)
                  message_text += f"\n\nGlue job URL: {job_url}"
                  
              resultID = str(event['detail']['resultId'])
              response = glue_client.get_data_quality_result(ResultId=resultID)
              RuleResults = response['RuleResults']
              message_text += "\n\nruleset details evaluation steps results:\n\n"
              subresult_info = []

              for dic in RuleResults:
                  subresult = "Name: {}\t\tResult: {}\t\tDescription: \t{}".format(dic['Name'], dic['Result'], dic['Description'])
                  if 'EvaluationMessage' in dic:
                      subresult += "\t\tEvaluationMessage: {}".format(dic['EvaluationMessage'])
                  subresult_info.append({
                      'Name': dic['Name'],
                      'Result': dic['Result'],
                      'Description': dic['Description'],
                      'EvaluationMessage': dic.get('EvaluationMessage', '')
                  })
                  message_text += "\n" + subresult

              log_metadata['resultrun'] = subresult_info

              write_logs(log_metadata)

              sns_client.publish(
                  TopicArn=sns_topic_arn,
                  Message=message_text,
                  Subject=subject_text
              )

              return {
                  'statusCode': 200,
                  'body': json.dumps('Message published to SNS topic')
              }

      Handler: index.lambda_handler
      Runtime: python3.10
      Timeout: 180
      Role: !GetAtt GlueDataQualityBlogRole.Arn

  GlueDataQualityBlogEventBridge:
    Type: 'AWS::Events::Rule'
    Properties:
      Description: Glue Data Quality Blog Event Bridge
      EventBusName: default
      EventPattern: 
        source:
          - aws.glue-dataquality
        detail-type:
          - Data Quality Evaluation Results Available
      Name: !Join ['-', [ 'GlueDataQualityBlogEventBridge' , !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]

      State: ENABLED
      Targets:
        - Arn: !GetAtt 
          - GlueDataQualityBlogAlertLambdaFunction
          - Arn
          Id: !Ref GlueDataQualityBlogAlertLambdaFunction

  LambdaInvokePermission:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName:
        Fn::GetAtt:
          - GlueDataQualityBlogAlertLambdaFunction
          - Arn
      Action: "lambda:InvokeFunction"
      Principal: "events.amazonaws.com"
      SourceArn:
        Fn::GetAtt:
          - GlueDataQualityBlogEventBridge
          - "Arn"

   
Outputs:
  DataQualityDatabase:
    Description: Glue database name for NYC Taxi Dataset
    Value: !Ref 'GlueDataQualityDatabase'
  DataQualityTable:
    Description: Glue table name for NYC Taxi Dataset
    Value: !Ref 'GlueDataQualityTable'
  DataQualityS3BucketNameStaging:
    Description: S3 bucket names to store the inputs dataset
    Value: !Ref 'GlueDataQualityS3Bucket'
  DataQualityS3BucketNameOutputs:
    Description: S3 bucket names to store the results/outputs dataset
    Value: !Ref 'GlueDataQualityResultsS3Bucket'
  SNSTopicNameAlertNotification: 
    Description: SNS standard Glue Data Quality alert and notification
    Value: !Ref ResourceSNSTopicNameAlertNotification
  GlueDataQualityCustomRulesetName: 
    Description: Glue Data Quality custom ruleset name
    Value: !Ref GlueDataQualityCustomRulesetName
  StateMachineGlueDataQualityRecommenedBlog: 
    Description: State machine Glue Data Quality recommened ruleset
    Value: !Ref StateMachineGlueDataQualityRecommenedBlog
  StateMachineGlueDataQualityCustomBlog: 
    Description: State machine Glue Data Quality custom ruleset
    Value: !Ref StateMachineGlueDataQualityCustomBlog
  GlueDataQualityBlogEventBridge:
    Description: GlueDataQualityBlogEventBridge
    Value: !Ref 'GlueDataQualityBlogEventBridge'
  GlueDataQualityBlogAlertLambdaFunction:
    Description: GlueDataQualityBlogAlertLambdaFunction
    Value: !Ref 'GlueDataQualityBlogAlertLambdaFunction'
