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

updated lambda code

Browse Source
This commit is contained in:
Grzegorz Huber
2024-07-16 11:17:40 +02:00
parent 64671c6593
commit 5f9adc6bf8
1 changed files with 96 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
textract-pipeline/src/lambda/pdf-validation/index.py
+96 -70
View File
@@ -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