diff --git a/sender/textract_wrapper.py b/sender/textract_wrapper.py deleted file mode 100644 index b16f695..0000000 --- a/sender/textract_wrapper.py +++ /dev/null @@ -1,276 +0,0 @@ -# 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