From 71041d290b04fa0eb8bf57410f3a5f976089e79f Mon Sep 17 00:00:00 2001 From: Pankaj Katariya Date: Wed, 14 Feb 2024 18:18:34 +0530 Subject: [PATCH] Added Sender-Receiver python code --- receiver/lambda_function.py | 228 +++++++++++++++++++++++++++++ sender/lambda_function.py | 233 ++++++++++++++++++++++++++++++ sender/textract_wrapper.py | 276 ++++++++++++++++++++++++++++++++++++ 3 files changed, 737 insertions(+) create mode 100644 receiver/lambda_function.py create mode 100644 sender/lambda_function.py create mode 100644 sender/textract_wrapper.py diff --git a/receiver/lambda_function.py b/receiver/lambda_function.py new file mode 100644 index 0000000..9d62cf2 --- /dev/null +++ b/receiver/lambda_function.py @@ -0,0 +1,228 @@ +import json +import logging +import boto3 +import os +from configparser import ConfigParser +from botocore.exceptions import ClientError + +# 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_analysis(job_id, textract_client): + # Initialize an empty list to store blocks + all_blocks = [] + next_token = None + response = {} + flag = True + + 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): + # 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' + ) + 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}") + +# 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) + + logger.info('STAGGING_LOCATION: ' + STAGING_LOCATION) + logger.info('OUTPUT_LOCATION: ' + OUTPUT_LOCATION) + logger.info('PROCESSED_LOCATION: ' + PROCESSED_LOCATION) + logger.info('UNPROCESSED_LOCATION: ' + UNPROCESSED_LOCATION) + + # 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') + + # Check if the status is "SUCCEEDED" + if message_body.get('Status') == 'SUCCEEDED': + + # Call the function to get document analysis using Textract + document_analysis = get_textract_document_analysis(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_analysis, S3_BUCKET_NAME, s3_object_key, s3_client) + + # 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 + } + diff --git a/sender/lambda_function.py b/sender/lambda_function.py new file mode 100644 index 0000000..5db2eb8 --- /dev/null +++ b/sender/lambda_function.py @@ -0,0 +1,233 @@ +import boto3 +import time +from configparser import ConfigParser +import logging +import os +from botocore.exceptions import ClientError + +# 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: + # Copy the file to the destination folder + s3_client.copy_object(Bucket=source_bucket, CopySource={'Bucket': source_bucket, 'Key': source_key}, Key=destination_key) + + # 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_analysis_job( + bucket_name, + document_file_name, + analysis_feature_type, + sns_topic_arn, + sns_role_arn, + ): + + 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': 'healthcare-contract', # 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 + +# 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 + + 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)) + + # List S3 Object & iterate (as per max files allowed) + files_list = get_pdf_files_list_from_s3(S3_BUCKET_NAME,SOURCE_LOCATION) + + if len(files_list): + # File count + file_count = 0 + + for s3_file_key in files_list: + + # Construct the source and destination paths + source_path = s3_file_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) + + # Start Textract analysis job + job_id = start_textract_analysis_job ( + S3_BUCKET_NAME, + destination_path, + ANALYSIS_FEATURE_TYPE, + SNS_TOPIC_ARN, + TEXTRACT_ROLE_ARN, + ) + + file_count = file_count + 1 + logger.info(str(file_count) + '. ' + str(s3_file_key) + " Job Id: " + str(job_id)) + + if SENDER_MAX_FILES == file_count: + break + + success_message = 'Total files sent to textract : '+ str(file_count) + logger.info(success_message) + return { + 'statusCode': 200, + 'body': success_message + } + else: + message = 'No files found' + logger.error(message) + return { + 'statusCode': 500, + 'body': 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 + } + + \ No newline at end of file diff --git a/sender/textract_wrapper.py b/sender/textract_wrapper.py new file mode 100644 index 0000000..b16f695 --- /dev/null +++ b/sender/textract_wrapper.py @@ -0,0 +1,276 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Purpose + +Shows how to use the AWS SDK for Python (Boto3) with Amazon Textract to +detect text, form, and table elements in document images. +""" + +import json +import logging +from botocore.exceptions import ClientError + +logger = logging.getLogger(__name__) + + +# snippet-start:[python.example_code.textract.TextractWrapper] +class TextractWrapper: + """Encapsulates Textract functions.""" + + def __init__(self, textract_client, s3_resource, sqs_resource): + """ + :param textract_client: A Boto3 Textract client. + :param s3_resource: A Boto3 Amazon S3 resource. + :param sqs_resource: A Boto3 Amazon SQS resource. + """ + self.textract_client = textract_client + self.s3_resource = s3_resource + self.sqs_resource = sqs_resource + + # snippet-end:[python.example_code.textract.TextractWrapper] + + # snippet-start:[python.example_code.textract.DetectDocumentText] + def detect_file_text(self, *, document_file_name=None, document_bytes=None): + """ + Detects text elements in a local image file or from in-memory byte data. + The image must be in PNG or JPG format. + + :param document_file_name: The name of a document image file. + :param document_bytes: In-memory byte data of a document image. + :return: The response from Amazon Textract, including a list of blocks + that describe elements detected in the image. + """ + if document_file_name is not None: + with open(document_file_name, "rb") as document_file: + document_bytes = document_file.read() + try: + response = self.textract_client.detect_document_text( + Document={"Bytes": document_bytes} + ) + logger.info("Detected %s blocks.", len(response["Blocks"])) + except ClientError: + logger.exception("Couldn't detect text.") + raise + else: + return response + + # snippet-end:[python.example_code.textract.DetectDocumentText] + + # snippet-start:[python.example_code.textract.AnalyzeDocument] + def analyze_file( + self, feature_types, *, document_file_name=None, document_bytes=None + ): + """ + Detects text and additional elements, such as forms or tables, in a local image + file or from in-memory byte data. + The image must be in PNG or JPG format. + + :param feature_types: The types of additional document features to detect. + :param document_file_name: The name of a document image file. + :param document_bytes: In-memory byte data of a document image. + :return: The response from Amazon Textract, including a list of blocks + that describe elements detected in the image. + """ + if document_file_name is not None: + with open(document_file_name, "rb") as document_file: + document_bytes = document_file.read() + try: + response = self.textract_client.analyze_document( + Document={"Bytes": document_bytes}, FeatureTypes=feature_types + ) + logger.info("Detected %s blocks.", len(response["Blocks"])) + except ClientError: + logger.exception("Couldn't detect text.") + raise + else: + return response + + # snippet-end:[python.example_code.textract.AnalyzeDocument] + + # snippet-start:[python.example_code.textract.helper.prepare_job] + def prepare_job(self, bucket_name, document_name, document_bytes): + """ + Prepares a document image for an asynchronous detection job by uploading + the image bytes to an Amazon S3 bucket. Amazon Textract must have permission + to read from the bucket to process the image. + + :param bucket_name: The name of the Amazon S3 bucket. + :param document_name: The name of the image stored in Amazon S3. + :param document_bytes: The image as byte data. + """ + try: + bucket = self.s3_resource.Bucket(bucket_name) + bucket.upload_fileobj(document_bytes, document_name) + logger.info("Uploaded %s to %s.", document_name, bucket_name) + except ClientError: + logger.exception("Couldn't upload %s to %s.", document_name, bucket_name) + raise + + # snippet-end:[python.example_code.textract.helper.prepare_job] + + # snippet-start:[python.example_code.textract.helper.check_job_queue] + def check_job_queue(self, queue_url, job_id): + """ + Polls an Amazon SQS queue for messages that indicate a specified Textract + job has completed. + + :param queue_url: The URL of the Amazon SQS queue to poll. + :param job_id: The ID of the Textract job. + :return: The status of the job. + """ + status = None + try: + queue = self.sqs_resource.Queue(queue_url) + messages = queue.receive_messages() + if messages: + msg_body = json.loads(messages[0].body) + msg = json.loads(msg_body["Message"]) + if msg.get("JobId") == job_id: + messages[0].delete() + status = msg.get("Status") + logger.info( + "Got message %s with status %s.", messages[0].message_id, status + ) + else: + logger.info("No messages in queue %s.", queue_url) + except ClientError: + logger.exception("Couldn't get messages from queue %s.", queue_url) + else: + return status + + # snippet-end:[python.example_code.textract.helper.check_job_queue] + + # snippet-start:[python.example_code.textract.StartDocumentTextDetection] + def start_detection_job( + self, bucket_name, document_file_name, sns_topic_arn, sns_role_arn + ): + """ + Starts an asynchronous job to detect text elements in an image stored in an + Amazon S3 bucket. Textract publishes a notification to the specified Amazon SNS + topic when the job completes. + The image must be in PNG, JPG, or PDF format. + + :param bucket_name: The name of the Amazon S3 bucket that contains the image. + :param document_file_name: The name of the document image stored in Amazon S3. + :param sns_topic_arn: The Amazon Resource Name (ARN) of an Amazon SNS topic + where the job completion notification is published. + :param sns_role_arn: The ARN of an AWS Identity and Access Management (IAM) + role that can be assumed by Textract and grants permission + to publish to the Amazon SNS topic. + :return: The ID of the job. + """ + try: + response = self.textract_client.start_document_text_detection( + DocumentLocation={ + "S3Object": {"Bucket": bucket_name, "Name": document_file_name} + }, + NotificationChannel={ + "SNSTopicArn": sns_topic_arn, + "RoleArn": sns_role_arn, + }, + ) + job_id = 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 + + # snippet-end:[python.example_code.textract.StartDocumentTextDetection] + + # snippet-start:[python.example_code.textract.GetDocumentTextDetection] + def get_detection_job(self, job_id): + """ + Gets data for a previously started text detection job. + + :param job_id: The ID of the job to retrieve. + :return: The job data, including a list of blocks that describe elements + detected in the image. + """ + try: + response = self.textract_client.get_document_text_detection(JobId=job_id) + job_status = response["JobStatus"] + logger.info("Job %s status is %s.", job_id, job_status) + except ClientError: + logger.exception("Couldn't get data for job %s.", job_id) + raise + else: + return response + + # snippet-end:[python.example_code.textract.GetDocumentTextDetection] + + # snippet-start:[python.example_code.textract.StartDocumentAnalysis] + def start_analysis_job( + self, + bucket_name, + document_file_name, + feature_types, + sns_topic_arn, + sns_role_arn, + ): + """ + Starts an asynchronous job to detect text and additional elements, such as + forms or tables, in an image stored in an Amazon S3 bucket. Textract publishes + a notification to the specified Amazon SNS topic when the job completes. + The image must be in PNG, JPG, or PDF format. + + :param bucket_name: The name of the Amazon S3 bucket that contains the image. + :param document_file_name: The name of the document image stored in Amazon S3. + :param feature_types: The types of additional document features to detect. + :param sns_topic_arn: The Amazon Resource Name (ARN) of an Amazon SNS topic + where job completion notification is published. + :param sns_role_arn: The ARN of an AWS Identity and Access Management (IAM) + role that can be assumed by Textract and grants permission + to publish to the Amazon SNS topic. + :return: The ID of the job. + """ + try: + response = self.textract_client.start_document_analysis( + DocumentLocation={ + "S3Object": {"Bucket": bucket_name, "Name": document_file_name} + }, + NotificationChannel={ + "SNSTopicArn": sns_topic_arn, + "RoleArn": sns_role_arn, + }, + FeatureTypes=feature_types, + ) + job_id = 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 + + # snippet-end:[python.example_code.textract.StartDocumentAnalysis] + + # snippet-start:[python.example_code.textract.GetDocumentAnalysis] + def get_analysis_job(self, job_id): + """ + Gets data for a previously started detection job that includes additional + elements. + + :param job_id: The ID of the job to retrieve. + :return: The job data, including a list of blocks that describe elements + detected in the image. + """ + try: + response = self.textract_client.get_document_analysis(JobId=job_id) + job_status = response["JobStatus"] + logger.info("Job %s status is %s.", job_id, job_status) + except ClientError: + logger.exception("Couldn't get data for job %s.", job_id) + raise + else: + return response + + +# snippet-end:[python.example_code.textract.GetDocumentAnalysis] \ No newline at end of file