若要開始貸款文件分析，請使用下列 CLI 命令。如果您想要查看分割的文件，請使用 output-config引數，否則您可以將其移除：
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'
若要取得貸款文件分析的結果，請使用下列 CLI 命令。引max-results數是選用的，如果您不想限制傳回的結果數量，可以將其移除：
aws textract get-lending-analysis \
--job-id 'jobId' \
--region 'us-west-2' \
--max-results 30
若要擷取結果的摘要：
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()
