View a markdown version of this page

Gestion des erreurs de connexion - Amazon Textract

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 erreurs de connexion

Une opération Amazon Textract peut échouer si vous dépassez le nombre maximum de transactions par seconde (TPS), ce qui ralentit le service pour votre application ou si votre connexion est interrompue. Par exemple, si vous passez trop d'appels vers Amazon Textract en peu de temps, cela limite vos appels et envoie une ProvisionedThroughputExceededException erreur dans la réponse à l'opération. Pour plus d'informations sur les quotas TPS Amazon Textract, consultez Amazon Textract Quotas. Pour modifier une limite, vous pouvez accéder à l'option Amazon Textract dans la Service Quotas console.

Vous pouvez gérer le ralentissement et les interruptions de connexion en réessayant automatiquement l'opération. Vous pouvez spécifier le nombre de tentatives en incluant le Config paramètre lorsque vous créez le client Amazon Textract. Nous recommandons un nombre de tentatives de 5. Le AWS SDK réessaie une opération le nombre de fois spécifié avant d'échouer et de lancer une exception. Pour plus d'informations, consultez la page Nouvelles tentatives après erreur et interruptions exponentielles dans AWS.

Note

Les tentatives automatiques fonctionnent à la fois pour les opérations synchrones et asynchrones. Avant de spécifier des tentatives automatiques, assurez-vous de disposer de la version la plus récente du SDK AWS. Pour de plus amples informations, veuillez consulter Étape 2 : configurer le AWS CLI and AWS Kits SDK.

L'exemple suivant montre comment réessayer automatiquement les opérations Amazon Textract lorsque vous traitez plusieurs documents.

Conditions préalables
Pour réessayer automatiquement les opérations

  1. Téléchargez plusieurs images de documents dans votre compartiment S3 pour exécuter l'exemple synchrone. Téléchargez un document de plusieurs pages dans votre compartiment S3 et exécutez-le StartDocumentTextDetection pour exécuter l'exemple asynchrone.

    Pour en savoir plus, consultez Chargement d’objets dans Amazon S3 dans le Guide de l’utilisateur Amazon Simple Storage Service.

  2. Les exemples suivants montrent comment utiliser le Config paramètre pour réessayer automatiquement une opération. L'exemple synchrone appelle l'DetectDocumentTextopération, tandis que l'exemple asynchrone appelle l'opération. GetDocumentTextDetection

    Sync Example

    Utilisez les exemples suivants pour appeler l'DetectDocumentTextopération sur les documents de votre compartiment Amazon S3. Dansmain, remplacez la valeur bucket de par votre compartiment S3. Modifiez la valeur de par documents les noms des images du 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. Cela suppose que vous avez déjà appelé StartDocumentTextDetection les documents de votre compartiment Amazon S3 et obtenu unJobId. Dansmain, modifiez la valeur bucket de votre compartiment S3 et la valeur de l'roleArnArn attribué à votre rôle Textract. Vous devrez également remplacer la valeur de par le nom document de votre document de plusieurs pages dans votre compartiment Amazon S3. Enfin, remplacez la valeur de region_name par le nom de votre région et fournissez à la GetResults fonction 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()