本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
处理连接错误
如果您超过每秒最大交易数 (TPS)，导致服务限制您的应用程序，或者您的连接中断，Amazon Textract 操作可能会失败。例如，如果您在短时间内对 Amazon Textract 操作进行了太多调用，它会限制您的呼叫并在操作响应中发送ProvisionedThroughputExceededException错误。有关亚马逊 Textract TPS 配额的信息，请参阅亚马逊 Textract 配额。要更改限制，您可以访问控制台中的 Amazon Textract 选项。 服务配额
您可以通过自动重试操作来管理限制和断开的连接。在创建 Amazon Textract 客户端时，您可以通过包含Config参数来指定重试次数。我们建议重试次数为 5。在失败并引发异常之前， AWS SDK 会重试指定次数。有关更多信息，请参阅 AWS 中的错误重试和指数退缩。
以下示例说明了在处理多个文档时如何自动重试 Amazon Textract 操作。
自动重试操作
-
将多个文档图像上传到您的 S3 存储桶以运行同步示例。将多页文档上传到您的 S3 存储桶，并在其StartDocumentTextDetection上运行以运行异步示例。
有关说明，请参阅《Amazon Simple Storage Service 用户指南》中的将对象上传到 Amazon S3。
-
以下示例演示如何使用Config参数自动重试操作。同步示例调用DetectDocumentText操作，而异步示例调用该GetDocumentTextDetection操作。
- Sync Example
-
使用以下示例对您的 Amazon S3 存储桶中的文档调用DetectDocumentText操作。在中main，将的值更改为您bucket的 S3 存储桶。将的值更改documents为您在步骤 2 中上传的文档图像的名称。
import boto3
from botocore.client import Config
# Documents
def process_multiple_documents(bucket, documents):
config = Config(retries = dict(max_attempts = 5))
# Amazon Textract client
textract = boto3.client('textract', config=config)
for documentName in documents:
print("\nProcessing: {}\n==========================================".format(documentName))
# Call Amazon Textract
response = textract.detect_document_text(
Document={
'S3Object': {
'Bucket': bucket,
'Name': documentName
}
})
# Print detected text
for item in response["Blocks"]:
if item["BlockType"] == "LINE":
print ('\033[94m' + item["Text"] + '\033[0m')
def main():
bucket = ""
documents = ["document-image-1.png",
"document-image-2.png", "document-image-3.png",
"document-image-4.png", "document-image-5.png" ]
process_multiple_documents(bucket, documents)
if __name__ == "__main__":
main()
- Async Example
-
使用以下示例调用 GetDocumentTextDetection 操作。它假设您已经调用StartDocumentTextDetection了 Amazon S3 存储桶中的文档并获得JobId了。在中main，将的值更改为您的 S3 存储桶，bucket将的值更改roleArn为分配给您的 Textract 角色的 Arn。您还需要将的document值更改为 Amazon S3 存储桶中多页文档的名称。最后，将的region_name值替换为您所在地区的名称，并为GetResults函数提供您的名称jobId。
import boto3
from botocore.client import Config
class DocumentProcessor:
jobId = ''
region_name = ''
roleArn = ''
bucket = ''
document = ''
sqsQueueUrl = ''
snsTopicArn = ''
processType = ''
def __init__(self, role, bucket, document, region):
self.roleArn = role
self.bucket = bucket
self.document = document
self.region_name = region
self.config = Config(retries = dict(max_attempts = 5))
self.textract = boto3.client('textract', region_name=self.region_name, config=self.config)
self.sqs = boto3.client('sqs')
self.sns = boto3.client('sns')
# Display information about a block
def DisplayBlockInfo(self, block):
print("Block Id: " + block['Id'])
print("Type: " + block['BlockType'])
if 'EntityTypes' in block:
print('EntityTypes: {}'.format(block['EntityTypes']))
if 'Text' in block:
print("Text: " + block['Text'])
if block['BlockType'] != 'PAGE':
print("Confidence: " + "{:.2f}".format(block['Confidence']) + "%")
print('Page: {}'.format(block['Page']))
if block['BlockType'] == 'CELL':
print('Cell Information')
print('\tColumn: {} '.format(block['ColumnIndex']))
print('\tRow: {}'.format(block['RowIndex']))
print('\tColumn span: {} '.format(block['ColumnSpan']))
print('\tRow span: {}'.format(block['RowSpan']))
if 'Relationships' in block:
print('\tRelationships: {}'.format(block['Relationships']))
print('Geometry')
print('\tBounding Box: {}'.format(block['Geometry']['BoundingBox']))
print('\tPolygon: {}'.format(block['Geometry']['Polygon']))
if block['BlockType'] == 'SELECTION_ELEMENT':
print(' Selection element detected: ', end='')
if block['SelectionStatus'] == 'SELECTED':
print('Selected')
else:
print('Not selected')
def GetResults(self, jobId):
maxResults = 1000
paginationToken = None
finished = False
while finished == False:
response = None
if paginationToken == None:
response = self.textract.get_document_text_detection(JobId=jobId,
MaxResults=maxResults)
else:
response = self.textract.get_document_text_detection(JobId=jobId,
MaxResults=maxResults,
NextToken=paginationToken)
blocks = response['Blocks']
print('Detected Document Text')
print('Pages: {}'.format(response['DocumentMetadata']['Pages']))
# Display block information
for block in blocks:
self.DisplayBlockInfo(block)
print()
print()
if 'NextToken' in response:
paginationToken = response['NextToken']
else:
finished = True
def main():
roleArn = 'role-arn'
bucket = 'bucket-name'
document = 'document-name'
region_name = 'region-name'
analyzer = DocumentProcessor(roleArn, bucket, document, region_name)
analyzer.GetResults("job-id")
if __name__ == "__main__":
main()
