diff --git a/lambda-layer/PyPDF2.zip b/lambda-layer/PyPDF2.zip deleted file mode 100644 index 334c6d0..0000000 Binary files a/lambda-layer/PyPDF2.zip and /dev/null differ diff --git a/lambda-layer/brotlipy-layer.zip b/lambda-layer/brotlipy-layer.zip deleted file mode 100644 index 2294e36..0000000 Binary files a/lambda-layer/brotlipy-layer.zip and /dev/null differ diff --git a/lambda-layer/lo.tar.zip b/lambda-layer/lo.tar.zip deleted file mode 100644 index 6a2b9a6..0000000 Binary files a/lambda-layer/lo.tar.zip and /dev/null differ diff --git a/lambda/docx_to_pdf/lambda_function.py b/lambda/docx_to_pdf/lambda_function.py deleted file mode 100644 index 77d5528..0000000 --- a/lambda/docx_to_pdf/lambda_function.py +++ /dev/null @@ -1,250 +0,0 @@ -import os -from io import BytesIO -import tarfile -import boto3 -import subprocess -import brotli -from botocore.exceptions import ClientError -import logging -from configparser import ConfigParser -from urllib.parse import unquote_plus -import json - -# Initialize logger -logger = logging.getLogger(__name__) -logger.setLevel(logging.INFO) - -# Initialize S3 & Textract clients -s3_client = boto3.client('s3') - -LIBRE_OFFICE_INSTALL_DIR = '/tmp/instdir' - -# 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 - -# 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) - - # Delete the file from the source folder - s3_client.delete_object(Bucket=source_bucket, Key=source_path) - - logger.info(f"File moved from {source_path} to {destination_path}") - except Exception as e: - logger.error(f"Error moving file: {e}") - -def load_libre_office(): - if os.path.exists(LIBRE_OFFICE_INSTALL_DIR) and os.path.isdir(LIBRE_OFFICE_INSTALL_DIR): - print('We have a cached copy of LibreOffice, skipping extraction') - else: - print('No cached copy of LibreOffice, extracting tar stream from Brotli file.') - buffer = BytesIO() - with open('/opt/lo.tar.br', 'rb') as brotli_file: - d = brotli.Decompressor() - while True: - chunk = brotli_file.read(1024) - buffer.write(d.decompress(chunk)) - if len(chunk) < 1024: - break - buffer.seek(0) - print('Extracting tar stream to /tmp for caching.') - with tarfile.open(fileobj=buffer) as tar: - tar.extractall('/tmp') - print('Done caching LibreOffice!') - return f'{LIBRE_OFFICE_INSTALL_DIR}/program/soffice.bin' - -def download_from_s3(bucket, key, download_path): - s3 = boto3.client("s3") - s3.download_file(bucket, key, download_path) - -def upload_to_s3(file_path, bucket, key): - s3 = boto3.client("s3") - s3.upload_file(file_path, bucket, key) - -def convert_word_to_pdf(soffice_path, word_file_path, output_dir): - print(word_file_path) - conv_cmd = f"{soffice_path} --headless --norestore --invisible --nodefault --nofirststartwizard --nolockcheck --nologo --convert-to pdf:writer_pdf_Export --outdir {output_dir} {word_file_path}" - print(conv_cmd) - response = subprocess.run(conv_cmd.split(), stdout=subprocess.PIPE, stderr=subprocess.PIPE) - if response.returncode != 0: - response = subprocess.run(conv_cmd.split(), stdout=subprocess.PIPE, stderr=subprocess.PIPE) - - print(response.returncode, response.stdout, response.stderr) - if len(response.stderr) > 0: - print(response.stderr) - - if len(response.stdout) > 0: - print(response.stdout) - - if response.returncode != 0: - return False - - return True - -def lambda_handler(event, context): - - try: - - # Read environment variables - property_file_path = os.environ.get('PROPERTY_FILE_S3_PATH', '') - batch_id = os.environ.get('BATCH_ID', '') - - # 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) - - #logger.info('## CONFIG DICTIONARY\r' + str(config_dict)) - - # Extract configuration values - SOURCE_PDF_LOCATION = config_dict['FOLDER_LOCATIONS']['SOURCE_LOCATION'].format(batch_id) # SOURCE_LOCATION - SOURCE_DOCX_LOCATION = config_dict['FOLDER_LOCATIONS']['SOURCE_DOCX_LOCATION'].format(batch_id) # SOURCE_DOCX_LOCATION - SOURCE_DOCX_PROCESSED_LOCATION = config_dict['FOLDER_LOCATIONS']['SOURCE_DOCX_PROCESSED_LOCATION'].format(batch_id) # SOURCE_DOCX_LOCATION - SOURCE_DOCX_UNPROCESSED_LOCATION = config_dict['FOLDER_LOCATIONS']['SOURCE_DOCX_UNPROCESSED_LOCATION'].format(batch_id) # SOURCE_DOCX_LOCATION - logger.info('SOURCE_PDF_LOCATION: ' + SOURCE_PDF_LOCATION) - logger.info('SOURCE_DOCX_LOCATION: ' + SOURCE_DOCX_LOCATION) - logger.info('SOURCE_DOCX_PROCESSED_LOCATION: ' + SOURCE_DOCX_PROCESSED_LOCATION) - logger.info('SOURCE_DOCX_UNPROCESSED_LOCATION: ' + SOURCE_DOCX_UNPROCESSED_LOCATION) - - files = {} - files_success = 0 - files_failure = 0 - - # 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']) - - print('SQS Message Count: ',len(record_body['Records'])) - - for sqs_record in record_body['Records']: - - print(sqs_record) - - # decode source path - source_path = unquote_plus(sqs_record['s3']['object']['key']) - logging.info("SOURCE_PATH: {source_path} ") - - # Verify source path have valid file extension - if os.path.splitext(source_path)[1].lower() not in ['.doc','.docx']: - print('File type not supported: ', os.path.splitext(source_path)[1]) - files_failure+=1 - continue - - # Construct destination path - #destination_path = SOURCE_PDF_LOCATION + source_path.replace(SOURCE_DOCX_LOCATION,"") - processed_destination_path = SOURCE_DOCX_PROCESSED_LOCATION + source_path.replace(SOURCE_DOCX_LOCATION,"") - unprocessed_destination_path = SOURCE_DOCX_UNPROCESSED_LOCATION + source_path.replace(SOURCE_DOCX_LOCATION,"") - destination_path = "pdf_converted_files/" - temporary_filename = "conversion_file" - - key_prefix, base_name = os.path.split(source_path) - file_name, file_ext = os.path.splitext(base_name) - download_path = f"/tmp/{temporary_filename}{file_ext}" - output_dir = "/tmp" - - files[source_path] = False - - # Load Libreoffice library - libreoffice_exec_path = "" - if os.path.isfile('/opt/lo.tar.br'): - logging.info('compressed Libreoffice found!') - libreoffice_exec_path = load_libre_office() - else: - print('libreoffice Layer not found!') - return {'body': 'libreoffice Layer not found!'} - - logging.info("DOWNLOAD_PATH: {download_path}") - download_from_s3(S3_BUCKET_NAME, source_path, download_path) - logging.info('Downloading Finished!') - - print("Files list after download: ",os.listdir('/tmp')) - - logger.info('Starting Conversion') - - is_converted = convert_word_to_pdf(libreoffice_exec_path, download_path, output_dir) - output_filepath = f"{output_dir}/{temporary_filename}.pdf" - - print("Files list after conversion: ",os.listdir('/tmp')) - - if is_converted and os.path.isfile(output_filepath): - logger.info('Conversion Success!') - file_name, _ = os.path.splitext(base_name) - files_success+=1 - files[source_path] = True - - logger.info(f'Saving converted file in Bucket {S3_BUCKET_NAME} with Key : {destination_path}{file_name}.pdf') - - # Upload file to S3 location - upload_to_s3(output_filepath, S3_BUCKET_NAME, f"{destination_path}{file_name}.pdf") - - # Move file to processed folder - move_file_within_s3(S3_BUCKET_NAME, source_path, processed_destination_path) - - logger.info("File converted: " + source_path) - else: - # Move file to unprocessed folder - move_file_within_s3(S3_BUCKET_NAME, source_path, unprocessed_destination_path) - - files_failure+=1 - logger.error("File not converted: " + source_path) - - files_converted = { - 'statusCode': 200, - 'body': { - 'Files_Processed': len(files), - 'Success' : files_success, - 'Failure': files_failure - } - } - logger.info(files_converted) - return files_converted - - except ClientError as e: - # Handle specific Textract client errors - error_message = f"Error in pdf conversion operation: {e}" - logger.error(error_message) - return { - 'statusCode': 500, - 'body': error_message - } - - except Exception as e: - # Handle other exceptions - error_message = f"Unexpected error: {e}" - logger.error(error_message) - return { - 'statusCode': 500, - 'body': error_message - } - - \ No newline at end of file diff --git a/lambda/pdf-validation/lambda_function.py b/lambda/pdf-validation/lambda_function.py deleted file mode 100644 index 88eb546..0000000 --- a/lambda/pdf-validation/lambda_function.py +++ /dev/null @@ -1,143 +0,0 @@ -import boto3 -import logging -import botocore -from PyPDF2 import PdfReader -from io import BytesIO -import os -from configparser import ConfigParser -import urllib.parse -from urllib.parse import urlparse - -# Configure logging -logger = logging.getLogger() -logger.setLevel(logging.INFO) - -# Initialize S3 & Textract clients -s3_client = boto3.client('s3') - -# 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 - -def lambda_handler(event, context): - - for record in event['Records']: - - # Retrieve the S3 bucket and key from the event - bucket = event['Records'][0]['s3']['bucket']['name'] - key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key']) - region = record['awsRegion'] - - logger.info(f"Processing S3 file - Bucket: {bucket}, Key: {key}, Region: {region}") - - # Read environment variables - property_file_path = os.environ.get('PROPERTY_FILE_S3_PATH', '') - batch_id = os.environ.get('BATCH_ID', '') - logger.info(f"Batch ID: {batch_id}") - file_path_array = property_file_path.split("/") - # 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"Using S3 bucket: {S3_BUCKET_NAME} and config file path: {CONFIG_FILE_PATH}") - - # Load config file - config_dict = load_config_from_s3(bucket, CONFIG_FILE_PATH) - INVALID_PDF_FILE_LOCATION = config_dict['FOLDER_LOCATIONS']['INVALID_PDF_FILE_LOCATION'].format(batch_id) - SOURCE_LOCATION = config_dict['FOLDER_LOCATIONS']['SOURCE_LOCATION'].format(batch_id) - logger.info("Config file loaded successfully.") - - logger.info(f"Processing file: s3://{bucket}/{key}") - - - # Check if the file has a '.filepart' extension - if key.lower().endswith('.filepart'): - new_key = key[:-9] + '.pdf' # Rename to have a '.pdf' extension - s3_client.copy_object(Bucket=bucket, CopySource={'Bucket': bucket, 'Key': key}, Key=new_key) - s3_client.delete_object(Bucket=bucket, Key=key) - key = new_key # Update key to the new filename - - - - # Validate number of pages - try: - pdf_reader = PdfReader(BytesIO(s3_client.get_object(Bucket=bucket, Key=key)['Body'].read())) - if is_password_protected(pdf_reader): - logger.info("File is password protected. Moving to 'unprocessed' folder.") - move_to_unprocessed(bucket, key,INVALID_PDF_FILE_LOCATION) - # Validate file size - file_size = s3_client.head_object(Bucket=bucket, Key=key)['ContentLength'] - logger.info(f"File size: {file_size} bytes") - if file_size > 500 * 1024 * 1024: # 500MB - logger.info("File size exceeds 500MB. Moving to 'unprocessed' folder.") - move_to_unprocessed(bucket, key,INVALID_PDF_FILE_LOCATION) - return - - num_pages = len(pdf_reader.pages) - logger.info(f"Number of pages: {num_pages}") - if num_pages > 3000: - logger.info("Number of pages exceeds 3000. Moving to 'unprocessed' folder.") - move_to_unprocessed(bucket,key,INVALID_PDF_FILE_LOCATION) - return - except botocore.exceptions.ClientError as e: - if e.response['Error']['Code'] == '404': - logger.error(f"File not found: s3://{bucket}/{key}") - # Handle the case where the file doesn't exist - return - else: - logger.error(f"Error checking file size: {str(e)}. Moving to 'unprocessed' folder.") - move_to_unprocessed(bucket, key, INVALID_PDF_FILE_LOCATION) - return - except Exception as e: - logger.error(f"Error checking file size: {str(e)}. Moving to 'unprocessed' folder.") - move_to_unprocessed(bucket, key, INVALID_PDF_FILE_LOCATION) - return - - # Validate password protection and resolution - - if not is_resolution_valid(pdf_reader): - logger.info("File has invalid resolution. Moving to 'unprocessed' folder.") - move_to_unprocessed(bucket, key,INVALID_PDF_FILE_LOCATION) - else: - logger.info("File passed all conditions. Moving to 'processed' folder.") - move_to_processed(bucket, key,SOURCE_LOCATION) - -def is_password_protected(pdf_reader): - return pdf_reader.is_encrypted - -def is_resolution_valid(pdf_reader, max_resolution=3000*4000): - for page_num in range(len(pdf_reader.pages)): - page = pdf_reader.pages[page_num] - page_width, page_height = page.mediabox.upper_right - if page_width > max_resolution or page_height > max_resolution: - return False - return True - - -def move_to_processed(bucket, key,valid_file_location): - #replace processed with valid_file_location like in unprocessed method - #s3_client.copy_object(Bucket=bucket, CopySource={'Bucket': bucket, 'Key': key}, Key=f'processed/{key}') - s3_client.copy_object(Bucket=bucket, CopySource={'Bucket': bucket, 'Key': key}, Key=f'{valid_file_location}/{get_filename_from_path(key)}') - s3_client.delete_object(Bucket=bucket, Key=key) - logger.info(f"File moved to 'processed' folder: s3://{bucket}/{get_filename_from_path(key)}") - -def move_to_unprocessed(bucket, key,invalid_file_location): - s3_client.copy_object(Bucket=bucket, CopySource={'Bucket': bucket, 'Key': key}, Key=f'{invalid_file_location}{get_filename_from_path(key)}') - s3_client.delete_object(Bucket=bucket, Key=key) - logger.info(f"File moved to 'unprocessed' folder: s3://{bucket}/{invalid_file_location}{get_filename_from_path(key)}") - -def get_filename_from_path(full_path): - return os.path.basename(full_path) diff --git a/lambda/receiver/lambda_function.py b/lambda/receiver/lambda_function.py deleted file mode 100644 index 1212a94..0000000 --- a/lambda/receiver/lambda_function.py +++ /dev/null @@ -1,318 +0,0 @@ -import json -import logging -import boto3 -import os -from configparser import ConfigParser -from botocore.exceptions import ClientError -from urllib.parse import urlencode - -# Initialize logger -logger = logging.getLogger(__name__) -logger.setLevel(logging.INFO) - -# Initialize S3 & Textract clients -s3_client = boto3.client('s3') -textract_client = boto3.client('textract') - -# 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 construct JSON path from S3 object path -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 - if txt.lower().endswith(".pdf"): - return dest_folder+txt[0:-4]+".json" - -# Function to get Textract document analysis -def get_textract_document_detection(job_id, textract_client): - # Initialize an empty list to store blocks - all_blocks = [] - next_token = None - response = {} - flag = True - logger.info("Get document detection") - try: - while True: - if flag: - # Call the Textract API to get document analysis - response = 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 - ) - - 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 - else: - # Remove unnecessary keys from the last response - last_response = response.copy() - last_response.pop('Blocks', None) - last_response.pop('ResponseMetadata', None) - 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") - 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 - all_blocks = [] - next_token = None - response = {} - flag = True - logger.info("Get document analysis") - try: - while True: - if flag: - # Call the Textract API to get document analysis - response = 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 - ) - - 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 - else: - # Remove unnecessary keys from the last response - last_response = response.copy() - last_response.pop('Blocks', None) - last_response.pop('ResponseMetadata', None) - 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") - 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( - Bucket=bucket_name, - Key=object_key, - Body=response_json, - ContentType='application/json', - 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)}") - -# 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) - - # Delete the file from the source folder - s3_client.delete_object(Bucket=source_bucket, Key=source_path) - - logger.info(f"File moved from {source_path} to {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( - 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}") - print(f"Error: {e}") - return None - -# AWS Lambda handler function -def lambda_handler(event, context): - - logger.info('## ENVIRONMENT VARIABLES\r' + str(os.environ)) - - # Read environment variables - property_file_path = os.environ.get('PROPERTY_FILE_S3_PATH', '') - batch_id = os.environ.get('BATCH_ID', '') - - # 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:]) - - 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) - - logger.info('## CONFIG DICTIONARY\r' + str(config_dict)) - - 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() - - logger.info('STAGGING_LOCATION: ' + STAGING_LOCATION) - logger.info('OUTPUT_LOCATION: ' + OUTPUT_LOCATION) - logger.info('PROCESSED_LOCATION: ' + PROCESSED_LOCATION) - logger.info('UNPROCESSED_LOCATION: ' + UNPROCESSED_LOCATION) - logger.info('PROCESS_TYPE: ' + str(PROCESS_TYPE)) - - # 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']) - 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) - - # 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) - - # Construct the destination paths - 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) - - success_message = 'Processed file '+ str(s3_object_key) - logger.info(success_message) - return { - 'statusCode': 200, - 'body': success_message - } - else: - error_message = f"Skipping message with JobId {message_body.get('JobId')} as Status is not 'SUCCEEDED'" - logger.info(error_message) - - # Construct the destination paths - 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) - 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) - return { - '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 - } - diff --git a/lambda/sender/lambda_function.py b/lambda/sender/lambda_function.py deleted file mode 100644 index c0efdfb..0000000 --- a/lambda/sender/lambda_function.py +++ /dev/null @@ -1,315 +0,0 @@ -import boto3 -import time -from configparser import ConfigParser -import logging -import os -from botocore.exceptions import ClientError -import json -from urllib.parse import unquote_plus - -# Initialize logger -logger = logging.getLogger(__name__) -logger.setLevel(logging.INFO) - -# Initialize S3 & Textract clients -s3_client = boto3.client('s3') -textract_client = boto3.client('textract') - -# Function to generate a Unix timestamp -def generate_unix_timestamp(): - # Get the current time in seconds since the epoch - unix_timestamp = int(time.time()) - return 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') - - # 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 move a file from source to destination in S3 -def move_file_within_s3(source_bucket, source_key, destination_key): - try: - tags = "env=dev" - # Copy the file to the destination folder - s3_client.copy_object(Bucket=source_bucket, CopySource={'Bucket': source_bucket, 'Key': source_key}, Key=destination_key, Tagging=f'{tags}') - - # Delete the file from the source folder - s3_client.delete_object(Bucket=source_bucket, Key=source_key) - - logger.info(f"File moved from {source_key} to {destination_key}") - except ClientError as e: - logger.error(f"Error moving file: {e}") - except Exception as e: - logger.error(f"Error moving file: {e}") - -# Function to get a list of PDF files in a given S3 folder -def get_pdf_files_list_from_s3(source_bucket, source_folder): - file_list = [] - # List S3 Object & iterate (as per max files allowed) - s3_list_response = s3_client.list_objects_v2(Bucket=source_bucket, Prefix=source_folder) - - if s3_list_response and s3_list_response['ResponseMetadata']['HTTPStatusCode'] == 200 and s3_list_response['KeyCount'] != 0: - objects = s3_list_response['Contents'] - - for s3_object in objects: - # Skip non-PDF files - if not s3_object['Key'].lower().endswith('.pdf'): - continue - - file_list.append(s3_object['Key']) - - 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 - } - } - - 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 - ) - 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, - ): - - 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 - '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_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 - ) - 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( - 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}") - print(f"Error: {e}") - return None - -# 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 - property_file_path = os.environ.get('PROPERTY_FILE_S3_PATH', '') - batch_id = os.environ.get('BATCH_ID', '') - - # 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) - - logger.info('## CONFIG DICTIONARY\r' + str(config_dict)) - - # 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']) - SNS_TOPIC_ARN = config_dict['RESOURCES']['SNS_TOPIC_ARN'].replace("{aws_region}",aws_region).replace("{aws_account_id}",aws_account_id) - TEXTRACT_ROLE_ARN = config_dict['RESOURCES']['TEXTRACT_ROLE_ARN'].replace("{aws_account_id}",aws_account_id) # Textract IAM Role ARN to publish to SNS - JOB_TAG = config_dict['OTHERS']['JOB_TAG'] - PROCESS_TYPE = str(config_dict['OTHERS']['PROCESS_TYPE']).upper() - - logger.info('SOURCE_LOCATION: ' + SOURCE_LOCATION) - logger.info('STAGING_LOCATION: ' + STAGING_LOCATION) - logger.info('ANALYSIS_FEATURE_TYPE: ' + str(ANALYSIS_FEATURE_TYPE)) - logger.info('SNS_TOPIC_ARN: ' + SNS_TOPIC_ARN) - logger.info('TEXTRACT_ROLE_ARN: ' + TEXTRACT_ROLE_ARN) - logger.info('SENDER_MAX_FILES: ' + str(SENDER_MAX_FILES)) - logger.info('JOB_TAG: ' + str(JOB_TAG)) - logger.info('PROCESS_TYPE: ' + str(PROCESS_TYPE)) - - # File count - file_count = 0 - - # 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 and destination paths - source_path = unquote_plus(sqs_record['s3']['object']['key']) - destination_path = STAGING_LOCATION + source_path.replace(SOURCE_LOCATION,"") - - # Move file to stagging - move_file_within_s3(S3_BUCKET_NAME, source_path, destination_path) - - # Read tags from file - tags_dict = get_s3_object_tags(S3_BUCKET_NAME,destination_path) - - # check if key exist in tags_dict - if "batch_id" in tags_dict.keys(): - JOB_TAG = JOB_TAG + "-" + tags_dict['batch_id'] - - 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, - ) - 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, - ) - - file_count = file_count + 1 - logger.info(str(file_count) + '. ' + str(source_path) + " Job Id: " + str(job_id)) - time.sleep(1) - - success_message = 'Total files sent to textract : '+ str(file_count) - logger.info(success_message) - return { - 'statusCode': 200, - 'body': success_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 - } - - except ClientError as e: - # Handle specific Textract client errors - error_message = f"Error in Textract operation: {e}" - logger.error(error_message) - return { - 'statusCode': 500, - 'body': error_message - } - - except Exception as e: - # Handle other exceptions - error_message = f"Unexpected error: {e}" - logger.error(error_message) - return { - 'statusCode': 500, - 'body': error_message - } - - \ No newline at end of file diff --git a/lambda/textract_logs_lambda.py b/lambda/textract_logs_lambda.py deleted file mode 100644 index 0faa05d..0000000 --- a/lambda/textract_logs_lambda.py +++ /dev/null @@ -1,259 +0,0 @@ -import json -import snowflake.connector -import boto3 -import snowflake.connector -import logging - - -""" -# Sample input events - -doc_input_event = { - "operation": "insert", - "table": "DOCUMENT_LOGS", - "data": { - "BATCH_ID": 101, - "JOB_ID": "J123456", - "STAGE": "Processing", - "TEXTRACT_STATUS": "Success", - "BUCKET_NAME": "doc-bucket", - "FILE_NAME": "file1.pdf", - "FILE_PATH": "/documents/2023/", - "DOCUMENT_TYPE": "Report", - "PAYER_SIGNED": True, - "PROVIDER_SIGNED": False, - "GROUP_ID": "G100", - "CREATED_TIME": "2024-03-21 10:00:00", - "MODIFIED_TIME": "2024-03-21 10:00:00", - "CREATED_BY": "admin", - "MODIFIED_BY": "admin", - "ORIGINAL_FILE_EXTENSION": "pdf", - "NO_OF_PAGES": 10, - "FILE_SIZE": 1048576 - } -} - - - -doc_update_event = { - "operation": "update", - "table": "DOCUMENT_LOGS", - "data": { - "DOCUMENT_ID": 1001, - "TEXTRACT_STATUS": "Failed", - "MODIFIED_TIME": "2024-03-22 15:00:00", - "MODIFIED_BY": "admin" - } -} - - - -batch_insert_event = { - "operation": "insert", - "table": "BATCH_LOGS", - "data": { - "CLIENT_ID": "C200", - "EXECUTION_START_TIME": "2024-03-21 09:00:00", - "NO_OF_DOCUMENTS": 150, - "USER_NAME": "batch_processor" - } -} - -batch_update_event = { - "operation": "update", - "table": "BATCH_LOGS", - "data": { - "BATCH_ID": 102, - "NO_OF_DOCUMENTS": 155, - "USER_NAME": "updated_processor" - } -} - -client_insert_event = { - "operation": "insert", - "table": "CLIENT_LOGS", - "data": { - "CLIENT_ID": "CL300", - "CLIENT_NAME": "Acme Corporation", - "BUCKET_NAME": "acme-docs" - } -} - -client_update_event = { - "operation": "update", - "table": "CLIENT_LOGS", - "data": { - "CLIENT_ID": "CL300", - "BUCKET_NAME": "new-acme-docs" - } -} -Operation can be: 'insert' or 'update' -Data is a dictionary with the columns and values to be inserted or updated - -""" - - -logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s', force=True) -logging.getLogger('snowflake.connector').setLevel(logging.WARNING) -logging.getLogger("botocore").setLevel(logging.WARNING) -logger = logging.getLogger(__name__) - - -def get_secret(secrets_name: str): - """Get credentials from Secret Manager as dict""" - secrets_manager = boto3.client('secretsmanager') - get_secret_value_response = secrets_manager.get_secret_value(SecretId=secrets_name) - - if 'SecretString' in get_secret_value_response: - secret_json = get_secret_value_response['SecretString'] - else: - secret_json = base64.b64decode(get_secret_value_response['SecretBinary']) - - return json.loads(secret_json) - - -def get_snowflake_db_connection(secrets_name: str): - """Create connection to Snowflake db.""" - try: - con_params = get_secret(secrets_name) - account = con_params['account_locator'] - user = con_params['user'] - password = con_params['password'] - database = con_params['database'].upper() - warehouse = con_params['warehouse'] - role= con_params['role'] - logger.info(f'Using credentials: account={account}, user={user}, password=***, database={database}, ' - f'warehouse={warehouse}') - snowflake_connection = snowflake.connector.connect(account=account, user=user, password=password, database=database, - warehouse=warehouse, autocommit=True) - logger.info(snowflake_connection) - return snowflake_connection - except Exception as e: - return e - - -# Leaving this statement outside the lambda_handler function to reuse the connection - # Secret has been setup to use the logging service account -conn = get_snowflake_db_connection('doczy-dev-db-svc-acc') -cur = conn.cursor() - - -def construct_doc_insert_sql(data): - """ - Constructs the SQL for an insert operation - - Sample return value: - INSERT INTO STG.DOCUMENT_LOGS (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 (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.DOCUMENT_LOGS ({columns}) VALUES ({values});" - return sql - - -def construct_doc_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.DOCUMENT_LOGS SET {set_clauses} WHERE DOCUMENT_ID = {document_id};" - return sql - - -def construct_batch_insert_sql(data): - """ - Constructs the SQL for an insert operation - - Sample return value: - INSERT INTO STG.BATCH_LOGS (CLIENT_ID, EXECUTION_START_TIME, NO_OF_DOCUMENTS, USER_NAME) VALUES ('C200', '2024-03-21 09:00:00', 150, 'batch_processor'); - """ - 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.BATCH_LOGS ({columns}) VALUES ({values});" - return sql - -def construct_client_insert_sql(data): - """ - Constructs the SQL for an insert operation - - Sample return value: - INSERT INTO STG.CLIENT_LOGS (CLIENT_ID, CLIENT_NAME, BUCKET_NAME) VALUES ('CL300', 'Acme Corporation', 'acme-docs'); - """ - 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.CLIENT_LOGS ({columns}) VALUES ({values});" - return sql - - -def construct_client_update_sql(data, client_id): - """ - Constructs the SQL for an update operation - - Sample return value: - UPDATE STG.CLIENT_LOGS SET BUCKET_NAME = 'new-acme-docs' WHERE CLIENT_ID = 'CL300'; - """ - 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.CLIENT_LOGS SET {set_clauses} WHERE CLIENT_ID = '{client_id}';" - return sql - -def construct_batch_update_sql(data, batch_id): - """ - Constructs the SQL for an update operation - - Sample return value: - UPDATE STG.BATCH_LOGS SET NO_OF_DOCUMENTS = 155, USER_NAME = 'updated_processor' WHERE BATCH_ID = 1; - """ - 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.BATCH_LOGS SET {set_clauses} WHERE BATCH_ID = {batch_id};" - return sql - - -# Main Lambda handler -def lambda_handler(event, context): - # Extract operation type and payload from event - - - operation = event['operation'] # 'insert' or 'update' - data = event['data'] - table = event['table'] - - - try: - if table == 'DOCUMENT_LOGS': - if operation == 'insert': - sql = construct_doc_insert_sql(data) - elif operation == 'update': - document_id = data.pop('DOCUMENT_ID', None) - sql = construct_doc_update_sql(data, document_id) - - elif table == 'BATCH_LOGS': - if operation == 'insert': - sql = construct_batch_insert_sql(data) - elif operation == 'update': - batch_id = data.pop('BATCH_ID', None) - sql = construct_batch_update_sql(data, batch_id) - - elif table == 'CLIENT_LOGS': - if operation == 'insert': - sql = construct_client_insert_sql(data) - elif operation == 'update': - client_id = data.pop('CLIENT_ID', None) - sql = construct_client_update_sql(data, client_id) - else: - raise ValueError("Unsupported table.") - logger.info(f"Executing the following logging SQL statement: {sql}") - cur.execute(sql) - return {'statusCode': 200, 'body': json.dumps('Operation successful')} - - except Exception as e: - return {'statusCode': 400, 'body': json.dumps(str(e))} - finally: - # conn.close() - # Need to close the connection to avoid reaching the limit of open connections - # However, we need the conn to be in hot state for subsequent concurrent executions - # Need to decide the best approach to handle this - pass