diff --git a/devops-pipeline/other-resources/locals.tf b/devops-pipeline/other-resources/locals.tf index be68b3a..77203e7 100644 --- a/devops-pipeline/other-resources/locals.tf +++ b/devops-pipeline/other-resources/locals.tf @@ -27,12 +27,23 @@ locals { global_prefix = "${var.project_name}-${local.region_short_name}-${local.environment_short_name}-${var.client_name}" # Resource prefix - lambda_prefix = "${local.global_prefix}-lmb" - sqs_prefix = "${local.global_prefix}-sqs" - sns_prefix = "${local.global_prefix}-sns" - dynamodb_prefix = "${local.global_prefix}-dyd" s3_prefix = "${local.global_prefix}-s3" + iam_role_prefix = "${local.global_prefix}-rol" + iam_policy_prefix = "${local.global_prefix}-pol" - # Terraform locking table + # s3 bukect name devops_s3_bucket_name = "${local.s3_prefix}-${var.devops_s3_bucket.name}" + mwaa_resources_s3_bucket_name = "${local.s3_prefix}-${var.mwaa_resources_s3_bucket.name}" + raw_data_ingestion_s3_bucket_name = "${local.s3_prefix}-${var.raw_data_ingestion_s3_bucket.name}" + + # IAM role name + snowflake_integration_role_name = "${local.iam_role_prefix}-snowflake-integration-role" + cross_account_role_name = "${local.iam_role_prefix}-cross-account-role" + mwaa_exec_role_role_name = "${local.iam_role_prefix}-mwaa-exec-role" + + # IAM policy name + snowflake_integration_policy_name = "${local.iam_policy_prefix}-snowflake-integration-policy" + cross_account_policy_name = "${local.iam_policy_prefix}-cross-account-policy" + mwaa_exec_policy_name = "${local.iam_policy_prefix}-mwaa-exec-policy" + } \ No newline at end of file diff --git a/devops-pipeline/other-resources/main.tf b/devops-pipeline/other-resources/main.tf index 0b5463c..d9febef 100644 --- a/devops-pipeline/other-resources/main.tf +++ b/devops-pipeline/other-resources/main.tf @@ -6,19 +6,21 @@ terraform { } } backend "s3" { - # bucket = "doczyai-use2-d-infra-s3-terraform-state" # Parameterize using -backend-config flag with "terraform init" - # key = "terraform/devops-pipeline/devops.tfstate" # Parameterize - # region = "us-east-2" # Parameterize - # profile = "doczyai" # Parameterize - # dynamodb_table = "doczyai-use2-d-infra-dyd-terraform-lock" # Parameterize + bucket = "doczyai-use2-d-infra-s3-terraform-state" # Parameterize using -backend-config flag with "terraform init" + key = "terraform/devops-pipeline/devops.tfstate" # Parameterize + region = "us-east-2" # Parameterize + profile = "doczyai" # Parameterize + dynamodb_table = "doczyai-use2-d-infra-dyd-terraform-lock" # Parameterize encrypt = true } } provider "aws" { - access_key = var.access_key - secret_key = var.secret_key + # access_key = var.access_key + # secret_key = var.secret_key + + profile = var.aws_profile default_tags { tags = { @@ -27,6 +29,7 @@ provider "aws" { } } +# s3 bucket - devops resource "aws_s3_bucket" "devops" { bucket = local.devops_s3_bucket_name tags = local.common_tags @@ -37,6 +40,29 @@ resource "aws_s3_bucket_policy" "deny_insecure_communication" { policy = data.aws_iam_policy_document.deny_insecure_communication.json } +# s3 bucket - raw-data-ingestion +resource "aws_s3_bucket" "raw_data_ingestion" { + bucket = local.raw_data_ingestion_s3_bucket_name + tags = local.common_tags +} + +resource "aws_s3_bucket_policy" "raw_data_ingestion_policy" { + bucket = aws_s3_bucket.raw_data_ingestion.id + policy = data.aws_iam_policy_document.raw_data_ingestion_deny_insecure_communication.json +} + +# s3 bucket - mwaa_resources +resource "aws_s3_bucket" "mwaa_resources" { + bucket = local.mwaa_resources_s3_bucket_name + tags = local.common_tags +} + +resource "aws_s3_bucket_policy" "mwaa_resources_policy" { + bucket = aws_s3_bucket.mwaa_resources.id + policy = data.aws_iam_policy_document.mwaa_resources_deny_insecure_communication.json +} + +# Deny insecure s3 bucket policy data "aws_iam_policy_document" "deny_insecure_communication" { statement { sid = "DenyInsecureCommunications" @@ -62,3 +88,285 @@ data "aws_iam_policy_document" "deny_insecure_communication" { ] } } + +# Deny insecure s3 bucket policy - raw_data_ingestion +data "aws_iam_policy_document" "raw_data_ingestion_deny_insecure_communication" { + statement { + sid = "DenyInsecureCommunications" + + principals { + type = "*" + identifiers = ["*"] + } + + condition { + test = "Bool" + variable = "aws:SecureTransport" + values = ["false"] + } + + effect = "Deny" + + actions = ["s3:*"] + + resources = [ + aws_s3_bucket.raw_data_ingestion.arn, + "${aws_s3_bucket.raw_data_ingestion.arn}/*", + ] + } +} + +# Deny insecure s3 bucket policy - mwaa_resources +data "aws_iam_policy_document" "mwaa_resources_deny_insecure_communication" { + statement { + sid = "DenyInsecureCommunications" + + principals { + type = "*" + identifiers = ["*"] + } + + condition { + test = "Bool" + variable = "aws:SecureTransport" + values = ["false"] + } + + effect = "Deny" + + actions = ["s3:*"] + + resources = [ + aws_s3_bucket.mwaa_resources.arn, + "${aws_s3_bucket.mwaa_resources.arn}/*", + ] + } +} + +# IAM role - snowflake-integration-role +resource "aws_iam_role" "snowflake_integration_role" { + name = local.snowflake_integration_role_name + + assume_role_policy = jsonencode({ + Version = "2012-10-17" + Statement = [ + { + Action = "sts:AssumeRole" + Effect = "Allow" + Sid = "" + Principal = { + AWS = "arn:aws:iam::851725635820:user/3nmi0000-s" + } + Condition = { + StringEquals = { + "sts:ExternalId" = "OQ11564_SFCRole=2_lNKzpyFn/e/xQgvGGpusVbEZA0A=" + } + } + }, + ] + }) +} + +# Construct IAM Policy for snowflake-integration-role +data "aws_iam_policy_document" "snowflake_integration_policy_document" { + statement { + effect = "Allow" + actions = [ + "s3:PutObject", + "s3:GetObject", + "s3:GetObjectVersion", + "s3:DeleteObject", + "s3:DeleteObjectVersion" + ] + resources = ["${aws_s3_bucket.raw_data_ingestion.arn}*"] + } + + statement { + effect = "Allow" + actions = [ + "s3:ListBucket", + "s3:GetBucketLocation" + ] + resources = ["${aws_s3_bucket.raw_data_ingestion.arn}"] + + condition { + test = "StringLike" + variable = "s3:prefix" + values = ["*"] + } + } +} + +# IAM Policy for snowflake-integration-role +resource "aws_iam_policy" "snowflake_integration_policy" { + name = local.snowflake_integration_policy_name + description = "Client textract permissions" + policy = data.aws_iam_policy_document.snowflake_integration_policy_document.json +} + +# Attach IAM Policy to snowflake-integration-role +resource "aws_iam_role_policy_attachment" "snowflake_integration_policy_attachment" { + role = aws_iam_role.snowflake_integration_role.name + policy_arn = aws_iam_policy.snowflake_integration_policy.arn +} + + +# IAM role - cross-account-role +resource "aws_iam_role" "cross_account_role" { + name = local.cross_account_role_name + + assume_role_policy = jsonencode({ + Version = "2012-10-17" + Statement = [ + { + Action = "sts:AssumeRole" + Effect = "Allow" + Sid = "" + Principal = { + AWS = "arn:aws:iam::873115228912:root" + } + }, + ] + }) +} + +# Construct IAM Policy for cross-account-role +data "aws_iam_policy_document" "cross_account_policy_document" { + statement { + effect = "Allow" + actions = [ + "s3:PutObject", + "s3:GetObject", + "s3:ListBucket", + "s3:PutObjectAcl", + "s3:GetObjectVersion" + ] + resources = [ + "${aws_s3_bucket.raw_data_ingestion.arn}", + "${aws_s3_bucket.raw_data_ingestion.arn}/*" + ] + } +} + +# IAM Policy for cross-account-role +resource "aws_iam_policy" "cross_account_policy" { + name = local.cross_account_policy_name + description = "This role is used for other AWS account to assume inorder to drop files to our data ingestion bucket. At the time of creation, this is being used by CODE DEV to drop all clients list." + policy = data.aws_iam_policy_document.cross_account_policy_document.json +} + +# Attach IAM Policy to cross-account-role +resource "aws_iam_role_policy_attachment" "cross_account_policy_attachment" { + role = aws_iam_role.cross_account_role.name + policy_arn = aws_iam_policy.cross_account_policy.arn +} + +# IAM role - mwaa-exec-role +resource "aws_iam_role" "mwaa_exec_role" { + name = local.mwaa_exec_role_role_name + + assume_role_policy = jsonencode({ + Version = "2012-10-17" + Statement = [ + { + Action = "sts:AssumeRole" + Effect = "Allow" + Sid = "" + Principal = { + Service = ["airflow-env.amazonaws.com","airflow.amazonaws.com"] + } + }, + ] + }) +} + +# Construct IAM Policy for mwaa-exec-role +data "aws_iam_policy_document" "mwaa_exec_policy_document" { + statement { + effect = "Allow" + actions = ["s3:*"] + resources = [ + "${aws_s3_bucket.raw_data_ingestion.arn}/*" + ] + } + statement { + effect = "Allow" + actions = ["airflow:CreateCliToken"] + resources = [ + "arn:aws:airflow:us-east-2:660131068782:environment/doczy-dev-infra-mwaa" + ] + } + statement { + effect = "Allow" + actions = ["airflow:PublishMetrics"] + resources = [ + "arn:aws:airflow:us-east-2:660131068782:environment/doczy-dev-infra-mwaa" + ] + } + statement { + effect = "Deny" + actions = ["s3:ListAllMyBuckets"] + resources = [ + "${aws_s3_bucket.mwaa_resources.arn}", + "${aws_s3_bucket.mwaa_resources.arn}/*" + ] + } + statement { + effect = "Allow" + actions = [ + "s3:GetObject*", + "s3:GetBucket*", + "s3:List*", + "s3:PutObject" + ] + resources = [ + "${aws_s3_bucket.mwaa_resources.arn}", + "${aws_s3_bucket.mwaa_resources.arn}/*" + ] + } + + statement { + effect = "Allow" + actions = [ + "logs:CreateLogStream", + "logs:CreateLogGroup", + "logs:PutLogEvents", + "logs:GetLogEvents", + "logs:GetLogRecord", + "logs:GetLogGroupFields", + "logs:GetQueryResults" + ] + resources = [ + "arn:aws:logs:us-east-2:660131068782:log-group:airflow-doczy-dev-infra-mwaa-*" + ] + } + + statement { + effect = "Allow" + actions = [ + "logs:DescribeLogGroups" + ] + resources = ["*"] + } + + statement { + effect = "Allow" + actions = [ + "cloudwatch:PutMetricData" + ] + resources = ["*"] + } +} + +# IAM Policy for mwaa-exec-role +resource "aws_iam_policy" "mwaa_exec_policy" { + name = local.mwaa_exec_policy_name + description = "" + policy = data.aws_iam_policy_document.mwaa_exec_policy_document.json +} + +# Attach IAM Policy to mwaa-exec-role +resource "aws_iam_role_policy_attachment" "mwaa_exec_policy_attachment" { + role = aws_iam_role.mwaa_exec_role.name + policy_arn = aws_iam_policy.mwaa_exec_policy.arn +} \ No newline at end of file diff --git a/devops-pipeline/other-resources/outputs.tf b/devops-pipeline/other-resources/outputs.tf index f27f57c..a8e15a4 100644 --- a/devops-pipeline/other-resources/outputs.tf +++ b/devops-pipeline/other-resources/outputs.tf @@ -1,3 +1,25 @@ +# S3 bucket name output "devops_s3_bucket_name" { value = aws_s3_bucket.devops.id +} + +output "raw_data_ingestion_bucket_name" { + value = aws_s3_bucket.raw_data_ingestion.id +} + +output "mwaa_resources_bucket_name" { + value = aws_s3_bucket.mwaa_resources.id +} + +# IAM Role ARN +output "snowflake_integration_role_arn" { + value = aws_iam_role.snowflake_integration_role.arn +} + +output "cross_account_role_arn" { + value = aws_iam_role.cross_account_role.id +} + +output "mwaa_exec_role_arn" { + value = aws_iam_role.mwaa_exec_role.id } \ No newline at end of file diff --git a/devops-pipeline/other-resources/variables.tf b/devops-pipeline/other-resources/variables.tf index 1cd1239..dbbfd87 100644 --- a/devops-pipeline/other-resources/variables.tf +++ b/devops-pipeline/other-resources/variables.tf @@ -1,23 +1,23 @@ # required -# variable "aws_profile" { +variable "aws_profile" { + type = string + default = "doczyai" +} + +# # required +# variable "secret_key" { # type = string -# default = "doczyai" # } -# required -variable "secret_key" { - type = string -} - -# required -variable "access_key" { - type = string -} +# # required +# variable "access_key" { +# type = string +# } # required variable "aws_region" { type = string - # default = "us-east-2" + default = "us-east-2" } # required variable "project_name" { @@ -27,7 +27,7 @@ variable "project_name" { # required variable "environment" { type = string - # default = "dev" + default = "dev" } # required variable "client_name" { @@ -44,4 +44,29 @@ variable "devops_s3_bucket" { name = "devops-resources" } -} \ No newline at end of file +} + +# s3 bucket - raw-data-ingestion +variable "raw_data_ingestion_s3_bucket" { + type = object({ + name = string + + }) + default = { + name = "raw-data-ingestion" + + } +} + +# s3 bucket - mwaa-resources +variable "mwaa_resources_s3_bucket" { + type = object({ + name = string + + }) + default = { + name = "mwaa-resources" + + } +} + diff --git a/textract-pipeline/src/lambda-layer/libre-office/lo.tar.zip b/textract-pipeline/src/lambda-layer/libre-office/lo.tar.br similarity index 94% rename from textract-pipeline/src/lambda-layer/libre-office/lo.tar.zip rename to textract-pipeline/src/lambda-layer/libre-office/lo.tar.br index 6a2b9a6..c3ebd1b 100644 Binary files a/textract-pipeline/src/lambda-layer/libre-office/lo.tar.zip and b/textract-pipeline/src/lambda-layer/libre-office/lo.tar.br differ diff --git a/textract-pipeline/src/lambda/batch-creation/index.py b/textract-pipeline/src/lambda/batch-creation/index.py index 0e74a2d..a957056 100644 --- a/textract-pipeline/src/lambda/batch-creation/index.py +++ b/textract-pipeline/src/lambda/batch-creation/index.py @@ -61,14 +61,17 @@ def create_folder_in_s3(client_bucket_name, CONTRACTS_LANDNING_ZONE): # Generate timestamp ID timestamp_id = datetime.now().strftime('%d%m%y%H%M%S') + # Generate Batch ID + batch_id = "batch_"+timestamp_id + # Folder key (name) in S3 bucket - folder_key = f"batch_{timestamp_id}/{CONTRACTS_LANDNING_ZONE}" + folder_key = f"{CONTRACTS_LANDNING_ZONE}".format(batch_id) # Create the folder in S3 bucket try: s3_client.put_object(Bucket=client_bucket_name, Key=folder_key) logger.info("Folder created successfully.") - return f"batch_{timestamp_id}" + return batch_id except Exception as e: logger.error(f"Error creating folder in S3: {str(e)}") raise @@ -83,7 +86,7 @@ def lambda_handler(event, context): if 'client-bucket-name' in event: client_bucket_name = event['client-bucket-name'] - client_id = event.get('client_id', 'dafault_client') + client_id = event.get('client_id', 'default_client') # Read environment variables property_file_path = os.environ.get('PROPERTY_FILE_S3_PATH', '') @@ -102,7 +105,7 @@ def lambda_handler(event, context): generate_batch_logs_input(batch_id,100,'',client_id) return { 'statusCode': 200, - 'body': json.dumps({'batch_id': batch_id, 'landing_zone': CONTRACTS_LANDNING_ZONE}) + 'body': json.dumps({'batch_id': batch_id, 'landing_zone': CONTRACTS_LANDNING_ZONE.format(batch_id)}) } except Exception as e: logger.error(f"Error processing event: {str(e)}") diff --git a/textract-pipeline/src/lambda/call-bedrock/index.py b/textract-pipeline/src/lambda/call-bedrock/index.py new file mode 100644 index 0000000..e3463a8 --- /dev/null +++ b/textract-pipeline/src/lambda/call-bedrock/index.py @@ -0,0 +1,115 @@ +# Import necessary libraries +import json +import logging +import boto3 +from botocore.exceptions import ClientError + +# Configure logging +logger = logging.getLogger() +logger.setLevel(logging.INFO) + +# Lambda function handler +def lambda_handler(event, context): + logger.info("Lambda function invoked.") + logger.info(f"Received event: {json.dumps(event)}") + + # Call the bedrock_llm function and return its result + try: + return call_bedrock_llm(event) + except ThrottlingException: + logger.error("Too many requests. Please wait before trying again.") + raise + +# Function to invoke the bedrock_llm with necessary parameters +def invoke(wrapper, event): + logger.info(f"Event details: {event}") + + # Extract parameters from the event + prompt = event['prompt'] + model_id = event['model_id'] + max_gen_len = event['max_gen_len'] + temperature = event['temperature'] + top_p = event['top_p'] + + try: + # Invoke the bedrock_llm using the provided wrapper + completion = wrapper.invoke_llm(prompt, model_id, max_gen_len, temperature, top_p) + logger.info("Bedrock LLM invoked successfully.") + return completion + + except ClientError as e: + # Log an exception if there is an issue invoking the model + logger.exception(f"Couldn't invoke model {model_id}. Error: {str(e)}") + raise + +# Function to call the Bedrock LLM +def call_bedrock_llm(event): + logger.info("Calling Bedrock LLM.") + + # Initialize Bedrock Runtime client + client = boto3.client(service_name="bedrock-runtime", region_name="us-east-1") + + # Create an instance of BedrockRuntimeWrapper + wrapper = BedrockRuntimeWrapper(client) + + # Invoke the wrapper and return the result + try: + answer = invoke(wrapper, event) + logger.info("Bedrock LLM call completed.") + return answer + except ClientError as e: + if e.response['Error']['Code'] == 'ThrottlingException': + raise ThrottlingException + else: + logger.error(f"Error calling Bedrock LLM: {str(e)}") + raise + +# Class to wrap Bedrock Runtime functionality +class BedrockRuntimeWrapper: + # Constructor to initialize the wrapper with a Bedrock Runtime client + def __init__(self, bedrock_runtime_client): + self.bedrock_runtime_client = bedrock_runtime_client + + # Method to invoke the Bedrock LLM model + def invoke_llm(self, prompt, model_id, max_gen_len, temperature, top_p): + logger.info("Invoking Bedrock LLM model.") + + try: + # Prepare the request body + if 'claude' in model_id.lower(): + # If yes, change the parameter name to 'max_tokens_to_sample' for claude + prompt='Human:'+prompt+'\n\nAssistant:You read and understand the USA healthcrae contract and able to answer questions based on given contract.' + body = { + "prompt": prompt, + "temperature": temperature, + "top_p": top_p, + "max_tokens_to_sample": max_gen_len, # Change the parameter name + } + else: + # If no, use the original parameter name 'max_gen_len' + body = { + "prompt": prompt, + "temperature": temperature, + "top_p": top_p, + "max_gen_len": max_gen_len, + } + + # Invoke the Bedrock Runtime model with the specified parameters + response = self.bedrock_runtime_client.invoke_model( + modelId=model_id, body=json.dumps(body) + ) + + # Parse the response body from JSON + response_body = json.loads(response["body"].read()) + logger.info(f"Response Body from Bedrock LLM: {response_body}") + logger.info("Bedrock LLM model invoked successfully.") + # Return the response body + return response_body + + except ClientError as e: + # Log an error if there is an issue invoking the model + logger.error(f"Problem in invoking model. Error: {str(e)}") + raise + +class ThrottlingException(Exception): + pass diff --git a/textract-pipeline/src/lambda/database-interface-get/index.py b/textract-pipeline/src/lambda/database-interface-get/index.py index 9c93bdc..dab457d 100644 --- a/textract-pipeline/src/lambda/database-interface-get/index.py +++ b/textract-pipeline/src/lambda/database-interface-get/index.py @@ -100,16 +100,17 @@ def lambda_handler(event, context): # Extract column names from the cursor description column_names = [col[0] for col in cursor.description] - + logger.info(f"column_names:: {column_names}") # Fetch all rows and convert each row to a dictionary with column names as keys rows = cursor.fetchall() result = [dict(zip(column_names, row)) for row in rows] - + return { 'statusCode': 200, 'body': json.dumps(result, default=default_converter) } except Exception as e: + logger.error(f"An error occurred: {str(e)}") return { 'statusCode': 400, 'body': json.dumps(str(e)) diff --git a/textract-pipeline/src/lambda/database-interface/index.py b/textract-pipeline/src/lambda/database-interface/index.py index 1dedbf6..e840a62 100644 --- a/textract-pipeline/src/lambda/database-interface/index.py +++ b/textract-pipeline/src/lambda/database-interface/index.py @@ -158,6 +158,30 @@ def construct_doc_insert_sql(data): sql = f"INSERT INTO STG.DOCUMENT_LOGS ({columns}) VALUES ({values});" return sql +def construct_doczy_pipeline_insert_sql(data): + """ + Constructs the SQL for an insert operation + + Sample return value: + INSERT INTO STG.DOCUMENT_LOGS (DOCUMENT_ID,BATCH_ID, JOB_ID, STAGE, TEXTRACT_STATUS, BUCKET_NAME, FILE_NAME, FILE_PATH, DOCUMENT_TYPE, PAYER_SIGNED, PROVIDER_SIGNED, GROUP_ID, CREATED_TIME, MODIFIED_TIME, CREATED_BY, MODIFIED_BY, ORIGINAL_FILE_EXTENSION, NO_OF_PAGES, FILE_SIZE) + VALUES ('doc_1212','batch_101', 'J123456', 'Processing', 'Success', 'doc-bucket', 'file1.pdf', '/documents/2023/', 'Report', True, False, 'G100', '2024-03-21 10:00:00', '2024-03-21 10:00:00', 'admin', 'admin', 'pdf', 10, 1048576); + """ + columns = ', '.join(data.keys()) + values = ', '.join(["'" + str(value).replace("'", "''") + "'" if isinstance(value, str) else str(value) for value in data.values()]) + sql = f"INSERT INTO STG.DOCZY_PIPELINE_RAW_OUTPUT ({columns}) VALUES ({values});" + return sql + +def construct_doczy_pipeline_update_sql(data, document_id): + """ + Constructs the SQL for an update operation + + Sample return value: + UPDATE STG.DOCUMENT_LOGS SET TEXTRACT_STATUS = 'Failed', MODIFIED_TIME = '2024-03-22 15:00:00', MODIFIED_BY = 'admin' WHERE DOCUMENT_ID = 1001; + """ + set_clauses = ', '.join([f"{key} = '" + str(value).replace("'", "''") + "'" if isinstance(value, str) else f"{key} = {value}" for key, value in data.items()]) + sql = f"UPDATE STG.DOCZY_PIPELINE_RAW_OUTPUT SET {set_clauses} WHERE DOCUMENT_ID = '{document_id}';" + logger.info(f"Executing the following logging SQL statement: {sql}") + return sql def construct_doc_update_sql(data, document_id): """ @@ -244,6 +268,13 @@ def lambda_handler(event, context): elif operation == 'update': batch_id = data.pop('BATCH_ID', None) sql = construct_batch_update_sql(data, batch_id) + + elif table == 'DOCZY_PIPELINE_RAW_OUTPUT': + if operation == 'insert': + sql = construct_doczy_pipeline_insert_sql(data) + #elif operation == 'update': + #batch_id = data.pop('BATCH_ID', None) + #sql = construct_doczy_pipeline_update_sql(data, batch_id) elif table == 'CLIENT_LOGS': if operation == 'insert': diff --git a/textract-pipeline/src/lambda/prompt-orchestrator/index.py b/textract-pipeline/src/lambda/prompt-orchestrator/index.py new file mode 100644 index 0000000..cfacc59 --- /dev/null +++ b/textract-pipeline/src/lambda/prompt-orchestrator/index.py @@ -0,0 +1,253 @@ +import json +import boto3 +import logging +from configparser import ConfigParser +import os +from urllib.parse import unquote_plus + +# Configure logging +logger = logging.getLogger() +logger.setLevel(logging.INFO) + +sqs = boto3.client('sqs') +# Initialize S3 & Textract clients +s3_client = boto3.client('s3') + +def lambda_handler(event, context): + + try: + # Read environment variables + property_file_path = os.environ.get('PROPERTY_FILE_S3_PATH', '') + + # Read config.properties + file_path_array = property_file_path.split("/") + + # Valid if file_path_array has more than 2 elements + if len(file_path_array) > 1: + + # Extract BUCKET_NAME and config_file_path + S3_BUCKET_NAME = file_path_array[0] + CONFIG_FILE_PATH = "/".join(file_path_array[1:]) + + # Load config file + config_dict = load_config_from_s3(S3_BUCKET_NAME, CONFIG_FILE_PATH) + + # Process each message from the SQS event + for record in event['Records']: + # Extract the message body from the record + record_body = json.loads(record['body']) + #logger.info('Message Count: ', str(len(record_body['Records'])) ) + for sqs_record in record_body['Records']: + # Construct the source paths + source_path = unquote_plus(sqs_record['s3']['object']['key']) + document_id = get_filename_from_path(source_path) + document_id = os.path.splitext(document_id)[0] + + prompt_config_query = "SELECT DISTINCT PC.FIELD_NAME,LENGTH(PC.PROMPT) as PROMPT_LENGTH, PC.FM_MODEL_ID,PC.GROUP_ID FROM PROMPT_CONFIG PC JOIN DOCUMENT_LOGS DL ON DL.GROUP_ID LIKE '%'||PC.GROUP_ID ||'%' AND DL.DOCUMENT_ID = '"+document_id+"' ORDER BY PROMPT_LENGTH" + field_Prompt_group_response = call_snowflake_db_select_lambda(prompt_config_query) + + # Maximum prompt length per group + max_prompt_length_per_group = 5000 # Adjust as needed + + # Divide data into batches + batches = divide_into_batches(field_Prompt_group_response, max_prompt_length_per_group) + + # Print batches + for i, (key, batch_list) in enumerate(batches.items(), 1): + group_id, fm_model_id = key + for j, batch in enumerate(batch_list, 1): + send_batch_to_sqs(batch,"https://sqs.us-east-2.amazonaws.com/660131068782/doczy-prompt-processor-sqs-dev",S3_BUCKET_NAME+"/"+source_path,i) + return { + 'statusCode': 200, + 'body': json.dumps('Message sent to SQS queue') + } + except Exception as e: + # Handle other exceptions + error_message = f"Unexpected error: {e}" + logger.error(error_message) + return { + 'statusCode': 500, + 'body': error_message + } + + +# Function to retrieve configuration values from S3 +def load_config_from_s3(bucket_name, file_key): + + # Download the config file from S3 + response = s3_client.get_object(Bucket=bucket_name, Key=file_key) + config_content = response['Body'].read().decode('utf-8') + + # Parse the config file + config_parser = ConfigParser() + config_parser.read_string(config_content) + + # Convert the configuration to a dictionary + config_dict = {} + for section in config_parser.sections(): + config_dict[section] = {key.upper(): value for key, value in config_parser.items(section)} + + return config_dict + +def get_filename_from_path(full_path): + return os.path.basename(full_path) + +# Function to get s3 object tags +def get_s3_object_tags(bucket_name, object_key): + + try: + # Get object tags + response = s3_client.get_object_tagging( + Bucket=bucket_name, + Key=object_key + ) + # Extract tags from the response and convert to dictionary + tags_list = response['TagSet'] + tags_dict = {tag['Key']: tag['Value'] for tag in tags_list} + + return tags_dict + + except Exception as e: + logger.exception(f"Error: {e}") + return None + + +def call_snowflake_db_select_lambda(query): + # Create a Lambda client + lambda_client = boto3.client('lambda') + + # Specify your Lambda function name + lambda_function_name = 'doczy-dev-get-snowflake-logs' + + # Prepare payload + payload = { + 'query': query + } + try: + # Call the Lambda function + response = lambda_client.invoke( + FunctionName=lambda_function_name, + InvocationType='RequestResponse', # Synchronous invocation + Payload=json.dumps(payload) + ) + + # Return response body + response_body = response['Payload'].read().decode('utf-8') + return response_body + except Exception as e: + logger.exception(f"Error: {e}") + return None + +def extract_group_id(json_string): + try: + logger.info(f"json_string: {json_string}") + data = json.loads(json_string) + body = json.loads(data['body']) # Parse the inner JSON string + group_id = body[0]['GROUP_ID'] + group_ids = group_id.split(',') + group_ids_quoted = ["'" + group_id.strip() + "'" for group_id in group_ids] + + # Join the elements back together with commas + return ','.join(group_ids_quoted) + except (json.JSONDecodeError, KeyError, IndexError) as e: + logger.error("Error parsing JSON or extracting GROUP_ID:", e) + return None + +def extract_body_tags(json_data): + try: + # Parse the JSON string + data = json.loads(json_data) + # Extract the body as a list + body_list = json.loads(data['body']) + # Create a list for tags within the body + tags_list = [] + # Iterate over the list elements and extract values + for item in body_list: + tags_list.extend(item.items()) + + logger.info(f"tags_list: {tags_list}") + return tags_list + + except (json.JSONDecodeError, KeyError) as e: + logger.error("Error parsing JSON or extracting 'body' tag:", e) + return None + + +def divide_into_batches(data, max_prompt_length_per_group): + + try: + # Parse the input data + data_dict = json.loads(data) + body = data_dict["body"] + body_dict = json.loads(body) + logger.info(f"Parsed body: {type(body_dict)}") + + # Initialize batches dictionary + batches = {} + # Initialize variables to track prompt length in each group + prompt_length_per_group = {} + + # Divide data into batches based on group_id and fm_model_id + for item in body_dict: + + group_id = item["GROUP_ID"] + fm_model_id = item["FM_MODEL_ID"] + prompt_length = item["PROMPT_LENGTH"] + + # Check if batch exists for the current group_id and fm_model_id + if (group_id, fm_model_id) not in batches: + batches[(group_id, fm_model_id)] = [] + prompt_length_per_group[(group_id, fm_model_id)] = 0 + + # Check if adding current item's prompt length exceeds the maximum + if prompt_length_per_group[(group_id, fm_model_id)] + prompt_length > max_prompt_length_per_group: + # If exceeds, move to the next batch for this group and model id + prompt_length_per_group[(group_id, fm_model_id)] = 0 + batches[(group_id, fm_model_id)].append([]) + + # Check if the list of batches is empty or not + if not batches[(group_id, fm_model_id)]: + # If it's empty, create a new sub-batch + batches[(group_id, fm_model_id)].append([]) + + # Retrieve the current batch (last sub-batch) + current_batch = batches[(group_id, fm_model_id)][-1] + + + # Append the current item to the current batch + current_batch.append(item) + # Update prompt length for the current group and model id + prompt_length_per_group[(group_id, fm_model_id)] += prompt_length + + logger.info(f"Generated batches: {batches}") + return batches + + except Exception as e: + logger.exception("Error occurred in divide_into_batches") + raise e + + +def send_batch_to_sqs(batch, queue_url,source_path,prompt_batch_id): + # Initialize SQS client + sqs = boto3.client('sqs') + + # Convert batch to JSON string + batch_json = json.dumps(batch) + try: + result = {"field_list": batch_json, + "file_name": source_path, + "prompt_batch_id" : prompt_batch_id + } + logger.info(f"Message sent to SQS queue: {json.dumps(result)}") + # Send message to SQS queue + response = sqs.send_message( + QueueUrl=queue_url, + MessageBody=json.dumps(result) + ) + # Log successful message sending + logger.info(f"Message sent to SQS queue: {response['MessageId']}") + return True + + except Exception as e: + logger.error('Error sending message to SQS queue:', e) + return False \ No newline at end of file diff --git a/textract-pipeline/src/lambda/prompt-processor/index.py b/textract-pipeline/src/lambda/prompt-processor/index.py new file mode 100644 index 0000000..a8413a3 --- /dev/null +++ b/textract-pipeline/src/lambda/prompt-processor/index.py @@ -0,0 +1,411 @@ +import os +import json +import datetime +import logging +import boto3 +import os +from configparser import ConfigParser +from urllib.parse import urlparse +from urllib.parse import unquote_plus +from urllib.parse import urlencode +import traceback + + +# Configure logging +logger = logging.getLogger() +logger.setLevel(logging.INFO) + +# Initialize S3 & Textract clients +s3_client = boto3.client('s3') +DATABASE_LOGGING_LAMBDA_FUNCTION_NAME = "" + + +# Function to load configuration from S3 +def load_config_from_s3(bucket_name, file_key): + + # Download the config file from S3 + response = s3_client.get_object(Bucket=bucket_name, Key=file_key) + config_content = response['Body'].read().decode('utf-8') + + # Parse the config file + config_parser = ConfigParser() + config_parser.read_string(config_content) + + # Convert the configuration to a dictionary + config_dict = {} + for section in config_parser.sections(): + config_dict[section] = {key.upper(): value for key, value in config_parser.items(section)} + + return config_dict + +# Function to get s3 object tags +def get_s3_object_tags(bucket_name, object_key): + + try: + # Get object tags + response = s3_client.get_object_tagging( + Bucket=bucket_name, + Key=object_key + ) + + # Extract tags from the response and convert to dictionary + tags_list = response['TagSet'] + tags_dict = {tag['Key']: tag['Value'] for tag in tags_list} + + return tags_dict + + except Exception as e: + logger.exception(f"Error: {e}") + logger.error(f"Error: {e}") + return None + +def lambda_handler(event, context): + + # Read environment variables + global DATABASE_LOGGING_LAMBDA_FUNCTION_NAME + DATABASE_LOGGING_LAMBDA_FUNCTION_NAME = os.environ.get('DATABASE_LOGGING_LAMBDA_FUNCTION_NAME', '') + + property_file_path = os.environ.get('PROPERTY_FILE_S3_PATH', '') + # Read config.properties + file_path_array = property_file_path.split("/") + # Valid if file_path_array has more than 2 elements + if len(file_path_array) > 1: + # Extract BUCKET_NAME and config_file_path + S3_BUCKET_NAME = file_path_array[0] + CONFIG_FILE_PATH = "/".join(file_path_array[1:]) + + logger.info(f'S3_BUCKET_NAME: {S3_BUCKET_NAME}') + logger.info(f'CONFIG_FILE_PATH: {CONFIG_FILE_PATH}') + + # Load config file + config_dict = load_config_from_s3(S3_BUCKET_NAME, CONFIG_FILE_PATH) + try: + # Process each message from the SQS event + for record in event['Records']: + + # Extract the message body from the record + logger.info(f"record: {record}") + record_body = json.loads(record['body']) + logger.info(f"record_body: {record_body}") + logger.info(f"record_body type: {type(record_body)}") + + field_list = record_body["field_list"] + prompt_batch_id = record_body["prompt_batch_id"] + file_name_with_path = record_body["file_name"] + logger.info(f"field_list: {field_list}") + logger.info(f"file_name: {file_name_with_path}") + + path_parts = file_name_with_path.split('/') + + # Extract the first string + s3_bucket = path_parts[0] + file_name_with_subfolder = '/'.join(path_parts[1:]) + fields = json.loads(field_list) + field_names = [field["FIELD_NAME"] for field in fields] + required_field_list = "'" + "', '".join(field_names) + "'" + fm_model_id = fields[0]["FM_MODEL_ID"] if fields else None + + # Download the file content from S3 + file_content = read_text_from_s3(file_name_with_path) + + prompt_config_query = "SELECT LISTAGG(CONCAT('[{\"FIELD_NAME:\"',FIELD_NAME,'\"},{\"PROMPT\":\"',PROMPT,'\"}]'),',') FROM DOCZY_DEV.STG.PROMPT_CONFIG WHERE FIELD_NAME IN ("+required_field_list+")" + + field_prompt_group_response = call_snowflake_db_select_lambda(prompt_config_query) + data_dict = json.loads(field_prompt_group_response) + body = data_dict["body"] + body_dict = json.loads(body) + # Convert body_dict to a string + body_str = json.dumps(body_dict) + logger.info(f"Parsed body: {type(body_dict)}") + logger.info(f"body_str: {body_str}") + + file_content += body_str+" Provide answers in VALID JSON in 'completion' tag format against the tag in the bracket along with the file page number. [{\"FIELD_NAME\":,\"ANSWER\":},\"PAGE_NO\":}]" + logger.info(f"file_content: {file_content}") + + api_payload = { + "prompt": file_content, + "model_id": 'anthropic.claude-instant-v1', + #"model_id": fm_model_id, + "max_gen_len": 40000, + "temperature": 0.3, + "top_p": 0.5 + } + api_response = make_api_call(api_payload) + + logger.info(f"API Response==: {api_response}") + # Save the response text to a file + #LLM_RESPONSE_FILE_LOCATION = config_dict['FOLDER_LOCATIONS']['LLM_RESPONSE_FILE_LOCATION'].format(batch_id) + LLM_RESPONSE_FILE_LOCATION = "/backup/contract-text-file/" + + logger.info(f"file_name_with_subfolder = {file_name_with_subfolder}") + logger.info(f"s3_bucket = {s3_bucket}") + final_path = LLM_RESPONSE_FILE_LOCATION+file_name_with_subfolder + logger.info(f"final_path = {final_path}") + + # Extract response from the API response + response_text = extract_response(api_response) + logger.info(f"Extracted 'generation' text: {response_text}") + new_file_name_with_subfolder = create_new_text_file(file_name_with_subfolder,prompt_batch_id) + upload_response_to_s3(response_text, s3_bucket, new_file_name_with_subfolder,"temp_tags") + save_output_to_db(final_path,response_text) + + + except Exception as e: + # Log any unhandled exceptions + logger.error(f"An error occurred: {str(e)}") + logger.error(traceback.format_exc()) + raise e + +def save_output_to_db(document_id,response_text): + logger.info(f"Saving records to DB for document_id: {document_id}") + logger.info(f"Saving records to DB for response_text: {response_text}") + #generate_doczy_pipeline_raw_output(document_id,sf_db_col_name, raw_value, original_page_number) + generate_doczy_pipeline_raw_output(document_id,"PAYER_NAME", "Client Company Inc.", "1") + +def create_new_text_file(file_name_with_subfolder,prompt_batch_id): + parts = file_name_with_subfolder.split('/') + # Modify the last part (filename) by adding '_temp' before the file extension + filename_parts = parts[-1].split('.') + new_filename = filename_parts[0] +'_' +str(prompt_batch_id) + '.' + filename_parts[1] + + # Combine the parts back into a string + parts[-1] = new_filename + new_string = '/'.join(parts) + return new_string + +def read_text_from_s3(s3_path): + # Initialize S3 client + s3 = boto3.client('s3') + + try: + bucket_name, file_key = s3_path.split("/", 1) + logger.info(f"Reading file from S3 - bucket_name: {bucket_name}, file_key: {file_key}") + # Get object from S3 + response = s3.get_object(Bucket=bucket_name, Key=file_key) + + # Read text from the response + text = response['Body'].read().decode('utf-8') + + return text + except Exception as e: + print("Error:", e) + return None + + + +def download_file_from_s3(bucket, key, region): + try: + logger.info(f"Downloading file from S3 - Bucket: {bucket}, Key: {key}, Region: {region}") + s3_client = boto3.client('s3', region_name=region) + response = s3_client.get_object(Bucket=bucket, Key=key) + file_content = response['Body'].read().decode('utf-8') + return file_content + except Exception as e: + # Log the error + logger.error(f"Error downloading file from S3: {str(e)}") + raise e + +def make_api_call(data): + + try: + lambda_function_name = 'bedrock_call' + input_json_payload = json.dumps(data) + logger.info(input_json_payload) + response = call_lambda_function(lambda_function_name, input_json_payload) + logger.info(f"Resposne from LLM: {response}") + return response + + except Exception as e: + # Log the error + logger.error(f"Error making API call: {str(e)}") + raise e + +def upload_response_to_s3(response_text, bucket_name, object_key,tags): + s3_client = boto3.client('s3') + text_value = str(response_text) + try: + # Upload the JSON response to S3 + s3_client.put_object( + Bucket=bucket_name, + Key=object_key, + Body=text_value, + ContentType='application/txt', + Tagging=tags + ) + logger.info(f"Saved analysis response to S3: {object_key}") + except Exception as e: + logger.error(f"Error saving response to S3 - {object_key}: {str(e)}") + +def extract_response(api_response): + try: + clean_json_response(api_response,["\n",""]) + generation_text = api_response.get("generation", "") + + if generation_text is None or not generation_text.strip(): + generation_text = api_response.get("completion", "") + if generation_text is None or not generation_text.strip(): + return generation_text + else: + start_index = generation_text.find('[') # Find the start of JSON + json_data_str = generation_text[start_index:] # Extract JSON data + # Parse the JSON data into a Python list + json_data = json.loads(json_data_str) + return json_data + return generation_text + except Exception as e: + # Log the error + logger.error(f"Error extracting 'generation' from API response: {str(e)}") + raise e + + + +def clean_json_response(json_data, substrings_to_remove): + """ + Clean JSON data by removing unwanted substrings from all string values recursively. + + Args: + - json_data (dict): JSON data (Python dictionary) to be cleaned. + - substrings_to_remove (list): List of substrings to be removed from string values. + + Returns: + - dict: Cleaned JSON data (Python dictionary). + """ + logger.info(f"clean_json_response = json_data : {json_data}") + # Function to recursively clean dictionary values + def clean_values(obj): + logger.info(f"clean_json_response = obj : {obj}") + if isinstance(obj, str): + # Remove unwanted substrings from string values + for substring in substrings_to_remove: + obj = obj.replace(substring, "") + return obj + elif isinstance(obj, dict): + # Recursively clean dictionary values + return {key: clean_values(value) for key, value in obj.items()} + elif isinstance(obj, list): + # Recursively clean list values + return [clean_values(item) for item in obj] + else: + return obj + + # Clean JSON data + cleaned_json_data = clean_values(json_data) + return cleaned_json_data + + +def get_filename_from_path(full_path): + return os.path.basename(full_path) + + +def call_lambda_function(lambda_function_name, json_payload, aws_region='us-east-2'): + + # Create a Lambda client + lambda_client = boto3.client('lambda', region_name=aws_region) + try: + # Invoke the Lambda function + response = lambda_client.invoke( + FunctionName=lambda_function_name, + InvocationType='RequestResponse', # Use 'Event' for asynchronous invocation + Payload=json_payload, + ) + # Parse and return the response payload + response_payload = json.loads(response['Payload'].read().decode('utf-8')) + return response_payload + + except Exception as e: + # Handle any exceptions (e.g., Lambda function not found, permission issues) + print(f"Error calling Lambda function: {e}") + return {'error': str(e)} + + +def generate_document_logs_input(document_id, stage): + current_time = datetime.datetime.now().isoformat() + + data = { + "operation": "update", + "table": "DOCUMENT_LOGS", + "data": { + "DOCUMENT_ID": document_id, + "STAGE" : stage, + "MODIFIED_TIME": current_time, + "MODIFIED_BY": "PROMPT_PROCESSOR" + } + } + logger.info(f"Request: {data}") + + lambda_client = boto3.client('lambda') + response = lambda_client.invoke( + FunctionName=DATABASE_LOGGING_LAMBDA_FUNCTION_NAME, + InvocationType='Event', # Asynchronous invocation + Payload=json.dumps(data).encode('utf-8') + ) + + logger.info("Generated document logs input successfully") + return response + +def call_snowflake_db_select_lambda(query): + # Create a Lambda client + lambda_client = boto3.client('lambda') + + # Specify your Lambda function name + lambda_function_name = 'doczy-dev-get-snowflake-logs' + + # Prepare payload + payload = { + 'query': query + } + try: + # Call the Lambda function + response = lambda_client.invoke( + FunctionName=lambda_function_name, + InvocationType='RequestResponse', # Synchronous invocation + Payload=json.dumps(payload) + ) + + # Return response body + response_body = response['Payload'].read().decode('utf-8') + return response_body + except Exception as e: + logger.exception(f"Error: {e}") + return None + + +def generate_doczy_pipeline_raw_output(document_id,sf_db_col_name, raw_value, original_page_number): + current_time = datetime.datetime.now().isoformat() + """ + DOCUMENT_ID VARCHAR(16777216) NOT NULL, + SF_DB_COL_NAME VARCHAR(255), + ACTUAL_VALUE_STORED VARCHAR(16777216), + RAW_VALUE VARCHAR(16777216), + NEW_EXTRACTED_VALUE VARCHAR(16777216), + SNIPPET VARCHAR(16777216), + ORIGINAL_PAGE_NUMBER VARCHAR(255), + NEW_PAGE_NUMBER VARCHAR(255), + RESULT VARCHAR(16777216), + BATCH_ID VARCHAR(255), + CREATED_TIME TIMESTAMP_NTZ(9), + MODIFIED_TIME TIMESTAMP_NTZ(9) + """ + + data = { + "operation": "insert", + "table": "DOCZY_PIPELINE_RAW_OUTPUT", + "data": { + "DOCUMENT_ID" : document_id, + "SF_DB_COL_NAME": sf_db_col_name, + "RAW_VALUE" : raw_value, + "ORIGINAL_PAGE_NUMBER" : original_page_number, + "CREATED_TIME" : current_time + } + } + logger.info(f"Request: {data}") + + lambda_client = boto3.client('lambda') + response = lambda_client.invoke( + FunctionName='doczy-dev-snowflake-log', + InvocationType='Event', # Asynchronous invocation + Payload=json.dumps(data).encode('utf-8') + ) + + logger.info("Generated document logs input successfully") + return response \ No newline at end of file diff --git a/textract-pipeline/src/lambda/s3-folder-details/index.py b/textract-pipeline/src/lambda/s3-folder-details/index.py new file mode 100644 index 0000000..f155181 --- /dev/null +++ b/textract-pipeline/src/lambda/s3-folder-details/index.py @@ -0,0 +1,53 @@ +import json +import boto3 +from urllib.parse import urlparse + +def list_files_s3(s3_url, file_extension=None): + # Parse the S3 URL to extract the bucket and key + parsed_url = urlparse(s3_url) + bucket = parsed_url.netloc + key = parsed_url.path.lstrip('/') + + # Create an S3 client + s3_client = boto3.client('s3') + + # List objects in the specified S3 path + response = s3_client.list_objects_v2(Bucket=bucket, Prefix=key) + + # Extract details of each object based on the specified file extension or retrieve all files + files_details = [] + for obj in response.get('Contents', []): + # Check if a file extension is specified and filter based on it, or retrieve all files + if file_extension is None or file_extension == "*" or obj['Key'].lower().endswith(f".{file_extension.lower()}"): + # Extract specific details for each object + file_details = { + 'Key': obj['Key'], + 'LastModified': obj['LastModified'].isoformat(), + 'Size': obj['Size'], + 'ETag': obj['ETag'] + } + # Append the details to the list + files_details.append(file_details) + + return files_details + +def lambda_handler(event, context): + # Extract the S3 URL and file extension from the Lambda event input + s3_url = event.get('s3_url') + file_extension = event.get('file_extension') + + # Check if the S3 URL is provided + if not s3_url: + return { + 'statusCode': 400, + 'body': json.dumps('Error: S3 URL is missing in the input.') + } + + # Retrieve details of files in the specified S3 path with optional file extension filtering + files_details = list_files_s3(s3_url, file_extension) + + # Return the details as a JSON response + return { + 'statusCode': 200, + 'body': json.dumps(files_details) + } diff --git a/textract-pipeline/src/s3-files/config.properties b/textract-pipeline/src/s3-files/config.properties index 85035dc..c0c0456 100644 --- a/textract-pipeline/src/s3-files/config.properties +++ b/textract-pipeline/src/s3-files/config.properties @@ -5,24 +5,29 @@ SQS_QUEUE_URL=https://sqs.{aws_region}.amazonaws.com/{aws_account_id}/doczy-text PDF_SQS_QUEUE_URL=https://sqs.{aws_region}.amazonaws.com/{aws_account_id}/doczy-textract-pdf-sqs-dev [FOLDER_LOCATIONS] -CONTRACTS_LANDNING_ZONE=contracts_landing_zone/ -ALL_PDF_LOCATION=batches/{}/all_pdf_files/ -SOURCE_LOCATION=batches/{}/source-documents-pdf/ -SOURCE_DOCX_LOCATION=batches/{}/source-documents-docx/ -SOURCE_DOCX_PROCESSED_LOCATION=batches/{}/source-documents-docx-processed/ -SOURCE_DOCX_UNPROCESSED_LOCATION=batches/{}/source-documents-docx-unprocessed/ -STAGING_LOCATION=batches/{}/textract-sender-staging-pdf/ -OUTPUT_LOCATION=batches/{}/textract-output-json/ -PROCESSED_LOCATION=batches/{}/textract-receiver-processed-file/ -UNPROCESSED_LOCATION=batches/{}/textract-receiver-unprocessed-file/ -CONTRACT_TEXT_FILE_LOCATION=batches/{}/contract-text-file/ -LLM_RESPONSE_FILE_LOCATION=batches/{}/llm-response-file/ -INVALID_PDF_FILE_LOCATION=batches/{}/invalid-pdf-file/ +CONTRACTS_LANDNING_ZONE=contracts-landing-zone/{}/ +ALL_PDF_LOCATION=all-pdfs/{}/ +SOURCE_LOCATION=valid-pdfs/{}/ +INVALID_PDF_FILE_LOCATION=invalid-pdfs/{}/ +SOURCE_DOCX_LOCATION=all-docx/{}/ +SOURCE_DOCX_PROCESSED_LOCATION=processed-docx/{}/ +SOURCE_DOCX_UNPROCESSED_LOCATION=unprocessed-docx/{}/ +SOURCE_TIFF_LOCATION=all-tiff/{}/ +SOURCE_TIFF_PROCESSED_LOCATION=processed-tiff/{}/ +SOURCE_TIFF_UNPROCESSED_LOCATION=unprocessed-tiff/{}/ +STAGING_LOCATION=textract-sender-staging-pdfs/{}/ +OUTPUT_LOCATION=textract-output-json/{}/ +PROCESSED_LOCATION=textract-receiver-processed-pdfs/{}/ +UNPROCESSED_LOCATION=textract-receiver-unprocessed-pdfs/{}/ +CONTRACT_TEXT_FILE_LOCATION=contract-text-file/{}/ +LLM_RESPONSE_FILE_LOCATION=llm-response-file/{}/ +TEXTRACT_SIGNATURE_JSON=textract-signature-output-json/{}/ +TEXTRACT_TABLE_FORM_JSON=textract-table-form-output-json/{}/ [OTHERS] SENDER_MAX_FILES=10 PROCESS_TYPE=ANALYSIS -ANALYSIS_FEATURE_TYPE=LAYOUT +ANALYSIS_FEATURE_TYPE=LAYOUT,TABLES,SIGNATURES JOB_TAG=healthcare-contract LOG_LEVEL=DEBUG \ No newline at end of file diff --git a/textract-pipeline/terraform/main.tf b/textract-pipeline/terraform/main.tf index f4103b1..ca35b17 100644 --- a/textract-pipeline/terraform/main.tf +++ b/textract-pipeline/terraform/main.tf @@ -35,6 +35,14 @@ locals { vpc_id = "vpc-0f9d7c913f6522079" secret_manager_name = "doczy-dev-db-svc-acc" devops_s3_bucket_name = "doczyai-use2-d-infra-s3-devops-resources" + api_access_required_roles_arns = ["arn:aws:iam::660131068782:role/doczy-dev-streamlit-ec2-role"] + } + + uat = { + vpc_id = "vpc-0392396d0e7bdd77f" + secret_manager_name = "" + devops_s3_bucket_name = "" + api_access_required_roles_arns = [] } } @@ -60,10 +68,10 @@ locals { } } provider "aws" { - # profile = var.aws_profile + profile = var.aws_profile - access_key = var.access_key - secret_key = var.secret_key + # access_key = var.access_key + # secret_key = var.secret_key default_tags { tags = local.common_tags @@ -84,6 +92,7 @@ module "textract_pipeline" { vpc_id = local.environment_resource_map[var.environment].vpc_id secret_manager_name = local.environment_resource_map[var.environment].secret_manager_name devops_s3_bucket = local.environment_resource_map[var.environment].devops_s3_bucket_name + api_access_required_roles_arns = local.environment_resource_map[var.environment].api_access_required_roles_arns lambda_role_arn = aws_iam_role.lambda_role.arn textract_role_arn = aws_iam_role.textract_role.arn @@ -128,4 +137,26 @@ resource "aws_iam_role" "textract_role" { }, ] }) +} + +data "aws_iam_policy" "BedrockFullAccess" { + arn = "arn:aws:iam::aws:policy/AmazonBedrockFullAccess" +} +data "aws_iam_policy" "S3FullAccess" { + arn = "arn:aws:iam::aws:policy/AmazonS3FullAccess" +} +data "aws_iam_policy" "CloudWatchFullAccess" { + arn = "arn:aws:iam::aws:policy/CloudWatchLogsFullAccess" +} +resource "aws_iam_role_policy_attachment" "bedrock-policy-attach" { + role = "${aws_iam_role.lambda_role.name}" + policy_arn = "${data.aws_iam_policy.BedrockFullAccess.arn}" +} +resource "aws_iam_role_policy_attachment" "s3-policy-attach" { + role = "${aws_iam_role.lambda_role.name}" + policy_arn = "${data.aws_iam_policy.S3FullAccess.arn}" +} +resource "aws_iam_role_policy_attachment" "cloudwatch-policy-attach" { + role = "${aws_iam_role.lambda_role.name}" + policy_arn = "${data.aws_iam_policy.CloudWatchFullAccess.arn}" } \ No newline at end of file diff --git a/textract-pipeline/terraform/modules/aws-textract-pipeline-part1/main.tf b/textract-pipeline/terraform/modules/aws-textract-pipeline-part1/main.tf index 4e9b6a3..9addf85 100644 --- a/textract-pipeline/terraform/modules/aws-textract-pipeline-part1/main.tf +++ b/textract-pipeline/terraform/modules/aws-textract-pipeline-part1/main.tf @@ -39,9 +39,11 @@ locals { sender_queue_name = "${local.sqs_prefix}-${var.sender_sqs_queue.name}" receiver_queue_name = "${local.sqs_prefix}-${var.receiver_sqs_queue.name}" text_creation_queue_name = "${local.sqs_prefix}-${var.text_creation_sqs_queue.name}" + prompt_orchestrator_queue_name = "${local.sqs_prefix}-${var.prompt_orchestrator_queue.name}" + prompt_processor_queue_name = "${local.sqs_prefix}-${var.prompt_processor_queue.name}" # SNS names - textract_notification_sns_name = "${local.sns_prefix}-${var.textract_notification_sns.name}" + textract_notification_sns_name = "AmazonTextract-${local.sns_prefix}-${var.textract_notification_sns.name}" # S3 names processing_s3_bucket_name = "${local.s3_prefix}-textract-processing-001" @@ -333,6 +335,76 @@ module "text_creation_sqs" { } } +# prompt-orchestrator SQS Queue +module "prompt_orchestrator_sqs" { + source = "terraform-aws-modules/sqs/aws" + version = "4.1.1" + name = local.prompt_orchestrator_queue_name + delay_seconds = var.prompt_orchestrator_queue.delay_seconds + max_message_size = var.prompt_orchestrator_queue.max_message_size + message_retention_seconds = var.prompt_orchestrator_queue.message_retention_seconds + visibility_timeout_seconds= var.prompt_orchestrator_queue.visibility_timeout_seconds + tags = local.common_tags + + create_queue_policy = true + + queue_policy_statements = { + s3 = { + + sid = "AllowS3SendMessage" + effect = "Allow" + + principals = [{ + type = "Service" + identifiers = ["s3.amazonaws.com"] + }] + + actions = ["sqs:SendMessage"] + + conditions = [{ + test = "ArnEquals" + variable = "aws:SourceArn" + values = [aws_s3_bucket.textract_processing.arn] + }] + } + } +} + +# prompt-processor SQS Queue +module "prompt_processor_sqs" { + source = "terraform-aws-modules/sqs/aws" + version = "4.1.1" + name = local.prompt_processor_queue_name + delay_seconds = var.prompt_processor_queue.delay_seconds + max_message_size = var.prompt_processor_queue.max_message_size + message_retention_seconds = var.prompt_processor_queue.message_retention_seconds + visibility_timeout_seconds= var.prompt_processor_queue.visibility_timeout_seconds + tags = local.common_tags + + create_queue_policy = true + + queue_policy_statements = { + s3 = { + + sid = "AllowS3SendMessage" + effect = "Allow" + + principals = [{ + type = "Service" + identifiers = ["s3.amazonaws.com"] + }] + + actions = ["sqs:SendMessage"] + + conditions = [{ + test = "ArnEquals" + variable = "aws:SourceArn" + values = [aws_s3_bucket.textract_processing.arn] + }] + } + } +} + resource "aws_s3_bucket" "textract_processing" { bucket = local.processing_s3_bucket_name diff --git a/textract-pipeline/terraform/modules/aws-textract-pipeline-part1/outputs.tf b/textract-pipeline/terraform/modules/aws-textract-pipeline-part1/outputs.tf index 23aff5d..bde7e81 100644 --- a/textract-pipeline/terraform/modules/aws-textract-pipeline-part1/outputs.tf +++ b/textract-pipeline/terraform/modules/aws-textract-pipeline-part1/outputs.tf @@ -3,6 +3,10 @@ output "insert_record_queue_arn" { description = "Insert record queue arn" value = module.insert_record_sqs.queue_arn } +output "insert_record_queue_url" { + description = "Insert record queue url" + value = module.insert_record_sqs.queue_url +} output "docx_to_pdf_queue_arn" { description = "Docx to PDF queue arn" @@ -34,6 +38,21 @@ output "text_creation_queue_arn" { value = module.text_creation_sqs.queue_arn } +output "prompt_orchestrator_queue_arn" { + description = "Prompt orchestrator queue arn" + value = module.prompt_orchestrator_sqs.queue_arn +} + +output "prompt_processor_queue_arn" { + description = "Prompt processor queue arn" + value = module.prompt_processor_sqs.queue_arn +} + +output "prompt_processor_queue_url" { + description = "Prompt processor queue url" + value = module.prompt_processor_sqs.queue_url +} + # SNS arn's output "textract_notification_sns_arn" { description = "sns topic arn for textract service" diff --git a/textract-pipeline/terraform/modules/aws-textract-pipeline-part1/variables.tf b/textract-pipeline/terraform/modules/aws-textract-pipeline-part1/variables.tf index 0e3501f..28d5e43 100644 --- a/textract-pipeline/terraform/modules/aws-textract-pipeline-part1/variables.tf +++ b/textract-pipeline/terraform/modules/aws-textract-pipeline-part1/variables.tf @@ -49,7 +49,7 @@ variable "insert_record_sqs_queue" { }) default = { name = "insert-record-queue" - delay_seconds = 90 + delay_seconds = 0 max_message_size = 2048 message_retention_seconds = 86400 visibility_timeout_seconds= 901 @@ -68,7 +68,7 @@ variable "docx_to_pdf_sqs_queue" { }) default = { name = "docx-to-pdf-queue" - delay_seconds = 90 + delay_seconds = 0 max_message_size = 2048 message_retention_seconds = 86400 visibility_timeout_seconds= 901 @@ -86,7 +86,7 @@ variable "tiff_to_pdf_sqs_queue" { }) default = { name = "tiff-to-pdf-queue" - delay_seconds = 90 + delay_seconds = 0 max_message_size = 2048 message_retention_seconds = 86400 visibility_timeout_seconds= 901 @@ -104,7 +104,7 @@ variable "pdf_validations_sqs_queue" { }) default = { name = "pdf-validation-queue" - delay_seconds = 90 + delay_seconds = 0 max_message_size = 2048 message_retention_seconds = 86400 visibility_timeout_seconds= 901 @@ -122,7 +122,7 @@ variable "sender_sqs_queue" { }) default = { name = "textract-sender-queue" - delay_seconds = 90 + delay_seconds = 0 max_message_size = 2048 message_retention_seconds = 86400 visibility_timeout_seconds= 901 @@ -140,7 +140,7 @@ variable "receiver_sqs_queue" { }) default = { name = "textract-receiver-queue" - delay_seconds = 90 + delay_seconds = 0 max_message_size = 2048 message_retention_seconds = 86400 visibility_timeout_seconds= 901 @@ -158,7 +158,43 @@ variable "text_creation_sqs_queue" { }) default = { name = "text-creation-queue" - delay_seconds = 90 + delay_seconds = 0 + max_message_size = 2048 + message_retention_seconds = 86400 + visibility_timeout_seconds= 901 + } +} + +# prompt orchestrator queue +variable "prompt_orchestrator_queue" { + type = object({ + name = string + delay_seconds = number + max_message_size = number + message_retention_seconds = number + visibility_timeout_seconds= number + }) + default = { + name = "prompt-orchestrator-queue" + delay_seconds = 0 + max_message_size = 2048 + message_retention_seconds = 86400 + visibility_timeout_seconds= 901 + } +} + +# prompt processor queue +variable "prompt_processor_queue" { + type = object({ + name = string + delay_seconds = number + max_message_size = number + message_retention_seconds = number + visibility_timeout_seconds= number + }) + default = { + name = "prompt-processor-queue" + delay_seconds = 0 max_message_size = 2048 message_retention_seconds = 86400 visibility_timeout_seconds= 901 diff --git a/textract-pipeline/terraform/modules/aws-textract-pipeline-part3/main.tf b/textract-pipeline/terraform/modules/aws-textract-pipeline-part3/main.tf index e84331c..7769a3f 100644 --- a/textract-pipeline/terraform/modules/aws-textract-pipeline-part3/main.tf +++ b/textract-pipeline/terraform/modules/aws-textract-pipeline-part3/main.tf @@ -32,6 +32,7 @@ locals { lambda_layer_prefix = "${local.global_prefix}-lyr" # Lambda names + s3_folder_details_lambda_name = "${local.lambda_prefix}-${var.s3_folder_detail_lambda.name}" batch_creation_lambda_name = "${local.lambda_prefix}-${var.batch_creation_lambda.name}" trigger_pipeline_lambda_name = "${local.lambda_prefix}-${var.trigger_pipeline_lambda.name}" insert_record_lambda_name = "${local.lambda_prefix}-${var.insert_record_lambda.name}" @@ -41,7 +42,11 @@ locals { sender_lambda_name = "${local.lambda_prefix}-${var.textract_sender_lambda.name}" receiver_lambda_name = "${local.lambda_prefix}-${var.textract_receiver_lambda.name}" text_creation_lambda_name = "${local.lambda_prefix}-${var.text_creation_lambda.name}" + prompt_orchestrator_lambda_name = "${local.lambda_prefix}-${var.prompt_orchestrator_lambda.name}" + prompt_processor_lambda_name = "${local.lambda_prefix}-${var.prompt_processor_lambda.name}" + call_bedrock_lambda_name = "${local.lambda_prefix}-${var.call_bedrock_lambda.name}" database_interface_lambda_name = "${local.lambda_prefix}-${var.database_interface_lambda.name}" + database_interface_get_lambda_name = "${local.lambda_prefix}-${var.database_interface_get_lambda.name}" # Lambda Layer names libre_office_lambda_layer_name = "${local.lambda_layer_prefix}-${var.libre_office_lambda_layer.name}" @@ -50,8 +55,9 @@ locals { # API Gateway configuration api_description = "API's used for DoczyAI pipeline" - batch_creation_api_path = "create-batch-new" + batch_creation_api_path = "create-batch" trigger_pipeline_api_path = "trigger-pipeline" + s3_detail_api_path = "s3-detail" stage_name = "dev" # property file path @@ -60,7 +66,8 @@ locals { # common env variables environment_variables = { PROPERTY_FILE_S3_PATH = local.property_file_s3_path - DATABASE_INTERFACE_FUNCTION_NAME = local.database_interface_lambda_name + DATABASE_LOGGING_LAMBDA_FUNCTION_NAME = local.database_interface_lambda_name + DATABASE_LOGGING_GET_LAMBDA_FUNCTION_NAME = local.database_interface_get_lambda_name } # S3 bucket and prefix for lambda's @@ -143,6 +150,33 @@ module "lambda_layer_pypdf2" { local_existing_package = var.pypdf2_lambda_layer.local_existing_package } +# batch creation lambda function +module "s3_folder_detail_lambda_function" { + source = "terraform-aws-modules/lambda/aws" + version = "7.2.2" + + function_name = local.s3_folder_details_lambda_name + description = var.s3_folder_detail_lambda.description + handler = var.s3_folder_detail_lambda.handler + runtime = var.s3_folder_detail_lambda.runtime + timeout = var.s3_folder_detail_lambda.timeout + memory_size = var.s3_folder_detail_lambda.memory_size + ephemeral_storage_size = var.s3_folder_detail_lambda.ephemeral_storage_size + architectures = var.s3_folder_detail_lambda.architectures + + environment_variables = {} + + source_path = var.s3_folder_detail_lambda.source_path + + lambda_role = var.lambda_role_arn + create_role = false + + # vpc_subnet_ids = data.aws_subnets.subnets.ids + # vpc_security_group_ids = data.aws_security_groups.security_groups.ids + + tags = local.common_tags +} + # batch creation lambda function module "batch_creation_lambda_function" { source = "terraform-aws-modules/lambda/aws" @@ -164,8 +198,8 @@ module "batch_creation_lambda_function" { lambda_role = var.lambda_role_arn create_role = false - vpc_subnet_ids = data.aws_subnets.subnets.ids - vpc_security_group_ids = data.aws_security_groups.security_groups.ids + # vpc_subnet_ids = data.aws_subnets.subnets.ids + # vpc_security_group_ids = data.aws_security_groups.security_groups.ids tags = local.common_tags } @@ -184,15 +218,17 @@ module "trigger_pipeline_lambda_function" { ephemeral_storage_size = var.trigger_pipeline_lambda.ephemeral_storage_size architectures = var.trigger_pipeline_lambda.architectures - environment_variables = local.environment_variables + environment_variables = merge(local.environment_variables,{ + INITIATE_DB_SQS_URL = var.insert_record_queue_url + }) source_path = var.trigger_pipeline_lambda.source_path lambda_role = var.lambda_role_arn create_role = false - vpc_subnet_ids = data.aws_subnets.subnets.ids - vpc_security_group_ids = data.aws_security_groups.security_groups.ids + # vpc_subnet_ids = data.aws_subnets.subnets.ids + # vpc_security_group_ids = data.aws_security_groups.security_groups.ids tags = local.common_tags } @@ -230,8 +266,8 @@ module "insert_record_lambda_function" { lambda_role = var.lambda_role_arn create_role = false - vpc_subnet_ids = data.aws_subnets.subnets.ids - vpc_security_group_ids = data.aws_security_groups.security_groups.ids + # vpc_subnet_ids = data.aws_subnets.subnets.ids + # vpc_security_group_ids = data.aws_security_groups.security_groups.ids tags = local.common_tags } @@ -274,8 +310,8 @@ module "docx_to_pdf_lambda_function" { lambda_role = var.lambda_role_arn create_role = false - vpc_subnet_ids = data.aws_subnets.subnets.ids - vpc_security_group_ids = data.aws_security_groups.security_groups.ids + # vpc_subnet_ids = data.aws_subnets.subnets.ids + # vpc_security_group_ids = data.aws_security_groups.security_groups.ids tags = local.common_tags } @@ -313,8 +349,8 @@ module "tiff_to_pdf_lambda_function" { lambda_role = var.lambda_role_arn create_role = false - vpc_subnet_ids = data.aws_subnets.subnets.ids - vpc_security_group_ids = data.aws_security_groups.security_groups.ids + # vpc_subnet_ids = data.aws_subnets.subnets.ids + # vpc_security_group_ids = data.aws_security_groups.security_groups.ids tags = local.common_tags } @@ -356,8 +392,8 @@ module "pdf_validation_lambda_function" { lambda_role = var.lambda_role_arn create_role = false - vpc_subnet_ids = data.aws_subnets.subnets.ids - vpc_security_group_ids = data.aws_security_groups.security_groups.ids + # vpc_subnet_ids = data.aws_subnets.subnets.ids + # vpc_security_group_ids = data.aws_security_groups.security_groups.ids tags = local.common_tags } @@ -398,8 +434,8 @@ module "textract_sender_lambda_function" { lambda_role = var.lambda_role_arn create_role = false - vpc_subnet_ids = data.aws_subnets.subnets.ids - vpc_security_group_ids = data.aws_security_groups.security_groups.ids + # vpc_subnet_ids = data.aws_subnets.subnets.ids + # vpc_security_group_ids = data.aws_security_groups.security_groups.ids tags = local.common_tags } @@ -437,8 +473,8 @@ module "textract_receiver_lambda_function" { lambda_role = var.lambda_role_arn create_role = false - vpc_subnet_ids = data.aws_subnets.subnets.ids - vpc_security_group_ids = data.aws_security_groups.security_groups.ids + # vpc_subnet_ids = data.aws_subnets.subnets.ids + # vpc_security_group_ids = data.aws_security_groups.security_groups.ids tags = local.common_tags } @@ -480,8 +516,125 @@ module "text_creation_lambda_function" { lambda_role = var.lambda_role_arn create_role = false - vpc_subnet_ids = data.aws_subnets.subnets.ids - vpc_security_group_ids = data.aws_security_groups.security_groups.ids + # vpc_subnet_ids = data.aws_subnets.subnets.ids + # vpc_security_group_ids = data.aws_security_groups.security_groups.ids + + tags = local.common_tags +} + +# prompt orchestration lambda function +module "prompt_orchestrator_lambda_function" { + source = "terraform-aws-modules/lambda/aws" + version = "7.2.2" + + function_name = local.prompt_orchestrator_lambda_name + description = var.prompt_orchestrator_lambda.description + handler = var.prompt_orchestrator_lambda.handler + runtime = var.prompt_orchestrator_lambda.runtime + timeout = var.prompt_orchestrator_lambda.timeout + memory_size = var.prompt_orchestrator_lambda.memory_size + ephemeral_storage_size = var.prompt_orchestrator_lambda.ephemeral_storage_size + architectures = var.prompt_orchestrator_lambda.architectures + + layers = [] + + environment_variables = merge(local.environment_variables,{ + PROMPT_PROCESSOR_QUEUE_URL = var.prompt_processor_queue_url + MAX_PROMPT_LENGTH_PER_GROUP = "5000" + }) + + source_path = var.prompt_orchestrator_lambda.source_path + + event_source_mapping = { + sqs = { + event_source_arn = var.prompt_orchestrator_queue_arn + function_response_types = ["ReportBatchItemFailures"] + scaling_config = { + maximum_concurrency = 20 + + } + batch_size = 1 + enabled = false + } + } + + lambda_role = var.lambda_role_arn + create_role = false + + # vpc_subnet_ids = data.aws_subnets.subnets.ids + # vpc_security_group_ids = data.aws_security_groups.security_groups.ids + + tags = local.common_tags +} + +# prompt processor lambda function +module "prompt_processor_lambda_function" { + source = "terraform-aws-modules/lambda/aws" + version = "7.2.2" + + function_name = local.prompt_processor_lambda_name + description = var.prompt_processor_lambda.description + handler = var.prompt_processor_lambda.handler + runtime = var.prompt_processor_lambda.runtime + timeout = var.prompt_processor_lambda.timeout + memory_size = var.prompt_processor_lambda.memory_size + ephemeral_storage_size = var.prompt_processor_lambda.ephemeral_storage_size + architectures = var.prompt_processor_lambda.architectures + + layers = [] + + environment_variables = merge(local.environment_variables,{ + CALL_BEDROCK_LAMBDA_FUNCTION_NAME = local.call_bedrock_lambda_name + }) + + source_path = var.prompt_processor_lambda.source_path + + event_source_mapping = { + sqs = { + event_source_arn = var.prompt_processor_queue_arn + function_response_types = ["ReportBatchItemFailures"] + scaling_config = { + maximum_concurrency = 20 + + } + batch_size = 1 + } + } + + lambda_role = var.lambda_role_arn + create_role = false + + # vpc_subnet_ids = data.aws_subnets.subnets.ids + # vpc_security_group_ids = data.aws_security_groups.security_groups.ids + + tags = local.common_tags +} + +# call-bedrock lambda function +module "call_bedrock_lambda_function" { + source = "terraform-aws-modules/lambda/aws" + version = "7.2.2" + + function_name = local.call_bedrock_lambda_name + description = var.call_bedrock_lambda.description + handler = var.call_bedrock_lambda.handler + runtime = var.call_bedrock_lambda.runtime + timeout = var.call_bedrock_lambda.timeout + memory_size = var.call_bedrock_lambda.memory_size + ephemeral_storage_size = var.call_bedrock_lambda.ephemeral_storage_size + architectures = var.call_bedrock_lambda.architectures + + layers = [] + + environment_variables = local.environment_variables + + source_path = var.call_bedrock_lambda.source_path + + lambda_role = var.lambda_role_arn + create_role = false + + # vpc_subnet_ids = data.aws_subnets.subnets.ids + # vpc_security_group_ids = data.aws_security_groups.security_groups.ids tags = local.common_tags } @@ -509,8 +662,37 @@ module "database_interface_lambda_function" { lambda_role = var.lambda_role_arn create_role = false - vpc_subnet_ids = data.aws_subnets.subnets.ids - vpc_security_group_ids = data.aws_security_groups.security_groups.ids + # vpc_subnet_ids = data.aws_subnets.subnets.ids + # vpc_security_group_ids = data.aws_security_groups.security_groups.ids + + tags = local.common_tags +} + +# database interface lambda function +module "database_interface_get_lambda_function" { + source = "terraform-aws-modules/lambda/aws" + version = "7.2.2" + + function_name = local.database_interface_get_lambda_name + description = var.database_interface_get_lambda.description + handler = var.database_interface_get_lambda.handler + runtime = var.database_interface_get_lambda.runtime + timeout = var.database_interface_get_lambda.timeout + memory_size = var.database_interface_get_lambda.memory_size + ephemeral_storage_size = var.database_interface_get_lambda.ephemeral_storage_size + architectures = var.database_interface_get_lambda.architectures + + environment_variables = merge(local.environment_variables,{ + SECRET_MANAGER_NAME = var.secret_manager_name + }) + + source_path = var.database_interface_get_lambda.source_path + + lambda_role = var.lambda_role_arn + create_role = false + + # vpc_subnet_ids = data.aws_subnets.subnets.ids + # vpc_security_group_ids = data.aws_security_groups.security_groups.ids tags = local.common_tags } @@ -523,7 +705,7 @@ resource "aws_s3_bucket_notification" "bucket_notification" { id = "docx-upload-event" queue_arn = var.docx_to_pdf_queue_arn events = ["s3:ObjectCreated:*"] - filter_prefix = "source-documents-docx/" + filter_prefix = "all-docx/" filter_suffix = ".docx" } @@ -531,7 +713,7 @@ resource "aws_s3_bucket_notification" "bucket_notification" { id = "doc-upload-event" queue_arn = var.docx_to_pdf_queue_arn events = ["s3:ObjectCreated:*"] - filter_prefix = "source-documents-docx/" + filter_prefix = "all-docx/" filter_suffix = ".doc" } @@ -539,7 +721,7 @@ resource "aws_s3_bucket_notification" "bucket_notification" { id = "tiff-upload-event" queue_arn = var.tiff_to_pdf_queue_arn events = ["s3:ObjectCreated:*"] - filter_prefix = "source-documents-tiff/" + filter_prefix = "all-tiff/" filter_suffix = ".tiff" } @@ -566,6 +748,14 @@ resource "aws_s3_bucket_notification" "bucket_notification" { filter_prefix = "textract-output-json/" filter_suffix = ".json" } + + queue { + id = "text-file-upload-event" + queue_arn = var.prompt_orchestrator_queue_arn + events = ["s3:ObjectCreated:*"] + filter_prefix = "contract-text-file/" + filter_suffix = ".txt" + } } # API Gateway resources @@ -580,18 +770,76 @@ resource "aws_api_gateway_rest_api" "project_api" { tags = local.common_tags } +# API Gateway - resource IAM policy +data "aws_iam_policy_document" "api_gateway_policy" { + statement { + effect = "Allow" + principals { + type = "AWS" + identifiers = var.api_access_required_roles_arns + } + actions = ["execute-api:Invoke"] + resources = [aws_api_gateway_rest_api.project_api.execution_arn] + } + + statement { + effect = "Deny" + principals { + type = "AWS" + identifiers = ["*"] + } + actions = ["execute-api:Invoke"] + resources = [aws_api_gateway_rest_api.project_api.execution_arn] + + condition { + test = "StringNotEquals" + variable = "aws:PrincipalArn" + values = var.api_access_required_roles_arns + } + } +} +# # API Gateway - attach resource policy +# resource "aws_api_gateway_rest_api_policy" "project_api_policy" { +# rest_api_id = aws_api_gateway_rest_api.project_api.id +# policy = data.aws_iam_policy_document.api_gateway_policy.json +# } + # API Gateway - Create batch - create resource resource "aws_api_gateway_resource" "create_batch" { rest_api_id = aws_api_gateway_rest_api.project_api.id parent_id = aws_api_gateway_rest_api.project_api.root_resource_id path_part = local.batch_creation_api_path + + depends_on = [ aws_api_gateway_rest_api.project_api ] } # API Gateway - Create batch - create method resource "aws_api_gateway_method" "create_batch_method" { rest_api_id = "${aws_api_gateway_rest_api.project_api.id}" resource_id = "${aws_api_gateway_resource.create_batch.id}" http_method = "POST" - authorization = "NONE" + authorization = "AWS_IAM" + + depends_on = [ + aws_api_gateway_rest_api.project_api, + aws_api_gateway_resource.create_batch + ] +} + +# API Gateway - Create batch - create method response +resource "aws_api_gateway_method_response" "create_batch_method_response_200" { + rest_api_id = aws_api_gateway_rest_api.project_api.id + resource_id = aws_api_gateway_resource.create_batch.id + http_method = aws_api_gateway_method.create_batch_method.http_method + status_code = "200" + response_models = { + "application/json" = "Empty" + } + + depends_on = [ + aws_api_gateway_rest_api.project_api, + aws_api_gateway_resource.create_batch, + aws_api_gateway_method.create_batch_method + ] } # API Gateway - Create batch - integration with lambda @@ -602,7 +850,32 @@ resource "aws_api_gateway_integration" "create_batch_integration" { integration_http_method = "POST" type = "AWS" uri = module.batch_creation_lambda_function.lambda_function_invoke_arn + + depends_on = [ + aws_api_gateway_rest_api.project_api, + aws_api_gateway_resource.create_batch, + aws_api_gateway_method.create_batch_method, + module.batch_creation_lambda_function + ] } + +# API Gateway - Create batch - create integration response +resource "aws_api_gateway_integration_response" "create_batch_integration_response" { + rest_api_id = aws_api_gateway_rest_api.project_api.id + resource_id = aws_api_gateway_resource.create_batch.id + http_method = aws_api_gateway_method.create_batch_method.http_method + status_code = aws_api_gateway_method_response.create_batch_method_response_200.status_code + + depends_on = [ + aws_api_gateway_rest_api.project_api, + aws_api_gateway_resource.create_batch, + aws_api_gateway_method.create_batch_method, + aws_api_gateway_method_response.create_batch_method_response_200, + aws_api_gateway_integration.create_batch_integration, + module.batch_creation_lambda_function + ] +} + # API Gateway - Create batch - add permissions for lambda resource "aws_lambda_permission" "create_batch_lambda_permission" { statement_id = "AllowMyDemoAPIInvoke" @@ -610,19 +883,101 @@ resource "aws_lambda_permission" "create_batch_lambda_permission" { function_name = "${module.batch_creation_lambda_function.lambda_function_name}" principal = "apigateway.amazonaws.com" source_arn = "${aws_api_gateway_rest_api.project_api.execution_arn}/*/${aws_api_gateway_method.create_batch_method.http_method}${aws_api_gateway_resource.create_batch.path}" + + depends_on = [ + aws_api_gateway_rest_api.project_api, + aws_api_gateway_resource.create_batch, + aws_api_gateway_method.create_batch_method, + module.batch_creation_lambda_function + ] } +############# +# API Gateway - s3 folder detail - create resource +resource "aws_api_gateway_resource" "s3_folder_detail" { + rest_api_id = aws_api_gateway_rest_api.project_api.id + parent_id = aws_api_gateway_rest_api.project_api.root_resource_id + path_part = local.s3_detail_api_path + + depends_on = [ aws_api_gateway_rest_api.project_api ] +} +# API Gateway - s3 folder detail - create method +resource "aws_api_gateway_method" "s3_folder_detail_method" { + rest_api_id = "${aws_api_gateway_rest_api.project_api.id}" + resource_id = "${aws_api_gateway_resource.s3_folder_detail.id}" + http_method = "POST" + authorization = "AWS_IAM" +} +# API Gateway - s3 folder detail - create method response +resource "aws_api_gateway_method_response" "s3_folder_detail_method_response_200" { + rest_api_id = aws_api_gateway_rest_api.project_api.id + resource_id = aws_api_gateway_resource.s3_folder_detail.id + http_method = aws_api_gateway_method.s3_folder_detail_method.http_method + status_code = "200" + response_models = { + "application/json" = "Empty" + } +} + +# API Gateway - s3 folder detail - integration with lambda +resource "aws_api_gateway_integration" "s3_folder_detail_integration" { + rest_api_id = aws_api_gateway_rest_api.project_api.id + resource_id = aws_api_gateway_resource.s3_folder_detail.id + http_method = aws_api_gateway_method.s3_folder_detail_method.http_method + integration_http_method = "POST" + type = "AWS" + uri = module.s3_folder_detail_lambda_function.lambda_function_invoke_arn +} + +# API Gateway - s3 folder detail - create integration response +resource "aws_api_gateway_integration_response" "s3_folder_detail_integration_response" { + rest_api_id = aws_api_gateway_rest_api.project_api.id + resource_id = aws_api_gateway_resource.s3_folder_detail.id + http_method = aws_api_gateway_method.s3_folder_detail_method.http_method + status_code = aws_api_gateway_method_response.s3_folder_detail_method_response_200.status_code + + depends_on = [ + aws_api_gateway_rest_api.project_api, + aws_api_gateway_resource.s3_folder_detail, + aws_api_gateway_method.s3_folder_detail_method, + aws_api_gateway_method_response.s3_folder_detail_method_response_200, + aws_api_gateway_integration.s3_folder_detail_integration, + module.s3_folder_detail_lambda_function + ] +} + +# API Gateway - s3 folder detail - add permissions for lambda +resource "aws_lambda_permission" "s3_folder_detail_lambda_permission" { + statement_id = "AllowMyDemoAPIInvoke" + action = "lambda:InvokeFunction" + function_name = "${module.s3_folder_detail_lambda_function.lambda_function_name}" + principal = "apigateway.amazonaws.com" + source_arn = "${aws_api_gateway_rest_api.project_api.execution_arn}/*/${aws_api_gateway_method.s3_folder_detail_method.http_method}${aws_api_gateway_resource.s3_folder_detail.path}" +} +############# # API Gateway - trigger pipeline - create resource resource "aws_api_gateway_resource" "trigger_pipeline" { rest_api_id = aws_api_gateway_rest_api.project_api.id parent_id = aws_api_gateway_rest_api.project_api.root_resource_id path_part = local.trigger_pipeline_api_path + + depends_on = [ aws_api_gateway_rest_api.project_api ] } # API Gateway - trigger pipeline - create method resource "aws_api_gateway_method" "trigger_pipeline_method" { rest_api_id = "${aws_api_gateway_rest_api.project_api.id}" resource_id = "${aws_api_gateway_resource.trigger_pipeline.id}" http_method = "POST" - authorization = "NONE" + authorization = "AWS_IAM" +} +# API Gateway - trigger pipeline - create method response +resource "aws_api_gateway_method_response" "trigger_pipeline_method_response_200" { + rest_api_id = aws_api_gateway_rest_api.project_api.id + resource_id = aws_api_gateway_resource.trigger_pipeline.id + http_method = aws_api_gateway_method.trigger_pipeline_method.http_method + status_code = "200" + response_models = { + "application/json" = "Empty" + } } # API Gateway - trigger pipeline - integration with lambda resource "aws_api_gateway_integration" "trigger_pipeline_integration" { @@ -633,6 +988,22 @@ resource "aws_api_gateway_integration" "trigger_pipeline_integration" { type = "AWS" uri = module.trigger_pipeline_lambda_function.lambda_function_invoke_arn } +# API Gateway - trigger pipeline - create integration response +resource "aws_api_gateway_integration_response" "trigger_pipeline_integration_response" { + rest_api_id = aws_api_gateway_rest_api.project_api.id + resource_id = aws_api_gateway_resource.trigger_pipeline.id + http_method = aws_api_gateway_method.trigger_pipeline_method.http_method + status_code = aws_api_gateway_method_response.trigger_pipeline_method_response_200.status_code + + depends_on = [ + aws_api_gateway_rest_api.project_api, + aws_api_gateway_resource.trigger_pipeline, + aws_api_gateway_method.trigger_pipeline_method, + aws_api_gateway_method_response.trigger_pipeline_method_response_200, + aws_api_gateway_integration.trigger_pipeline_integration, + module.trigger_pipeline_lambda_function + ] +} # API Gateway - trigger pipeline - add permissions for lambda resource "aws_lambda_permission" "trigger_pipeline_lambda_permission" { statement_id = "AllowMyDemoAPIInvoke" @@ -646,6 +1017,7 @@ resource "aws_api_gateway_deployment" "deployment" { depends_on = [ aws_api_gateway_integration.create_batch_integration, aws_api_gateway_integration.trigger_pipeline_integration, + aws_api_gateway_integration.s3_folder_detail_integration ] rest_api_id = "${aws_api_gateway_rest_api.project_api.id}" diff --git a/textract-pipeline/terraform/modules/aws-textract-pipeline-part3/variables.tf b/textract-pipeline/terraform/modules/aws-textract-pipeline-part3/variables.tf index da61de4..3a7c776 100644 --- a/textract-pipeline/terraform/modules/aws-textract-pipeline-part3/variables.tf +++ b/textract-pipeline/terraform/modules/aws-textract-pipeline-part3/variables.tf @@ -48,10 +48,18 @@ variable "devops_s3_bucket" { type = string } +# Required +variable "api_access_required_roles_arns" { + type = list(string) +} + # SQS Required variable "insert_record_queue_arn" { type = string } +variable "insert_record_queue_url" { + type = string +} variable "docx_to_pdf_queue_arn" { type = string } @@ -70,6 +78,15 @@ variable "textract_receiver_queue_arn" { variable "text_creation_queue_arn" { type = string } +variable "prompt_orchestrator_queue_arn" { + type = string +} +variable "prompt_processor_queue_arn" { + type = string +} +variable "prompt_processor_queue_url" { + type = string +} # SNS Required variable "textract_sns_topic_arn" { @@ -142,6 +159,36 @@ variable "pypdf2_lambda_layer" { } } +# s3-folder-details +variable "s3_folder_detail_lambda" { + type = object({ + name = string + description = string + handler = string + runtime = string + timeout = number + memory_size = number + ephemeral_storage_size = number + architectures = list(string) + layers = list(string) + environment_variables = map(string) + source_path = string + }) + default = { + name = "s3-folder-details" + description = "This function gets s3 folder details through API Gateway" + handler = "index.lambda_handler" + runtime = "python3.12" + timeout = 300 + memory_size = 256 + ephemeral_storage_size = 512 + architectures = ["x86_64"] + layers = [] + environment_variables = {} + source_path = "../src/lambda/s3-folder-details" + } +} + # batch-creation variable "batch_creation_lambda" { type = object({ @@ -162,7 +209,7 @@ variable "batch_creation_lambda" { description = "This function creates batch folder" handler = "index.lambda_handler" runtime = "python3.12" - timeout = 60 + timeout = 300 memory_size = 256 ephemeral_storage_size = 512 architectures = ["x86_64"] @@ -191,8 +238,8 @@ variable "trigger_pipeline_lambda" { description = "This function triggers doczy.ai pipeline" handler = "index.lambda_handler" runtime = "python3.12" - timeout = 60 - memory_size = 256 + timeout = 300 + memory_size = 512 ephemeral_storage_size = 512 architectures = ["x86_64"] layers = [] @@ -221,7 +268,7 @@ variable "insert_record_lambda" { description = "This function inserts record in database" handler = "index.lambda_handler" runtime = "python3.12" - timeout = 60 + timeout = 300 memory_size = 256 ephemeral_storage_size = 512 architectures = ["x86_64"] @@ -251,9 +298,9 @@ variable "docx_to_pdf_lambda" { description = "This function converts docx file to pdf file" handler = "index.lambda_handler" runtime = "python3.8" - timeout = 60 - memory_size = 256 - ephemeral_storage_size = 512 + timeout = 900 + memory_size = 5120 + ephemeral_storage_size = 1024 architectures = ["x86_64"] layers = [] environment_variables = {} @@ -281,9 +328,9 @@ variable "tiff_to_pdf_lambda" { description = "This function converts tiff file to pdf file" handler = "index.lambda_handler" runtime = "python3.12" - timeout = 60 - memory_size = 256 - ephemeral_storage_size = 512 + timeout = 900 + memory_size = 1024 + ephemeral_storage_size = 1024 architectures = ["x86_64"] layers = [] environment_variables = {} @@ -308,10 +355,10 @@ variable "pdf_validation_lambda" { }) default = { name = "pdf-validation" - description = "This function valids pdf against textract limitations" + description = "This lambda checks if pdf has less than 3000 pages, size is less than 500mb, if file is not password protected, check if resolution is expected and also if extension is .filepart, it will convert it to .pdf" handler = "index.lambda_handler" runtime = "python3.12" - timeout = 60 + timeout = 300 memory_size = 256 ephemeral_storage_size = 512 architectures = ["x86_64"] @@ -341,8 +388,8 @@ variable "textract_sender_lambda" { description = "This function sends async request to TEXTRACT service" handler = "index.lambda_handler" runtime = "python3.12" - timeout = 60 - memory_size = 256 + timeout = 600 + memory_size = 512 ephemeral_storage_size = 512 architectures = ["x86_64"] layers = [] @@ -371,8 +418,8 @@ variable "textract_receiver_lambda" { description = "This function receives async request from TEXTRACT service" handler = "index.lambda_handler" runtime = "python3.12" - timeout = 60 - memory_size = 256 + timeout = 600 + memory_size = 1024 ephemeral_storage_size = 512 architectures = ["x86_64"] layers = [] @@ -401,8 +448,8 @@ variable "text_creation_lambda" { description = "This function creates text from Textract JSON" handler = "index.lambda_handler" runtime = "python3.12" - timeout = 60 - memory_size = 256 + timeout = 900 + memory_size = 512 ephemeral_storage_size = 512 architectures = ["x86_64"] layers = [] @@ -411,6 +458,96 @@ variable "text_creation_lambda" { } } +# prompt-orchestrator lambda +variable "prompt_orchestrator_lambda" { + type = object({ + name = string + description = string + handler = string + runtime = string + timeout = number + memory_size = number + ephemeral_storage_size = number + architectures = list(string) + layers = list(string) + environment_variables = map(string) + source_path = string + }) + default = { + name = "prompt-orchestrator" + description = "This Lambda functions breaks prompts into chunks and send messages into SQS" + handler = "index.lambda_handler" + runtime = "python3.12" + timeout = 600 + memory_size = 256 + ephemeral_storage_size = 512 + architectures = ["x86_64"] + layers = [] + environment_variables = {} + source_path = "../src/lambda/prompt-orchestrator" + } +} + +# prompt-processor lambda +variable "prompt_processor_lambda" { + type = object({ + name = string + description = string + handler = string + runtime = string + timeout = number + memory_size = number + ephemeral_storage_size = number + architectures = list(string) + layers = list(string) + environment_variables = map(string) + source_path = string + }) + default = { + name = "prompt-processor" + description = "This Lambda functions sends request to LLM models hosted Bedrock and save response into Snowflake database" + handler = "index.lambda_handler" + runtime = "python3.12" + timeout = 900 + memory_size = 1024 + ephemeral_storage_size = 512 + architectures = ["x86_64"] + layers = [] + environment_variables = {} + source_path = "../src/lambda/prompt-processor" + } +} + +# call-bedrock lambda +variable "call_bedrock_lambda" { + type = object({ + name = string + description = string + handler = string + runtime = string + timeout = number + memory_size = number + ephemeral_storage_size = number + architectures = list(string) + layers = list(string) + environment_variables = map(string) + source_path = string + }) + default = { + name = "call-bedrock" + description = "This Lambda functions sends request to LLM models hosted Bedrock service" + handler = "index.lambda_handler" + runtime = "python3.12" + timeout = 600 + memory_size = 512 + ephemeral_storage_size = 512 + architectures = ["x86_64"] + layers = [] + environment_variables = {} + source_path = "../src/lambda/call-bedrock" + } +} + # database-interface lambda variable "database_interface_lambda" { type = object({ @@ -431,7 +568,7 @@ variable "database_interface_lambda" { description = "This function acts as an interface to communicate with database" handler = "index.lambda_handler" runtime = "python3.12" - timeout = 60 + timeout = 300 memory_size = 256 ephemeral_storage_size = 512 architectures = ["x86_64"] @@ -439,4 +576,34 @@ variable "database_interface_lambda" { environment_variables = {} source_path = "../src/lambda/database-interface" } +} + +# database-interface lambda +variable "database_interface_get_lambda" { + type = object({ + name = string + description = string + handler = string + runtime = string + timeout = number + memory_size = number + ephemeral_storage_size = number + architectures = list(string) + layers = list(string) + environment_variables = map(string) + source_path = string + }) + default = { + name = "database-interface-get" + description = "This function acts as an interface to get records from Snowflake database" + handler = "index.lambda_handler" + runtime = "python3.12" + timeout = 300 + memory_size = 256 + ephemeral_storage_size = 512 + architectures = ["x86_64"] + layers = [] + environment_variables = {} + source_path = "../src/lambda/database-interface-get" + } } \ No newline at end of file diff --git a/textract-pipeline/terraform/modules/aws-textract-pipeline/main.tf b/textract-pipeline/terraform/modules/aws-textract-pipeline/main.tf index 085646b..0925392 100644 --- a/textract-pipeline/terraform/modules/aws-textract-pipeline/main.tf +++ b/textract-pipeline/terraform/modules/aws-textract-pipeline/main.tf @@ -56,17 +56,21 @@ module "textract_pipeline_part3" { vpc_id = var.vpc_id secret_manager_name = var.secret_manager_name devops_s3_bucket = var.devops_s3_bucket + api_access_required_roles_arns = var.api_access_required_roles_arns s3_bucket_name = module.textract_pipeline_part1.textract_s3_bucket_name insert_record_queue_arn = module.textract_pipeline_part1.insert_record_queue_arn + insert_record_queue_url = module.textract_pipeline_part1.insert_record_queue_url docx_to_pdf_queue_arn = module.textract_pipeline_part1.docx_to_pdf_queue_arn tiff_to_pdf_queue_arn = module.textract_pipeline_part1.tiff_to_pdf_queue_arn pdf_validations_queue_arn = module.textract_pipeline_part1.pdf_validation_queue_arn textract_sender_queue_arn = module.textract_pipeline_part1.sender_queue_arn textract_receiver_queue_arn = module.textract_pipeline_part1.receiver_queue_arn text_creation_queue_arn = module.textract_pipeline_part1.text_creation_queue_arn - + prompt_orchestrator_queue_arn = module.textract_pipeline_part1.prompt_orchestrator_queue_arn + prompt_processor_queue_arn = module.textract_pipeline_part1.prompt_processor_queue_arn + prompt_processor_queue_url = module.textract_pipeline_part1.prompt_processor_queue_url textract_role_arn = var.textract_role_arn textract_sns_topic_arn = module.textract_pipeline_part1.textract_notification_sns_arn diff --git a/textract-pipeline/terraform/modules/aws-textract-pipeline/variables.tf b/textract-pipeline/terraform/modules/aws-textract-pipeline/variables.tf index a12b3c5..ea91c51 100644 --- a/textract-pipeline/terraform/modules/aws-textract-pipeline/variables.tf +++ b/textract-pipeline/terraform/modules/aws-textract-pipeline/variables.tf @@ -33,6 +33,11 @@ variable "devops_s3_bucket" { type = string } +# Required +variable "api_access_required_roles_arns" { + type = list(string) +} + # Required variable "lambda_role_arn" { type = string diff --git a/textract-pipeline/terraform/variables.tf b/textract-pipeline/terraform/variables.tf index e793036..bf55392 100644 --- a/textract-pipeline/terraform/variables.tf +++ b/textract-pipeline/terraform/variables.tf @@ -1,23 +1,23 @@ -# required -variable "secret_key" { - type = string -} - -# required -variable "access_key" { - type = string -} - +# # required +# variable "secret_key" { +# type = string +# } # # required -# variable "aws_profile" { +# variable "access_key" { # type = string -# default = "doczyai" # } + + +# required +variable "aws_profile" { + type = string + default = "doczyai" +} # required variable "aws_region" { type = string - # default = "us-east-2" + default = "us-east-2" } # required variable "project_name" { @@ -27,7 +27,7 @@ variable "project_name" { # required variable "environment" { type = string - # default = "dev" + default = "dev" }