diff --git a/lambda/receiver/lambda_function.py b/lambda/receiver/lambda_function.py index 3bbfa17..1212a94 100644 --- a/lambda/receiver/lambda_function.py +++ b/lambda/receiver/lambda_function.py @@ -4,6 +4,7 @@ import boto3 import os from configparser import ConfigParser from botocore.exceptions import ClientError +from urllib.parse import urlencode # Initialize logger logger = logging.getLogger(__name__) @@ -147,18 +148,18 @@ def get_textract_document_analysis(job_id, textract_client): return final_response # Function to save Textract response to S3 -def upload_response_to_s3(response, bucket_name, object_key, s3_client): +def upload_response_to_s3(response, bucket_name, object_key, s3_client, tags): # Convert the response to JSON response_json = json.dumps(response) try: - tags = "env=dev" # Upload the JSON response to S3 s3_client.put_object( Bucket=bucket_name, Key=object_key, Body=response_json, - ContentType='application/json' + ContentType='application/json', + Tagging=tags ) logger.info(f"Saved analysis response to S3: {object_key}") except Exception as e: @@ -177,6 +178,27 @@ def move_file_within_s3(source_bucket, source_path, destination_path): except Exception as e: logger.error(f"Error moving file: {e}") +# Function to get s3 object tags +def get_s3_object_tags(bucket_name, object_key): + + try: + # Get object tags + response = s3_client.get_object_tagging( + 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): @@ -226,7 +248,13 @@ def lambda_handler(event, context): 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': @@ -243,7 +271,7 @@ def lambda_handler(event, context): # 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) + 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,"") diff --git a/lambda/sender/lambda_function.py b/lambda/sender/lambda_function.py index 29dc1f5..c0efdfb 100644 --- a/lambda/sender/lambda_function.py +++ b/lambda/sender/lambda_function.py @@ -152,6 +152,27 @@ def start_textract_analysis_job( 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): @@ -223,6 +244,13 @@ def lambda_handler(event, context): # 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":