Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.
Gestion des appels restreintés et des connexions abandonnées
Une opération Amazon Textract peut échouer si vous dépassez le nombre maximal de transactions par seconde (TPS), ce qui entraîne une limitation du service pour votre application ou lorsque votre connexion est interrompue. Par exemple, si vous passez trop d'appels à des opérations Amazon Textract en peu de temps, il limite vos appels et envoie unProvisionedThroughputExceededExceptionerreur dans la réponse de l'opération. Pour obtenir des informations sur les quotas TPS Amazon Textract, veuillez consulterQuotas Amazon Textract.
Vous pouvez gérer la limitation et les connexions abandonnées en réessayant automatiquement l'opération. Vous pouvez spécifier le nombre de nouvelles tentatives en incluant leConfigparamètre lorsque vous créez le client Amazon Textract. Nous recommandons un nombre de nouvelles tentatives de 5. LeAWSLe kit SDK tente à nouveau une opération autant de fois qu'indiqué avant d'échouer et de lever une exception. Pour plus d'informations, consultez la page Nouvelles tentatives après erreur et interruptions exponentielles dans AWS.
Les nouvelles tentatives automatiques fonctionnent à la fois pour les opérations synchrone et asynchrone. Avant de spécifier des tentatives automatiques, assurez-vous que vous disposez de la version la plus récente du kit AWS SDK. Pour plus d'informations, consultez Étape 2 : Configuration de l'AWS CLIetAWSKits SDK.
L'exemple suivant montre comment réessayer automatiquement les opérations Amazon Textract lorsque vous traitez plusieurs documents.
Pour relancer automatiquement les opérations
-
Chargez plusieurs images de documents dans votre compartiment S3 pour exécuter l'exemple Synchrone. Chargez un document de plusieurs pages dans votre compartiment S3 et exécutezStartDocumentTextDetectiondessus pour exécuter l'exemple asynchrone.
Pour obtenir des instructions, consultezChargement d'objets dans Amazon S3dans leGuide de l'utilisateur Amazon Simple Storage Service.
-
Les exemples suivants montrent comment utiliser leConfigpour relancer automatiquement une opération. L'exemple synchrone appelle leDetectDocumentText, tandis que l'exemple asynchrone appelle l'optionGetDocumentTextDetection.
- Sync Example
-
Utilisez les exemples suivants pour appeler leDetectDocumentTextopération sur les documents de votre compartiment Amazon S3. Dansmain, modifiez la valeur debucketdans votre compartiment S3. Modifiez la valeur dedocumentsaux noms des images de document que vous avez téléchargées à l'étape 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
-
Utilisez les exemples suivants pour appeler l'opération GetDocumentTextDetection. Il suppose que vous avez déjà appeléStartDocumentTextDetectionsur les documents de votre compartiment Amazon S3 et obtenu unJobId. Dansmain, modifiez la valeur debucketà votre compartiment S3 et à la valeur deroleArnà l'arn affecté à votre rôle Textract. Vous devez également modifier la valeur dedocumentau nom de votre document multi-pages dans votre compartiment Amazon S3. Enfin, remplacez la valeur deregion_nameavec le nom de votre région et fournissez leGetResultsavec le nom de votrejobId.
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()
