jacob-mathison/doczyai-pipelines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Updated lambda function code for textract & changed config to address TF drift

Browse Source
This commit is contained in:
Umang Mistry
2024-09-27 10:38:49 -05:00
parent 5d566a098a
commit 1370553175
5 changed files with 367 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files
textract-pipeline/src/lambda/text-creation/index.py
+133 -55
View File
@@ -9,6 +9,7 @@ from urllib.parse import unquote_plus
from PyPDF2 import PdfReader, PdfWriter
import pandas as pd
import datetime
from PyPDF2.errors import PdfReadError
# Configure logging
logger = logging.getLogger()
@@ -158,6 +159,7 @@ def lambda_handler(event, context):
# Extract text from PAGE block type and concatenate with page numbers
for block_id, block in enumerate(textract_json['Blocks']):
if block['BlockType'] == 'PAGE':
logger.info(f"Processing page {page_number}... Block ID: {block_id}, block type: {block['BlockType']}")
extracted_text += f'\n\nStart of Page No. = {page_number}\n'
child_ids = get_child_ids(textract_json, block_id)
extracted_text += concatenate_page_text_from_child_ids(child_ids, block_id_map,pdf_file_name_with_path,page_number,S3_BUCKET_NAME,final_number_of_signature,temp_file_path, tags)
@@ -206,9 +208,11 @@ def get_child_ids(textract_json, block_id):
# Check if the block_id is within the valid range
block = textract_json['Blocks'][block_id]
#logger.info(f"Processing block: {block}")
# Check if the block has 'Relationships' key
if 'Relationships' in block:
logger.info(f"Block has relationships: {block['Relationships']}")
for relationship in block['Relationships']:
if relationship['Type'] == 'CHILD':
child_ids.extend(relationship['Ids'])
@@ -220,6 +224,7 @@ def concatenate_text_from_child_ids(child_ids, block_id_map,block_type):
concatenated_text = ''
for child_id in child_ids:
logger.debug(f"Processing child ID: {child_id}")
# Use the block_id_map to access the block using the ID directly
block = block_id_map.get(child_id)
@@ -229,7 +234,7 @@ def concatenate_text_from_child_ids(child_ids, block_id_map,block_type):
concatenated_text += block['Text']+'\n'
else :
concatenated_text += block['Text'] + ' ' + str(block['Page'])
logger.debug(f"Concatenated text: {concatenated_text}")
return concatenated_text.strip()
@@ -240,6 +245,8 @@ def concatenate_page_text_from_child_ids(child_ids, block_id_map,pdf_file_name_w
table_found = False # Flag to indicate if a table block is found
for child_id in child_ids:
# Adding extensive logging to ensure the hierarchy of blocks is correct and if we are catching all elements of the document
logger.debug(f"Processing child ID: {child_id}, block type: {block_id_map[child_id]['BlockType']}")
# Use the block_id_map to access the block using the ID directly
block = block_id_map.get(child_id)
@@ -306,12 +313,20 @@ def strings_contained(substrings, main_string):
return sum(1 for substring in substrings if substring in main_string)
# Function to extract a specific page from a PDF file and send it to Textract
def extract_page_and_send_to_textract(page_number, s3_bucket, s3_key,temp_file_path,tags,feature_types=None):
def extract_page_and_send_to_textract(page_number, s3_bucket, s3_key, temp_file_path, tags, feature_types=None):
try:
# Read the PDF file and extract the specified page
# Read the PDF file and check if it's encrypted
with open(temp_file_path, 'rb') as file:
pdf_reader = PdfReader(file)
if pdf_reader.is_encrypted:
logging.info("File is encrypted. Trying to decrypt.")
try:
pdf_reader.decrypt("")
logging.info("File decrypted successfully.")
except Exception as e:
logging.error(f"Failed to decrypt file: {e}")
return None
if page_number < 0 or page_number >= len(pdf_reader.pages):
raise ValueError(f"Invalid page number: {page_number}")
page = pdf_reader.pages[page_number]
@@ -328,26 +343,23 @@ def extract_page_and_send_to_textract(page_number, s3_bucket, s3_key,temp_file_p
# Call AWS Textract synchronous API to analyze the extracted page
with open(temp_page_file_path, 'rb') as page_file:
textract_response = textract_client.analyze_document(Document={'Bytes': page_file.read()},
FeatureTypes=feature_types)
FeatureTypes=feature_types)
logging.info(f"Page {page_number} analyzed by Textract.")
# Extract sub folder with filename
sub_folder_with_filename = s3_key.replace(PROCESSED_LOCATION,"")
sub_folder_with_filename = s3_key.replace(PROCESSED_LOCATION, "")
# Store the JSON response in S3
page_json_key = sub_folder_with_filename.replace('.pdf', '') + f"_page_{page_number + 1}.json"
logging.info(f"TEXTRACT_SIGNATURE_JSON: {TEXTRACT_SIGNATURE_JSON} TEXTRACT_TABLE_FORM_JSON: {TEXTRACT_TABLE_FORM_JSON}")
if any('SIGNATURES' in feature for feature in feature_types):
page_json_key = TEXTRACT_SIGNATURE_JSON+page_json_key
page_json_key = TEXTRACT_SIGNATURE_JSON + page_json_key
else:
page_json_key = TEXTRACT_TABLE_FORM_JSON+page_json_key
page_json_key = TEXTRACT_TABLE_FORM_JSON + page_json_key
#page_json_key = f"page_{page_number}_textract.json"
s3_client.put_object(Bucket=s3_bucket, Key=page_json_key,
Body=json.dumps(textract_response), ContentType='application/json',Tagging=tags)
Body=json.dumps(textract_response), ContentType='application/json', Tagging=tags)
logging.info(f"Textract JSON response for page {page_number} stored in S3 with key: {page_json_key}")
# Remove the temporary files
@@ -356,12 +368,14 @@ def extract_page_and_send_to_textract(page_number, s3_bucket, s3_key,temp_file_p
logging.info(f"Page {page_number} extraction and Textract processing completed successfully.")
return textract_response
except PdfReadError as e:
logging.error(f"PdfReadError: {str(e)}. Unable to process page {page_number}.")
return None
except Exception as e:
logging.error(f"Error extracting page {page_number} and sending to Textract: {e}")
return None
def get_child_table_ids(block):
child_ids = []
@@ -428,6 +442,7 @@ def get_lines(json_data,child_ids_list):
def extract_text(data):
text_lines = ""
for block_id, block_data in data.items():
# TODO: UNDO this as the next line is commented for HN fac table troubleshoot
if block_id not in child_ids_list:
if block_data.get("BlockType") == "LINE":
text_lines += (
@@ -438,34 +453,65 @@ def get_lines(json_data,child_ids_list):
return extract_text(lines)
def get_table_text(json_data,child_ids_list):
def get_table_text(json_data, child_ids_list):
try:
blocks = json_data["Blocks"]
# Map the blocks by type
tables = map_blocks(blocks, "TABLE")
layout_tables = map_blocks(blocks, "LAYOUT_TABLE")
if len(layout_tables) != 0:
tables.update(layout_tables)
cells = map_blocks(blocks, "CELL")
merged_cells = map_blocks(blocks, "MERGED_CELL")
words = map_blocks(blocks, "WORD")
lines = map_blocks(blocks, "LINE")
selections = map_blocks(blocks, "SELECTION_ELEMENT")
def get_children_ids(block):
# Helper function to get child IDs from a block
def get_children_ids(block, relationship_type="CHILD"):
for rels in block.get("Relationships", []):
if rels["Type"] == "CHILD":
if rels["Type"] == relationship_type:
yield from rels["Ids"]
def get_text_from_cell_blocks(cells, cell_ids):
extracted_text = []
for cell_id in cell_ids:
cell = cells.get(cell_id, {})
# Check if the cell has the 'Text' key
if 'Text' in cell:
extracted_text.append(cell['Text'])
else:
# If 'Text' key is not present, look for child words
if 'Relationships' in cell:
for rel in cell['Relationships']:
if rel['Type'] == 'CHILD':
for child_id in rel['Ids']:
if child_id in words:
extracted_text.append(words[child_id]["Text"])
return ' '.join(extracted_text)
def get_table_title_children_ids(block):
logger.info(f"Finding table title: {block}")
table_title_id = ""
for rels in block.get("Relationships", []):
if rels["Type"] == "TABLE_TITLE":
logger.info(f"Table title found: {rels}")
table_title_id = rels["Ids"][0]
def get_table_footer_children_ids(block):
logger.info(f"Finding table footer: {block}")
table_title_id = ""
for rels in block.get("Relationships", []):
if rels["Type"] == "TABLE_FOOTER":
logger.info(f"Table FOOTER found: {rels}")
table_title_id = rels["Ids"][0]
if table_title_id != "":
for block2 in json_data.get("Blocks", []):
if (
block2["BlockType"] == "TABLE_TITLE"
and block2["Id"] == table_title_id
):
if block2["BlockType"] == "TABLE_TITLE" and block2["Id"] == table_title_id:
if block2["Relationships"][0]["Type"] == "CHILD":
child_ids_list.extend(block2["Relationships"][0]["Ids"])
@@ -475,62 +521,94 @@ def get_table_text(json_data,child_ids_list):
if block2["BlockType"] == "WORD":
if block2["Id"] == childe_id:
full_child_text = full_child_text + " " + block2["Text"]
if full_child_text=="":
if full_child_text == "":
return None
return full_child_text
def ensure_content_size(content, required_rows, required_cols):
# Expand rows if needed
while len(content) < required_rows:
content.append([None] * len(content[0])) # Add new rows with the same number of columns
# Expand columns if needed
for row in content:
while len(row) < required_cols:
row.append(None)
dataframe_dicts = []
for index, table in enumerate(tables.values()):
# Get all the cells belonging to this table
table_cells = [cells[cell_id] for cell_id in get_children_ids(table) if cell_id in cells]
# Determine all the cells that belong to this table
table_cells = [cells[cell_id] for cell_id in get_children_ids(table)]
# Get all the merged cells belonging to this table
table_merged_cells = [merged_cells[cell_id] for cell_id in get_children_ids(table, "MERGED_CELL") if
cell_id in merged_cells]
# Determine the table's number of rows and columns
n_rows = max(cell["RowIndex"] for cell in table_cells)
n_cols = max(cell["ColumnIndex"] for cell in table_cells)
if table_cells: # Check if table_cells is not empty
n_rows = max(cell["RowIndex"] for cell in table_cells)
else:
n_rows = 0
if table_cells: # Check if table_cells is not empty
n_cols = max(cell["ColumnIndex"] for cell in table_cells)
else:
n_cols = 0
content = [[None for _ in range(n_cols)] for _ in range(n_rows)]
# Fill in each cell
# Fill in each regular cell
for cell in table_cells:
cell_contents = [
(
words[child_id]["Text"]
if child_id in words
else selections[child_id]["SelectionStatus"]
)
for child_id in get_children_ids(cell)
]
i = cell["RowIndex"] - 1
j = cell["ColumnIndex"] - 1
content[i][j] = " ".join(cell_contents)
# Extract text from the cell using WORD blocks
cell_contents = []
if cell.get("Relationships"):
for child_id in get_children_ids(cell):
if child_id in words:
cell_contents.append(words[child_id]["Text"])
if content:
if i >= len(content):
content.append([None] * n_cols)
# We assume that the first row corresponds to the column names
dataframe = pd.DataFrame(content[1:], columns=content[0])
# Dynamically adjust columns
if j >= len(content[i]):
content[i].extend([None] * (j - len(content[i]) + 1))
content[i][j] = ' '.join(cell_contents)
else:
content = ''
dataframe_dict = dataframe.to_dict(orient="list")
# dataframe_dicts += str(dataframe_dict)
# Handle merged cells
for merged_cell in table_merged_cells:
row_start = merged_cell["RowIndex"] - 1
col_start = merged_cell["ColumnIndex"] - 1
row_span = merged_cell.get("RowSpan", 1)
col_span = merged_cell.get("ColumnSpan", 1)
# Concatenate text from all words associated with the merged cell
merged_text = []
if 'Relationships' in merged_cell:
for rel in merged_cell['Relationships']:
if rel['Type'] == 'CHILD':
merged_text = get_text_from_cell_blocks(cells, rel['Ids'])
for row_offset in range(row_span):
for col_offset in range(col_span):
content[row_start + row_offset][col_start + col_offset] = merged_text
logger.info(f"Table content = {content}")
dataframe_dicts.append("-------Table Start--------")
dataframe_dicts.append(get_table_title_children_ids(table))
dataframe_dicts.append(dataframe_dict)
dataframe_dicts.append(content)
dataframe_dicts.append(get_table_footer_children_ids(table))
dataframe_dicts.append("-------Table End--------")
dataframe_dicts_str = "\n".join(
str(dataframe_dict) for dataframe_dict in dataframe_dicts
)
dataframe_dicts_str = "\n".join(str(dataframe_dict) for dataframe_dict in dataframe_dicts)
return dataframe_dicts_str
except Exception as e:
logger.error(f"Error in get_table_text: {e}")
print(traceback.format_exc())
return None
textract-pipeline/src/lambda/textract-receiver/index.py
+102 -95
View File
@@ -2,6 +2,7 @@ import json
import logging
import boto3
import os
import time
from configparser import ConfigParser
from botocore.exceptions import ClientError
from urllib.parse import urlencode
@@ -16,9 +17,9 @@ s3_client = boto3.client('s3')
textract_client = boto3.client('textract')
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')
@@ -34,20 +35,39 @@ def load_config_from_s3(bucket_name, file_key):
return config_dict
# Function to construct JSON path from S3 object path
def generate_output_json_path(src_folder,dest_folder, s3_object):
def generate_output_json_path(src_folder, dest_folder, s3_object):
# split string to remove source folder path
txt = s3_object.split(src_folder)
if len(txt) == 2:
txt = txt[1]
else:
txt = txt[0]
# change file extension - remove .pdf and add .json
# change file extension - remove .pdf and add .json
if txt.lower().endswith(".pdf"):
return dest_folder+txt[0:-4]+".json"
return dest_folder + txt[0:-4] + ".json"
# Function to get Textract document analysis
# Exponential backoff utility function
def exponential_backoff_retry(func, *args, retries=5, initial_delay=1, max_delay=32, **kwargs):
delay = initial_delay
for attempt in range(retries):
try:
return func(*args, **kwargs)
except ClientError as e:
error_code = e.response['Error']['Code']
if error_code == 'ProvisionedThroughputExceededException' and attempt < retries - 1:
logger.warning(
f"ProvisionedThroughputExceededException on attempt {attempt + 1}/{retries}, retrying in {delay} seconds...")
time.sleep(delay)
delay = min(delay * 2, max_delay)
else:
logger.error(f"Exceeded maximum retries. Function {func.__name__} failed with error: {e}")
raise
# Function to get Textract document detection
def get_textract_document_detection(job_id, textract_client):
# Initialize an empty list to store blocks
all_blocks = []
@@ -58,30 +78,24 @@ def get_textract_document_detection(job_id, textract_client):
try:
while True:
if flag:
# Call the Textract API to get document analysis
response = textract_client.get_document_text_detection(
JobId=job_id
)
response = exponential_backoff_retry(textract_client.get_document_text_detection, JobId=job_id)
flag = False
else:
# Call the Textract API to get document analysis
response = textract_client.get_document_text_detection(
JobId=job_id,
NextToken=next_token
)
response = exponential_backoff_retry(textract_client.get_document_text_detection, JobId=job_id,
NextToken=next_token)
job_status = response["JobStatus"]
logger.info("Job %s status is %s.", job_id, job_status)
# Merge the blocks from the current response
all_blocks.extend(response.get('Blocks', []))
# Check if there are more blocks to retrieve
next_token = response.get('NextToken')
if not next_token:
logger.info("No more Textract response to retrieve")
break
except ClientError:
logger.exception("Couldn't get data for job %s.", job_id)
raise
@@ -90,14 +104,15 @@ def get_textract_document_detection(job_id, textract_client):
last_response = response.copy()
last_response.pop('Blocks', None)
last_response.pop('ResponseMetadata', None)
#logger.info("Removed 'ResponseMetadata' key")
# logger.info("Removed 'ResponseMetadata' key")
# Merge with {'Blocks': all_blocks}
final_response = {'Blocks': all_blocks}
final_response.update(last_response)
logger.info("Final Textract response is contructed")
logger.info("Final Textract response is constructed")
return final_response
# Function to get Textract document analysis
def get_textract_document_analysis(job_id, textract_client):
# Initialize an empty list to store blocks
@@ -109,30 +124,24 @@ def get_textract_document_analysis(job_id, textract_client):
try:
while True:
if flag:
# Call the Textract API to get document analysis
response = textract_client.get_document_analysis(
JobId=job_id
)
response = exponential_backoff_retry(textract_client.get_document_analysis, JobId=job_id)
flag = False
else:
# Call the Textract API to get document analysis
response = textract_client.get_document_analysis(
JobId=job_id,
NextToken=next_token
)
response = exponential_backoff_retry(textract_client.get_document_analysis, JobId=job_id,
NextToken=next_token)
job_status = response["JobStatus"]
logger.info("Job %s status is %s.", job_id, job_status)
# Merge the blocks from the current response
all_blocks.extend(response.get('Blocks', []))
# Check if there are more blocks to retrieve
next_token = response.get('NextToken')
if not next_token:
logger.info("No more Textract response to retrieve")
break
except ClientError:
logger.exception("Couldn't get data for job %s.", job_id)
raise
@@ -141,19 +150,20 @@ def get_textract_document_analysis(job_id, textract_client):
last_response = response.copy()
last_response.pop('Blocks', None)
last_response.pop('ResponseMetadata', None)
#logger.info("Removed 'ResponseMetadata' key")
# logger.info("Removed 'ResponseMetadata' key")
# Merge with {'Blocks': all_blocks}
final_response = {'Blocks': all_blocks}
final_response.update(last_response)
logger.info("Final Textract response is contructed")
logger.info("Final Textract response is constructed")
return final_response
# Function to save Textract response to S3
def upload_response_to_s3(response, bucket_name, object_key, s3_client, tags):
# Convert the response to JSON
response_json = json.dumps(response)
try:
# Upload the JSON response to S3
s3_client.put_object(
@@ -165,13 +175,16 @@ def upload_response_to_s3(response, bucket_name, object_key, s3_client, 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)}")
logger.error(f"Error saving response to S3 - {object_key}: {str(e)}")
# Function to move file within S3
# Function to move file within S3
def move_file_within_s3(source_bucket, source_path, destination_path):
try:
# Copy the file to the destination folder
s3_client.copy_object(Bucket=source_bucket, CopySource={'Bucket': source_bucket, 'Key': source_path}, Key=destination_path)
s3_client.copy_object(Bucket=source_bucket, CopySource={'Bucket': source_bucket, 'Key': source_path},
Key=destination_path)
# Delete the file from the source folder
s3_client.delete_object(Bucket=source_bucket, Key=source_path)
@@ -180,9 +193,9 @@ def move_file_within_s3(source_bucket, source_path, destination_path):
except Exception as e:
logger.error(f"Error moving file: {e}")
# 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(
@@ -200,24 +213,24 @@ def get_s3_object_tags(bucket_name, object_key):
logger.exception(f"Error: {e}")
print(f"Error: {e}")
return None
# AWS Lambda handler function
def lambda_handler(event, context):
#logger.info('## ENVIRONMENT VARIABLES\r' + str(os.environ))
# logger.info('## ENVIRONMENT VARIABLES\r' + str(os.environ))
print(event)
# Read environment variables
property_file_path = os.environ.get('PROPERTY_FILE_S3_PATH', '')
global DATABASE_LOGGING_LAMBDA_FUNCTION_NAME
DATABASE_LOGGING_LAMBDA_FUNCTION_NAME = os.environ.get('DATABASE_LOGGING_LAMBDA_FUNCTION_NAME', '')
batch_id = ""
logger.info('DATABASE_LOGGING_LAMBDA_FUNCTION_NAME: ' + str(DATABASE_LOGGING_LAMBDA_FUNCTION_NAME))
# Validate environment variables
file_path_array = property_file_path.split("/")
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:])
@@ -228,8 +241,7 @@ def lambda_handler(event, context):
# Load config file
config_dict = load_config_from_s3(S3_BUCKET_NAME, CONFIG_FILE_PATH)
#logger.info('## CONFIG DICTIONARY\r' + str(config_dict))
# logger.info('## CONFIG DICTIONARY\r' + str(config_dict))
# Process each message from the SQS event
for record in event['Records']:
@@ -238,58 +250,53 @@ def lambda_handler(event, context):
message_body = json.loads(record_body['Message'])
logger.info('MESSAGE_BODY: ' + str(message_body))
try:
# Extract relevant information from the message body
job_id = message_body.get('JobId')
document_location = message_body.get('DocumentLocation')
s3_object_name = document_location.get('S3ObjectName')
# Read tags from stagging file
tags_dict = get_s3_object_tags(S3_BUCKET_NAME,s3_object_name)
# Read tags from staging file
tags_dict = get_s3_object_tags(S3_BUCKET_NAME, s3_object_name)
# check if key exist in tags_dict
if "BatchId" in tags_dict:
batch_id = tags_dict['BatchId']
else:
logger.error(f"BatchId not found in file tag")
return
STAGING_LOCATION = config_dict['FOLDER_LOCATIONS']['STAGING_LOCATION'].format(batch_id)
OUTPUT_LOCATION = config_dict['FOLDER_LOCATIONS']['OUTPUT_LOCATION'].format(batch_id)
PROCESSED_LOCATION = config_dict['FOLDER_LOCATIONS']['PROCESSED_LOCATION'].format(batch_id)
UNPROCESSED_LOCATION = config_dict['FOLDER_LOCATIONS']['UNPROCESSED_LOCATION'].format(batch_id)
PROCESS_TYPE = str(config_dict['OTHERS']['PROCESS_TYPE']).upper()
# encode to URL Query parameter
tags = urlencode(tags_dict)
# Check if the status is "SUCCEEDED"
if message_body.get('Status') == 'SUCCEEDED':
document = {}
if PROCESS_TYPE == "ANALYSIS":
# Call the function to get document analysis using Textract
document = get_textract_document_analysis(job_id, textract_client)
elif PROCESS_TYPE == "DETECTION":
# Call the function to get document analysis using Textract
document = get_textract_document_detection(job_id, textract_client)
# Save the document analysis response to S3
s3_object_key = generate_output_json_path(STAGING_LOCATION,OUTPUT_LOCATION, s3_object_name)
upload_response_to_s3(document, S3_BUCKET_NAME, s3_object_key, s3_client,tags)
s3_object_key = generate_output_json_path(STAGING_LOCATION, OUTPUT_LOCATION, s3_object_name)
upload_response_to_s3(document, S3_BUCKET_NAME, s3_object_key, s3_client, tags)
# Construct the destination paths
destination_path = PROCESSED_LOCATION + s3_object_name.replace(STAGING_LOCATION,"")
destination_path = PROCESSED_LOCATION + s3_object_name.replace(STAGING_LOCATION, "")
# Move file to processed folder
move_file_within_s3(S3_BUCKET_NAME, s3_object_name, destination_path)
document_id = get_filename_from_path(s3_object_name)
document_id = os.path.splitext(document_id)[0]
generate_document_logs_input(document_id,message_body.get('Status'))
success_message = 'Processed file '+ str(s3_object_key)
generate_document_logs_input(document_id, message_body.get('Status'))
success_message = 'Processed file ' + str(s3_object_key)
logger.info(success_message)
return {
'statusCode': 200,
@@ -298,66 +305,66 @@ def lambda_handler(event, context):
else:
error_message = f"Skipping message with JobId {message_body.get('JobId')} as Status is not 'SUCCEEDED'"
logger.error(error_message)
# Construct the destination paths
destination_path = UNPROCESSED_LOCATION + s3_object_name.replace(STAGING_LOCATION,"")
destination_path = UNPROCESSED_LOCATION + s3_object_name.replace(STAGING_LOCATION, "")
# Move file to unprocessed folder
move_file_within_s3(S3_BUCKET_NAME, s3_object_name, destination_path)
document_id = get_filename_from_path(s3_object_name)
document_id = os.path.splitext(document_id)[0]
generate_document_logs_input(document_id,message_body.get('Status'))
generate_document_logs_input(document_id, message_body.get('Status'))
return {
'statusCode': 500,
'body': error_message
}
except Exception as e:
error_message = f"Error processing message with JobId {message_body.get('JobId')}: {str(e)}"
logger.error(error_message)
logger.error(error_message)
document_id = get_filename_from_path(s3_object_name)
document_id = os.path.splitext(document_id)[0]
generate_document_logs_input(document_id,message_body.get('Status'))
generate_document_logs_input(document_id, message_body.get('Status'))
return {
'statusCode': 500,
'body': error_message
}
'statusCode': 500,
'body': error_message
}
else:
error_message = 'Incorrect value for ENVIRONMENT VARIABLES: PROPERTY_FILE_S3_PATH\r' + str(property_file_path)
logger.error(error_message)
return {
'statusCode': 500,
'body': error_message
}
'statusCode': 500,
'body': error_message
}
def get_filename_from_path(full_path):
return os.path.basename(full_path)
def generate_document_logs_input(document_id,textract_status):
current_time = datetime.datetime.now().isoformat()
data = {
def generate_document_logs_input(document_id, textract_status):
current_time = datetime.datetime.now().isoformat()
data = {
"operation": "update",
"table": "DOCUMENT_LOGS",
"data": {
"DOCUMENT_ID": document_id,
"STAGE" : "RECEIVED_FROM_TEXTRACT",
"STAGE": "RECEIVED_FROM_TEXTRACT",
"TEXTRACT_STATUS": textract_status,
"MODIFIED_TIME": current_time,
"MODIFIED_BY": "TEXTRACT_RECEIVER"
}
}
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
}
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
textract-pipeline/src/lambda/textract-sender/index.py
+129 -134
View File
@@ -25,7 +25,6 @@ def generate_unix_timestamp():
# 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')
@@ -75,90 +74,92 @@ def get_pdf_files_list_from_s3(source_bucket, source_folder):
return file_list
def start_textract_detection_job( bucket_name,
document_file_name,
sns_topic_arn,
sns_role_arn,
job_tag,):
try:
# Define the parameters for the start_document_analysis API
start_document_detection_params = {
'DocumentLocation': {
'S3Object': {
'Bucket': bucket_name,
'Name': document_file_name
}
},
'ClientRequestToken': 'unique-token-'+str(generate_unix_timestamp()), # Use a unique token for each request
'JobTag': job_tag, # Use a tag to identify your job
'NotificationChannel': {
'SNSTopicArn': sns_topic_arn,
'RoleArn': sns_role_arn # Role to allow Textract service to notify SNS topic when response is ready
# Exponential backoff utility function
def exponential_backoff_retry(func, *args, retries=5, initial_delay=1, max_delay=32, **kwargs):
delay = initial_delay
for attempt in range(retries):
try:
return func(*args, **kwargs)
except ClientError as e:
error_code = e.response['Error']['Code']
if error_code in ['ProvisionedThroughputExceededException', 'LimitExceededException'] and attempt < retries - 1:
logger.warning(f"{error_code} on attempt {attempt + 1}/{retries}, retrying in {delay} seconds...")
time.sleep(delay)
delay = min(delay * 2, max_delay)
else:
logger.error(f"Exceeded maximum retries. Function {func.__name__} failed with error: {e}")
raise
# Function to start a Textract text detection job
def start_textract_detection_job(bucket_name, document_file_name, sns_topic_arn, sns_role_arn, job_tag):
# Define the parameters for the start_document_text_detection API
start_document_detection_params = {
'DocumentLocation': {
'S3Object': {
'Bucket': bucket_name,
'Name': document_file_name
}
},
'ClientRequestToken': 'unique-token-' + str(generate_unix_timestamp()), # Use a unique token for each request
'JobTag': job_tag, # Use a tag to identify your job
'NotificationChannel': {
'SNSTopicArn': sns_topic_arn,
'RoleArn': sns_role_arn # Role to allow Textract service to notify SNS topic when response is ready
}
logger.info('start_document_detection_params ' + str(start_document_detection_params))
# Send the request to start document detection
textract_response = textract_client.start_document_text_detection(**start_document_detection_params)
job_id = textract_response["JobId"]
logger.info(
"Started text detection job %s on %s.", job_id, document_file_name
}
logger.info('start_document_detection_params ' + str(start_document_detection_params))
try:
# Send the request to start document detection with exponential backoff
textract_response = exponential_backoff_retry(
textract_client.start_document_text_detection,
**start_document_detection_params
)
job_id = textract_response["JobId"]
logger.info("Started text detection job %s on %s.", job_id, document_file_name)
return job_id
except ClientError:
logger.exception("Couldn't detect text in %s.", document_file_name)
raise
else:
return job_id
def start_textract_analysis_job(
bucket_name,
document_file_name,
analysis_feature_type,
sns_topic_arn,
sns_role_arn,
job_tag,
):
# Function to start a Textract analysis job
def start_textract_analysis_job(bucket_name, document_file_name, analysis_feature_type, sns_topic_arn, sns_role_arn, job_tag):
# Define the parameters for the start_document_analysis API
start_document_analysis_params = {
'DocumentLocation': {
'S3Object': {
'Bucket': bucket_name,
'Name': document_file_name
}
},
'FeatureTypes': analysis_feature_type, # Customize based on requirements
'JobTag': job_tag, # Use a tag to identify your job
'NotificationChannel': {
'SNSTopicArn': sns_topic_arn,
'RoleArn': sns_role_arn # Role to allow Textract service to notify SNS topic when response is ready
}
}
logger.info('start_document_analysis_params ' + str(start_document_analysis_params))
try:
# Define the parameters for the start_document_analysis API
start_document_analysis_params = {
'DocumentLocation': {
'S3Object': {
'Bucket': bucket_name,
'Name': document_file_name
}
},
'FeatureTypes': analysis_feature_type, # Customize based on requirements
'JobTag': job_tag, # Use a tag to identify your job
'NotificationChannel': {
'SNSTopicArn': sns_topic_arn,
'RoleArn': sns_role_arn # Role to allow Textract service to notify SNS topic when response is ready
}
}
# 'ClientRequestToken': 'unique-token-'+str(generate_unix_timestamp()), # Use a unique token for each request
logger.info('start_document_analysis_params ' + str(start_document_analysis_params))
# Send the request to start document analysis
textract_response = textract_client.start_document_analysis(**start_document_analysis_params)
job_id = textract_response["JobId"]
logger.info(
"Started text analysis job %s on %s.", job_id, document_file_name
# Send the request to start document analysis with exponential backoff
textract_response = exponential_backoff_retry(
textract_client.start_document_analysis,
**start_document_analysis_params
)
job_id = textract_response["JobId"]
logger.info("Started text analysis job %s on %s.", job_id, document_file_name)
return job_id
except ClientError:
logger.exception("Couldn't analyze text in %s.", document_file_name)
raise
else:
return job_id
# 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(
@@ -171,7 +172,6 @@ def get_s3_object_tags(bucket_name, object_key):
tags_dict = {tag['Key']: tag['Value'] for tag in tags_list}
return tags_dict
except Exception as e:
logger.exception(f"Error: {e}")
print(f"Error: {e}")
@@ -179,12 +179,7 @@ def get_s3_object_tags(bucket_name, object_key):
# AWS Lambda handler function
def lambda_handler(event, context):
try:
# Extract AWS account ID and region from the Lambda ARN
# aws_account_id = context.invoked_function_arn.split(":")[4]
# aws_region = context.invoked_function_arn.split(":")[3]
logger.info('## ENVIRONMENT VARIABLES\r' + str(os.environ))
# Read environment variables
@@ -194,21 +189,21 @@ def lambda_handler(event, context):
global DATABASE_LOGGING_LAMBDA_FUNCTION_NAME
DATABASE_LOGGING_LAMBDA_FUNCTION_NAME = os.environ.get('DATABASE_LOGGING_LAMBDA_FUNCTION_NAME', '')
batch_id = ""
logger.info('DATABASE_LOGGING_LAMBDA_FUNCTION_NAME: ' + str(DATABASE_LOGGING_LAMBDA_FUNCTION_NAME))
logger.info('SNS_TOPIC_ARN: ' + SNS_TOPIC_ARN)
logger.info('TEXTRACT_ROLE_ARN: ' + TEXTRACT_ROLE_ARN)
# Read config.properties
file_path_array = property_file_path.split("/")
# Valid if file_path_array has more than 2 elements
# Validate 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}')
@@ -222,44 +217,44 @@ def lambda_handler(event, context):
# 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'])
# Read tags from file
tags_dict = get_s3_object_tags(S3_BUCKET_NAME,source_path)
tags_dict = get_s3_object_tags(S3_BUCKET_NAME, source_path)
JOB_TAG = config_dict['OTHERS']['JOB_TAG']
# check if key exist in tags_dict
# check if key exists in tags_dict
if "BatchId" in tags_dict.keys():
JOB_TAG = JOB_TAG + "-" + tags_dict['BatchId']
batch_id = tags_dict['BatchId']
else:
logger.error(f"BatchId not found in file tag")
return
# Extract configuration values
SOURCE_LOCATION = config_dict['FOLDER_LOCATIONS']['SOURCE_LOCATION'].format(batch_id) # SOURCE_LOCATION
STAGING_LOCATION = config_dict['FOLDER_LOCATIONS']['STAGING_LOCATION'].format(batch_id)
ANALYSIS_FEATURE_TYPE = config_dict['OTHERS']['ANALYSIS_FEATURE_TYPE'].split(",") # Analysis FeatureType
SENDER_MAX_FILES = int(config_dict['OTHERS']['SENDER_MAX_FILES'])
PROCESS_TYPE = str(config_dict['OTHERS']['PROCESS_TYPE']).upper()
# Construct the destination paths
destination_path = STAGING_LOCATION + source_path.replace(SOURCE_LOCATION,"")
# Move file to stagging
# Construct the destination paths
destination_path = STAGING_LOCATION + source_path.replace(SOURCE_LOCATION, "")
# Move file to staging
move_file_within_s3(S3_BUCKET_NAME, source_path, destination_path)
logger.info(f"Batch ID: {batch_id}")
logger.info('SOURCE_LOCATION: ' + SOURCE_LOCATION)
@@ -268,29 +263,29 @@ def lambda_handler(event, context):
logger.info('SENDER_MAX_FILES: ' + str(SENDER_MAX_FILES))
logger.info('JOB_TAG: ' + str(JOB_TAG))
logger.info('PROCESS_TYPE: ' + str(PROCESS_TYPE))
job_id = ""
if PROCESS_TYPE == "ANALYSIS":
# Start Textract analysis job
job_id = start_textract_analysis_job (
S3_BUCKET_NAME,
destination_path,
ANALYSIS_FEATURE_TYPE,
SNS_TOPIC_ARN,
TEXTRACT_ROLE_ARN,
JOB_TAG,
)
S3_BUCKET_NAME,
destination_path,
ANALYSIS_FEATURE_TYPE,
SNS_TOPIC_ARN,
TEXTRACT_ROLE_ARN,
JOB_TAG,
)
elif PROCESS_TYPE == "DETECTION":
# Start Textract detection job
job_id = start_textract_detection_job (
S3_BUCKET_NAME,
destination_path,
SNS_TOPIC_ARN,
TEXTRACT_ROLE_ARN,
JOB_TAG,
)
S3_BUCKET_NAME,
destination_path,
SNS_TOPIC_ARN,
TEXTRACT_ROLE_ARN,
JOB_TAG,
)
file_count = file_count + 1
logger.info(str(file_count) + '. ' + str(source_path) + " Job Id: " + str(job_id))
@@ -300,8 +295,8 @@ def lambda_handler(event, context):
logger.info(success_message)
document_id = get_filename_from_path(source_path)
document_id = os.path.splitext(document_id)[0]
generate_document_logs_input(document_id,job_id,"SENT_SUCCESSFULLY")
generate_document_logs_input(document_id, job_id, "SENT_SUCCESSFULLY")
return {
'statusCode': 200,
'body': success_message
@@ -310,11 +305,11 @@ def lambda_handler(event, context):
else:
error_message = 'Incorrect value for ENVIRONMENT VARIABLES: PROPERTY_FILE_S3_PATH\r' + str(property_file_path)
logger.error(error_message)
generate_document_logs_input(document_id,job_id,"FAILED_TO_SENT")
generate_document_logs_input(document_id, job_id, "FAILED_TO_SENT")
return {
'statusCode': 500,
'body': error_message
}
'statusCode': 500,
'body': error_message
}
except ClientError as e:
# Handle specific Textract client errors
@@ -336,30 +331,30 @@ def lambda_handler(event, context):
def get_filename_from_path(full_path):
return os.path.basename(full_path)
def generate_document_logs_input(document_id,job_id,textract_status):
current_time = datetime.datetime.now().isoformat()
data = {
def generate_document_logs_input(document_id, job_id, textract_status):
current_time = datetime.datetime.now().isoformat()
data = {
"operation": "update",
"table": "DOCUMENT_LOGS",
"data": {
"DOCUMENT_ID": document_id,
"STAGE" : "SENT_TO_TEXTRACT",
"JOB_ID" : job_id,
"STAGE": "SENT_TO_TEXTRACT",
"JOB_ID": job_id,
"MODIFIED_TIME": current_time,
"TEXTRACT_STATUS": textract_status,
"MODIFIED_BY": "TEXTRACT_SENDER"
}
}
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
}
logger.info(f"Request: {data}")
lambda_client = boto3.client('lambda')
response = lambda_client.invoke(
FunctionName=DATABASE_LOGGING_LAMBDA_FUNCTION_NAME,
InvocationType='Event',
Payload=json.dumps(data).encode('utf-8')
)
logger.info("Generated document logs input successfully")
return response
textract-pipeline/src/s3-files/config.properties
+1 -1
View File
@@ -28,6 +28,6 @@ TEXTRACT_TABLE_FORM_JSON=textract-table-form-output-json/{}/
[OTHERS]
SENDER_MAX_FILES=10
PROCESS_TYPE=ANALYSIS
ANALYSIS_FEATURE_TYPE=LAYOUT,TABLES,SIGNATURES
ANALYSIS_FEATURE_TYPE=LAYOUT
JOB_TAG=healthcare-contract
LOG_LEVEL=DEBUG
textract-pipeline/terraform/modules/aws-textract-pipeline-part3/variables.tf
+2 -2
View File
@@ -518,7 +518,7 @@ variable "textract_receiver_lambda" {
handler = "index.lambda_handler"
runtime = "python3.12"
timeout = 600
memory_size = 1024
memory_size = 2048
ephemeral_storage_size = 512
architectures = ["x86_64"]
layers = []
@@ -548,7 +548,7 @@ variable "text_creation_lambda" {
handler = "index.lambda_handler"
runtime = "python3.12"
timeout = 900
memory_size = 512
memory_size = 1024
ephemeral_storage_size = 512
architectures = ["x86_64"]
layers = []