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

removed lambda and lambda-layer folders

Browse Source
This commit is contained in:
Pankaj Katariya
2024-05-10 19:21:38 +05:30
parent 117e5e22dc
commit 5e93589694
8 changed files with 0 additions and 1285 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lambda-layer/PyPDF2.zip
BIN
View File
Binary file not shown.
lambda-layer/brotlipy-layer.zip
BIN
View File
Binary file not shown.
lambda-layer/lo.tar.zip
BIN
View File
Binary file not shown.
lambda/docx_to_pdf/lambda_function.py
-250
View File
@@ -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
}
lambda/pdf-validation/lambda_function.py
-143
View File
@@ -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)
lambda/receiver/lambda_function.py
-318
View File
@@ -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
}
lambda/sender/lambda_function.py
-315
View File
@@ -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
}
lambda/textract_logs_lambda.py
-259
View File
@@ -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