Uso del flujo de trabajo de Analyze Lending

Para iniciar el análisis del documento de préstamo, utilice el siguiente comando de la CLI. Si desea ver documentos divididos, utilice el output-config argumento; de lo contrario, puede eliminarlo:

aws textract start-lending-analysis \
--document-location '{"S3Object":{"Bucket":"S3Bucket","Name":"FileName"}}' \
--output-config '{"S3Bucket": "S3Bucket",  "S3Prefix": "S3Prefix"}' \
--kms-key-id '1234abcd-12ab-34cd-56ef-1234567890ab' \
--region 'region-name'

Para obtener los resultados del análisis del documento de préstamo, utilice el siguiente comando de la CLI. El max-results argumento es opcional y, si no desea limitar la cantidad de resultados devueltos, puede eliminarlo:

aws textract get-lending-analysis \
--job-id 'jobId' \
--region 'us-west-2' \
--max-results 30

Para recuperar un resumen de los resultados:

aws textract get-lending-analysis-summary \
--job-id 'jobId' \
--region 'us-west-2' 

import boto3
import json
import sys
import time

class DocumentProcessor:

    def __init__(self, role, bucket, document, region):
        self.roleArn = role
        self.bucket = bucket
        self.document = document
        self.region_name = region

        self.textract = boto3.client('textract', region_name=self.region_name)
        self.sqs = boto3.client('sqs')
        self.sns = boto3.client('sns')

    def ProcessDocument(self):
        jobFound = False

        response = self.textract.start_lending_analysis(
            DocumentLocation={'S3Object': {'Bucket': self.bucket, 'Name': self.document}},
            NotificationChannel={'RoleArn': self.roleArn, 'SNSTopicArn': self.snsTopicArn})
        print('Processing type: Analysis')

        print('Start Job Id: ' + response['JobId'])
        dotLine = 0
        while jobFound == False:
            sqsResponse = self.sqs.receive_message(QueueUrl=self.sqsQueueUrl, MessageAttributeNames=['ALL'],
                                                   MaxNumberOfMessages=10)
            if sqsResponse:
                if 'Messages' not in sqsResponse:
                    if dotLine < 40:
                        print('.', end='')
                        dotLine = dotLine + 1
                    else:
                        print()
                        dotLine = 0
                    sys.stdout.flush()
                    time.sleep(5)
                    continue

                for message in sqsResponse['Messages']:
                    notification = json.loads(message['Body'])
                    textMessage = json.loads(notification['Message'])
                    print(textMessage['JobId'])
                    print(textMessage['Status'])
                    if str(textMessage['JobId']) == response['JobId']:
                        print('Matching Job Found:' + textMessage['JobId'])
                        jobFound = True
                        self.GetResults(textMessage['JobId'])
                        self.GetSummary(textMessage['JobId'])
                        self.sqs.delete_message(QueueUrl=self.sqsQueueUrl,
                                                ReceiptHandle=message['ReceiptHandle'])
                    else:
                        print("Job didn't match:" +
                              str(textMessage['JobId']) + ' : ' + str(response['JobId']))
                    # Delete the unknown message. Consider sending to dead letter queue
                    self.sqs.delete_message(QueueUrl=self.sqsQueueUrl,
                                            ReceiptHandle=message['ReceiptHandle'])

        print('Done!')

    def CreateTopicandQueue(self):

        millis = str(int(round(time.time() * 1000)))

        # Create SNS topic
        snsTopicName = "AmazonTextractTopic" + millis

        topicResponse = self.sns.create_topic(Name=snsTopicName)
        self.snsTopicArn = topicResponse['TopicArn']

        # create SQS queue
        sqsQueueName = "AmazonTextractQueue" + millis
        self.sqs.create_queue(QueueName=sqsQueueName)
        self.sqsQueueUrl = self.sqs.get_queue_url(QueueName=sqsQueueName)['QueueUrl']

        attribs = self.sqs.get_queue_attributes(QueueUrl=self.sqsQueueUrl,
                                                AttributeNames=['QueueArn'])['Attributes']

        sqsQueueArn = attribs['QueueArn']

        # Subscribe SQS queue to SNS topic
        self.sns.subscribe(
            TopicArn=self.snsTopicArn,
            Protocol='sqs',
            Endpoint=sqsQueueArn)

        # Authorize SNS to write SQS queue
        policy = """{{
  "Version":"2012-10-17",
  "Statement":[
    {{
      "Sid":"MyPolicy",
      "Effect":"Allow",
      "Principal" : {{"AWS" : "*"}},
      "Action":"sqs:*",
      "Resource": "{}",
      "Condition":{{
        "ArnEquals":{{
          "aws:SourceArn": "{}"
        }}
      }}
    }}
  ]
}}""".format(sqsQueueArn, self.snsTopicArn)

        response = self.sqs.set_queue_attributes(
            QueueUrl=self.sqsQueueUrl,
            Attributes={
                'Policy': policy
            })

    def DeleteTopicandQueue(self):
        self.sqs.delete_queue(QueueUrl=self.sqsQueueUrl)
        self.sns.delete_topic(TopicArn=self.snsTopicArn)

    # Display information about a block
    def DisplayExtractInfo(self, response):
        results = response['Results']
        for page in results:
            print("Page Classification: {}".format(page["PageClassification"]["PageType"]))
            print("Page Number: {}".format(page["Page"]))
            for extract in page["Extractions"]:
                for fields, vals in extract['LendingDocument'].items():
                    for val in vals:
                        print("Document Type: {}".format(val['Type']))
                        detections = val['ValueDetections']
                        for i in detections:
                            print(i['Text'])
                            print('Geometry')
                            print('\tBounding Box: {}'.format(i['Geometry']['BoundingBox']))
                            print('\tPolygon: {}'.format(i['Geometry']['Polygon']))

    def GetSummary(self, jobId):
        
        maxResults = 1000
        response = self.textract.get_lending_analysis_summary(JobId=jobId, MaxResults=maxResults)
        doc_groups = response['DocumentGroups']
        print("Summary info:")
        for group in doc_groups:
            print("Document type: " + group['Type'])
            split_docs = group['SplitDocuments']
            for doc in split_docs:
                print(doc)
                for idx, page in doc.items():
                    print(str(idx) + " - " + str(page))

    def GetResults(self, jobId):

        maxResults = 1000
        paginationToken = None
        finished = False

        while finished == False:

            response = None
            if paginationToken == None:
                response = self.textract.get_lending_analysis(JobId=jobId,
                                                              MaxResults=maxResults)
            else:
                response = self.textract.get_lending_analysis(JobId=jobId,
                                                              MaxResults=maxResults,
                                                              NextToken=paginationToken)

            print('Detected Document Text')
            print('Pages: {}'.format(response['DocumentMetadata']['Pages']))

            self.DisplayExtractInfo(response)

            if 'NextToken' in response:
                paginationToken = response['NextToken']
            else:
                finished = True

def main():
    roleArn = ''
    bucket = ''
    document = ''
    region_name = ''

    analyzer = DocumentProcessor(roleArn, bucket, document, region_name)
    analyzer.CreateTopicandQueue()
    analyzer.ProcessDocument()
    analyzer.DeleteTopicandQueue()


if __name__ == "__main__":
    main()