diff --git a/textract-pipeline/src/lambda/pdf-validation/index.py b/textract-pipeline/src/lambda/pdf-validation/index.py index bc31861..4604760 100644 --- a/textract-pipeline/src/lambda/pdf-validation/index.py +++ b/textract-pipeline/src/lambda/pdf-validation/index.py @@ -1,7 +1,8 @@ import boto3 import logging import botocore -from PyPDF2 import PdfReader +from PyPDF2 import PdfReader, PdfWriter +from PyPDF2.errors import PdfReadError from io import BytesIO import os from configparser import ConfigParser @@ -17,9 +18,9 @@ logger.setLevel(logging.INFO) s3_client = boto3.client('s3') DATABASE_LOGGING_LAMBDA_FUNCTION_NAME = "" + # 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( @@ -30,7 +31,7 @@ def get_s3_object_tags(bucket_name, 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} - + print(f"get_s3_object_tags: response={response}") return tags_dict except Exception as e: @@ -38,9 +39,9 @@ def get_s3_object_tags(bucket_name, object_key): print(f"Error: {e}") return None + # 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') @@ -56,15 +57,15 @@ def load_config_from_s3(bucket_name, file_key): return config_dict -def lambda_handler(event, context): +def lambda_handler(event, context): # 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 = "" - + # Read config.properties file_path_array = property_file_path.split("/") @@ -81,7 +82,6 @@ def lambda_handler(event, context): # Load config file config_dict = load_config_from_s3(S3_BUCKET_NAME, CONFIG_FILE_PATH) - logger.info("Config file loaded successfully.") print(event) @@ -99,20 +99,21 @@ def lambda_handler(event, context): logging.info("SOURCE_PATH: {key} ") logger.info(f"Processing file: s3://{S3_BUCKET_NAME}/{key}") - + # Get file tags tags_dict = get_s3_object_tags(S3_BUCKET_NAME, key) - + if "BatchId" in tags_dict: batch_id = tags_dict['BatchId'] else: logger.error("BatchId not found in file tag") return - + logger.info(f"Batch ID: {batch_id}") - + # Extract configuration values - INVALID_PDF_FILE_LOCATION = config_dict['FOLDER_LOCATIONS']['INVALID_PDF_FILE_LOCATION'].format(batch_id) + 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) ALL_PDF_LOCATION = config_dict['FOLDER_LOCATIONS']['ALL_PDF_LOCATION'].format(batch_id) @@ -121,7 +122,7 @@ def lambda_handler(event, context): logger.info(f"ALL_PDF_LOCATION : {ALL_PDF_LOCATION}") # Extract sub folder with filename - sub_folder_with_filename = key.replace(ALL_PDF_LOCATION,"") + sub_folder_with_filename = key.replace(ALL_PDF_LOCATION, "") logger.info(f"sub_folder_with_filename : {sub_folder_with_filename}") @@ -132,37 +133,50 @@ def lambda_handler(event, context): document_id = os.path.splitext(document_id)[0] logger.info(f"Processing S3 file - Bucket: {S3_BUCKET_NAME}, Key: {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=S3_BUCKET_NAME, CopySource={'Bucket': S3_BUCKET_NAME, 'Key': key}, Key=new_key) + s3_client.copy_object(Bucket=S3_BUCKET_NAME, CopySource={'Bucket': S3_BUCKET_NAME, 'Key': key}, + Key=new_key) s3_client.delete_object(Bucket=S3_BUCKET_NAME, 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=S3_BUCKET_NAME, Key=key)['Body'].read())) - if is_password_protected(pdf_reader): - logger.info("File is password protected. Moving to 'unprocessed' folder.") - move_to_unprocessed(S3_BUCKET_NAME, key,INVALID_PDF_FILE_LOCATION,sub_folder_with_filename) - generate_document_logs_input(document_id,num_pages,file_size,"CONTRACT_VALIDATION - PASSWORD PROTECTED") - return + s3_file = s3_client.get_object(Bucket=S3_BUCKET_NAME, Key=key) + pdf_reader = PdfReader(BytesIO(s3_file['Body'].read())) + + if pdf_reader.is_encrypted: + logger.info("File is encrypted. Trying to decrypt.") + try: + pdf_reader.decrypt("") + logger.info("File decrypted successfully.") + except Exception as e: + logger.error(f"Failed to decrypt file: {e}") + move_to_unprocessed(S3_BUCKET_NAME, key, INVALID_PDF_FILE_LOCATION, + sub_folder_with_filename) + generate_document_logs_input(document_id, num_pages, file_size, + "CONTRACT_VALIDATION - ENCRYPTED FILE") + return + # Validate file size file_size = s3_client.head_object(Bucket=S3_BUCKET_NAME, 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(S3_BUCKET_NAME, key,INVALID_PDF_FILE_LOCATION,sub_folder_with_filename) - generate_document_logs_input(document_id,num_pages,file_size,"CONTRACT_VALIDATION - INVALID FILE SIZE") + move_to_unprocessed(S3_BUCKET_NAME, key, INVALID_PDF_FILE_LOCATION, sub_folder_with_filename) + generate_document_logs_input(document_id, num_pages, file_size, + "CONTRACT_VALIDATION - INVALID FILE SIZE") 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(S3_BUCKET_NAME,key,INVALID_PDF_FILE_LOCATION,sub_folder_with_filename) - generate_document_logs_input(document_id,num_pages,file_size,"CONTRACT_VALIDATION - INVALID NUMBER OF PAGES") + move_to_unprocessed(S3_BUCKET_NAME, key, INVALID_PDF_FILE_LOCATION, sub_folder_with_filename) + generate_document_logs_input(document_id, num_pages, file_size, + "CONTRACT_VALIDATION - INVALID NUMBER OF PAGES") return except botocore.exceptions.ClientError as e: if e.response['Error']['Code'] == '404': @@ -171,82 +185,94 @@ def lambda_handler(event, context): return else: logger.error(f"Error checking file size: {str(e)}. Moving to 'unprocessed' folder.") - move_to_unprocessed(S3_BUCKET_NAME, key, INVALID_PDF_FILE_LOCATION,sub_folder_with_filename) - generate_document_logs_input(document_id,num_pages,file_size,"CONTRACT_VALIDATION - ERROR FILE SIZE") + move_to_unprocessed(S3_BUCKET_NAME, key, INVALID_PDF_FILE_LOCATION, sub_folder_with_filename) + generate_document_logs_input(document_id, num_pages, file_size, + "CONTRACT_VALIDATION - ERROR FILE SIZE") return + except PdfReadError as e: + logger.exception(f"PdfReadError: {str(e)}. Moving to 'unprocessed' folder.") + move_to_unprocessed(S3_BUCKET_NAME, key, INVALID_PDF_FILE_LOCATION, sub_folder_with_filename) + generate_document_logs_input(document_id, num_pages, file_size, + "CONTRACT_VALIDATION - ERROR FILE SIZE") + return except Exception as e: - logger.error(f"Error checking file size: {str(e)}. Moving to 'unprocessed' folder.") - move_to_unprocessed(S3_BUCKET_NAME, key, INVALID_PDF_FILE_LOCATION,sub_folder_with_filename) - generate_document_logs_input(document_id,num_pages,file_size,"CONTRACT_VALIDATION - ERROR FILE SIZE") - return + logger.error(f"Error checking file size: {str(e)}. Moving to 'unprocessed' folder.") + move_to_unprocessed(S3_BUCKET_NAME, key, INVALID_PDF_FILE_LOCATION, sub_folder_with_filename) + generate_document_logs_input(document_id, num_pages, file_size, + "CONTRACT_VALIDATION - ERROR FILE SIZE") + 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(S3_BUCKET_NAME, key,INVALID_PDF_FILE_LOCATION,sub_folder_with_filename) - generate_document_logs_input(document_id,num_pages,file_size,"CONTRACT_VALIDATION - INVALID RESOLUTION") + move_to_unprocessed(S3_BUCKET_NAME, key, INVALID_PDF_FILE_LOCATION, sub_folder_with_filename) + generate_document_logs_input(document_id, num_pages, file_size, + "CONTRACT_VALIDATION - INVALID RESOLUTION") return else: logger.info("File passed all conditions. Moving to 'processed' folder.") - move_to_processed(S3_BUCKET_NAME, key,SOURCE_LOCATION, sub_folder_with_filename) - + move_to_processed(S3_BUCKET_NAME, key, SOURCE_LOCATION, sub_folder_with_filename) + + generate_document_logs_input(document_id, num_pages, file_size, "CONTRACT_VALIDATION") - generate_document_logs_input(document_id,num_pages,file_size,"CONTRACT_VALIDATION") def is_password_protected(pdf_reader): return pdf_reader.is_encrypted -def is_resolution_valid(pdf_reader, max_resolution=3000*4000): + +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 + 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, sub_folder_with_filename): - #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}{sub_folder_with_filename}') +def move_to_processed(bucket, key, valid_file_location, sub_folder_with_filename): + # Replace processed with valid_file_location like in unprocessed method + s3_client.copy_object(Bucket=bucket, CopySource={'Bucket': bucket, 'Key': key}, + Key=f'{valid_file_location}{sub_folder_with_filename}') s3_client.delete_object(Bucket=bucket, Key=key) logger.info(f"File moved to 'processed' folder: s3://{bucket}/{valid_file_location}{sub_folder_with_filename}") -def move_to_unprocessed(bucket, key,invalid_file_location, sub_folder_with_filename): - s3_client.copy_object(Bucket=bucket, CopySource={'Bucket': bucket, 'Key': key}, Key=f'{invalid_file_location}{sub_folder_with_filename}') + +def move_to_unprocessed(bucket, key, invalid_file_location, sub_folder_with_filename): + s3_client.copy_object(Bucket=bucket, CopySource={'Bucket': bucket, 'Key': key}, + Key=f'{invalid_file_location}{sub_folder_with_filename}') s3_client.delete_object(Bucket=bucket, Key=key) - logger.info(f"File moved to 'unprocessed' folder: s3://{bucket}/{invalid_file_location}{sub_folder_with_filename}") + logger.info(f"File NOT ;] moved to 'unprocessed' folder: s3://{bucket}/{invalid_file_location}{sub_folder_with_filename}") + def get_filename_from_path(full_path): return os.path.basename(full_path) - - -def generate_document_logs_input(document_id,num_pages,file_size,stage): - current_time = datetime.datetime.now().isoformat() - - logger.info('DATABASE_LOGGING_LAMBDA_FUNCTION_NAME: ' + str(DATABASE_LOGGING_LAMBDA_FUNCTION_NAME)) - - data = { + + +def generate_document_logs_input(document_id, num_pages, file_size, stage): + current_time = datetime.datetime.now().isoformat() + + logger.info('DATABASE_LOGGING_LAMBDA_FUNCTION_NAME: ' + str(DATABASE_LOGGING_LAMBDA_FUNCTION_NAME)) + + data = { "operation": "update", "table": "DOCUMENT_LOGS", "data": { "DOCUMENT_ID": document_id, - "STAGE" : stage, - "NO_OF_PAGES" : num_pages, - "FILE_SIZE" : file_size, + "STAGE": stage, + "NO_OF_PAGES": num_pages, + "FILE_SIZE": file_size, "MODIFIED_TIME": current_time, "MODIFIED_BY": "CONTRACT_VALIDATION" - } } - 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(f"Generated document logs input successfully : {response}") - 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(f"Generated document logs input successfully : {response}") + return response