View a markdown version of this page

Menangani Kesalahan Koneksi - Amazon Textract

Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Menangani Kesalahan Koneksi

Operasi Amazon Textract dapat gagal jika Anda melebihi jumlah maksimum transaksi per detik (TPS), menyebabkan layanan menghambat aplikasi Anda, atau ketika koneksi Anda turun. Misalnya, jika Anda melakukan terlalu banyak panggilan ke operasi Amazon Textract dalam waktu singkat, itu membatasi panggilan Anda dan mengirimkan ProvisionedThroughputExceededException kesalahan dalam respons operasi. Untuk informasi tentang kuota TPS Amazon Textract, lihat Kuota Amazon Textract. Untuk mengubah batas, Anda dapat mengakses opsi Amazon Textract di konsol. Service Quotas

Anda dapat mengelola pelambatan dan koneksi terputus dengan mencoba kembali operasi secara otomatis. Anda dapat menentukan jumlah percobaan ulang dengan menyertakan Config parameter saat Anda membuat klien Amazon Textract. Kami merekomendasikan hitungan coba lagi 5. AWS SDK mencoba ulang operasi beberapa kali yang ditentukan sebelum gagal dan melempar pengecualian. Untuk informasi selengkapnya, lihat Error Retries dan Exponential Backoff di AWS.

catatan

Percobaan ulang otomatis berfungsi untuk operasi sinkron dan asinkron. Sebelum menentukan percobaan ulang otomatis, pastikan Anda memiliki AWS SDK versi terbaru. Untuk informasi selengkapnya, lihat Langkah 2: Siapkan AWS CLI and AWS SDK.

Contoh berikut menunjukkan cara mencoba ulang operasi Amazon Textract secara otomatis saat Anda memproses beberapa dokumen.

Prasyarat
Untuk secara otomatis mencoba kembali operasi

  1. Unggah beberapa gambar dokumen ke bucket S3 Anda untuk menjalankan contoh Synchronous. Unggah dokumen multi-halaman ke bucket S3 Anda dan jalankan StartDocumentTextDetection di atasnya untuk menjalankan contoh Asynchronous.

    Untuk petunjuk, lihat Mengunggah Objek ke Amazon S3 di Panduan Pengguna Layanan Penyimpanan Sederhana Amazon.

  2. Contoh berikut menunjukkan cara menggunakan Config parameter untuk mencoba kembali operasi secara otomatis. Contoh Synchronous memanggil DetectDocumentText operasi, sedangkan contoh Asynchronous memanggil operasi. GetDocumentTextDetection

    Sync Example

    Gunakan contoh berikut untuk memanggil DetectDocumentText operasi pada dokumen di bucket Amazon S3 Anda. Dimain, ubah nilai bucket ke bucket S3 Anda. Ubah nilai documents menjadi nama gambar dokumen yang Anda unggah di langkah 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

    Gunakan contoh berikut untuk memanggil operasi GetDocumentTextDetection. Ini mengasumsikan Anda telah memanggil dokumen StartDocumentTextDetection di ember Amazon S3 Anda dan memperoleh file. JobId Dimain, ubah nilai bucket bucket S3 Anda dan nilai Arn yang ditetapkan roleArn ke peran Textract Anda. Anda juga harus mengubah nilai ke nama dokumen multi-halaman Anda di bucket Amazon S3 Anda. document Terakhir, ganti nilai region_name dengan nama wilayah Anda dan berikan GetResults fungsi dengan nama AndajobId.

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