Orchestration of sender lambda, tagging of textract request, tagging of files, added textract detection capability

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,"")

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":