연결 오류 처리

사전 조건

• 아직 설정하지 않았다면 다음과 같이 하세요.

  1. 사용자에게 AmazonTextractFullAccess 및 AmazonS3ReadOnlyAccess 권한을 부여합니다. 자세한 내용은 1단계: AWS 계정 설정 및 사용자 만들기 단원을 참조하십시오.

  2. AWS CLI 및 AWS SDKs를 설치하고 구성합니다. 자세한 내용은 2단계: AWS CLI 및 AWS SDKs 설정 단원을 참조하십시오.

작업을 자동으로 재시도하려면

1. S3 버킷에 여러 문서 이미지를 업로드하여 동기 예제를 실행합니다. S3 버킷에 여러 페이지 문서를 업로드하고 StartDocumentTextDetection 실행하여 비동기 예제를 실행합니다.

   이에 관한 지침은 Amazon Simple Storage Service 사용 설명서에서 Amazon S3에 객체 업로드를 참조하세요.

2. 다음 예제에서는 Config 파라미터를 사용하여 작업을 자동으로 재시도하는 방법을 보여줍니다. 동기 예제는 DetectDocumentText 작업을 호출하고 비동기 예제는 GetDocumentTextDetection 작업을 호출합니다.

   Sync Example

   다음 예제를 사용하여 Amazon S3 버킷의 문서에 대한 DetectDocumentText 작업을 호출합니다. 에서의 값을 S3 버킷bucket으로 main변경합니다. 의 값을 2단계에서 업로드한 문서 이미지의 documents 이름으로 변경합니다.

   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 작업을 호출합니다. Amazon S3 버킷의 문서에서 StartDocumentTextDetection를 이미 호출하고를 얻었다고 가정합니다JobId. 에서의 값을 S3 버킷bucket으로,의 값을 Textract 역할에 할당된 roleArn Arn으로 main변경합니다. 또한의 값을 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()