Gestione di chiamate limitate e connessioni interrotte

Un'operazione Amazon Textract può non riuscire se superi il numero massimo di transazioni al secondo (TPS), causando la limitazione del servizio dell'applicazione o quando la connessione viene interrotta. Ad esempio, se effettui troppe chiamate alle operazioni di Amazon Textract in un breve periodo di tempo, questo riduce le chiamate e invia unProvisionedThroughputExceededExceptionerrore nella risposta dell'operazione. Per informazioni sulle quote Amazon Textract TPS, consultaQuote Amazon Textract.

È possibile gestire la limitazione e l'interruzione delle connessioni ripetendo automaticamente l'operazione. È possibile specificare il numero di tentativi includendo ilConfigparametro quando crei il client Amazon Textract. Raccomandiamo un numero di tentativi di 5. LaAWSSDK tenta un'operazione il numero di volte specificato prima di fallire e lanciare un'eccezione. Per ulteriori informazioni, consulta Ripetizione dei tentativi in caso di errore e backoff esponenziale in AWS.

Nota

I tentativi automatici funzionano sia per operazioni sincrone che asincrone. Prima di specificare tentativi automatici, assicurati di avere la versione più recente dell'SDK AWS. Per ulteriori informazioni, consultare Fase 2: Configurazione diAWS CLIeAWSSDK.

Nell'esempio seguente viene illustrato come riprovare automaticamente le operazioni Amazon Textract durante l'elaborazione di più documenti.

Prerequisiti

• Se non lo hai già fatto:

  1. Creare o aggiornare un utente IAM conAmazonTextractFullAccesseAmazonS3ReadOnlyAccessautorizzazioni. Per ulteriori informazioni, consultare Fase 1: Impostazione di un account AWS e creazione di un utente IAM.

  2. Installa e configura la AWS CLI e gli SDK AWS. Per ulteriori informazioni, consultare Fase 2: Configurazione diAWS CLIeAWSSDK.

Per riprovare automaticamente le operazioni

1. Caricare più immagini di documenti nel bucket S3 per eseguire l'esempio sincrono. Caricare un documento multipagina nel bucket S3 ed eseguireStartDocumentTextDetectionsu di esso per eseguire l'esempio asincrono.

   Per istruzioni, consultaCaricamento di oggetti in Amazon S3nellaAmazon Simple Storage Service.

2. Negli esempi seguenti viene illustrato come utilizzare ilConfigparametro per riprovare automaticamente un'operazione. L'esempio sincrono chiama ilDetectDocumentTextoperazione, mentre l'esempio asincrono chiama ilGetDocumentTextDetectionoperazione.

   Sync Example

   Usare i seguenti esempi per richiamare ilDetectDocumentTextOperazioni sui documenti nel bucket Amazon S3. Nello statomain, modificare il valore dibucketnel proprio bucket S3. Modificare il valore didocumentsai nomi delle immagini del documento caricate nel passaggio 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

   Utilizza i seguenti esempi per richiamare l'operazione GetDocumentTextDetection. Presuppone che tu abbia già chiamatoStartDocumentTextDetectionsui documenti nel bucket Amazon S3 e ottenuto unJobId. Nello statomain, modificare il valore dibucketnel proprio bucket S3 e al valoreroleArnall'Arn assegnato al ruolo Textract. Dovrai anche modificare il valore didocumentnel nome del documento multipagina nel bucket Amazon S3. Infine, sostituisci il valore diregion_namecon il nome della regione e fornisciGetResultsfunzione con il nome deljobId.

   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()