Verwenden Sie den Workflow „Kreditvergabe analysieren“

Verwenden Sie den folgenden CLI-Befehl, um die Analyse von Kreditdokumenten zu starten. Wenn Sie geteilte Dokumente sehen möchten, verwenden Sie das output-config Argument, andernfalls können Sie es entfernen:

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'

Verwenden Sie den folgenden CLI-Befehl, um die Ergebnisse der Analyse der Kreditdokumente abzurufen. Das max-results Argument ist optional. Wenn Sie die Anzahl der zurückgegebenen Ergebnisse nicht einschränken möchten, können Sie es entfernen:

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

Um eine Zusammenfassung der Ergebnisse abzurufen:

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()