本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。
處理連線錯誤
如果您超過每秒交易數上限 (TPS)，導致服務限流您的應用程式或連線中斷時，Amazon Textract 操作可能會失敗。例如，如果您在短時間內對 Amazon Textract 操作進行太多呼叫，它會調節您的呼叫，並在操作回應中傳送ProvisionedThroughputExceededException錯誤。如需 Amazon Textract TPS 配額的相關資訊，請參閱 Amazon Textract Quotas。若要變更限制，您可以在 Service Quotas 主控台中存取 Amazon Textract 選項。
您可以透過自動重試 操作來管理限流和中斷連線。您可以在建立 Amazon Textract 用戶端時包含 Config 參數，以指定重試次數。我們建議重試計數為 5。 AWS 開發套件會在失敗並擲回例外狀況之前，以指定的次數重試 操作。如需詳細資訊，請參閱 AWS 中的錯誤重試與指數退避。
下列範例顯示如何在處理多個文件時自動重試 Amazon Textract 操作。
自動重試操作
-
將多個文件映像上傳至 S3 儲存貯體，以執行同步範例。將多頁文件上傳至 S3 儲存貯體，並在其StartDocumentTextDetection上執行以執行非同步範例。
如需指示說明，請參閱 Amazon 簡單儲存服務使用者指南中的將物件上傳至 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，將 的值bucket變更為 S3 儲存貯體，將 的值roleArn變更為指派給 Textract 角色的 Arn。您也需要在 Amazon S3 儲存貯體中將 的值document變更為多頁文件的名稱。最後，以您區域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()
