diff --git a/textract-pipeline/src/lambda/text-creation/index.py b/textract-pipeline/src/lambda/text-creation/index.py index f1ce175..bf4be48 100644 --- a/textract-pipeline/src/lambda/text-creation/index.py +++ b/textract-pipeline/src/lambda/text-creation/index.py @@ -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 diff --git a/textract-pipeline/src/lambda/textract-receiver/index.py b/textract-pipeline/src/lambda/textract-receiver/index.py index 6bfe86d..54aca8e 100644 --- a/textract-pipeline/src/lambda/textract-receiver/index.py +++ b/textract-pipeline/src/lambda/textract-receiver/index.py @@ -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 diff --git a/textract-pipeline/src/lambda/textract-sender/index.py b/textract-pipeline/src/lambda/textract-sender/index.py index 8eeef24..763a7c8 100644 --- a/textract-pipeline/src/lambda/textract-sender/index.py +++ b/textract-pipeline/src/lambda/textract-sender/index.py @@ -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 \ No newline at end of file + } + 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 diff --git a/textract-pipeline/src/s3-files/config.properties b/textract-pipeline/src/s3-files/config.properties index c0c0456..76b11c3 100644 --- a/textract-pipeline/src/s3-files/config.properties +++ b/textract-pipeline/src/s3-files/config.properties @@ -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 \ No newline at end of file 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 d0e2298..bdfc627 100644 --- a/textract-pipeline/terraform/modules/aws-textract-pipeline-part3/variables.tf +++ b/textract-pipeline/terraform/modules/aws-textract-pipeline-part3/variables.tf @@ -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 = []