AWS Textract Integration for Document Data Extraction

Integration AWS Textract для извлечения данных из документов

Amazon Textract — сервис AWS для OCR и структурированного извлечения данных из документов. Ключевое отличие от обычного OCR: Textract умеет извлекать не просто текст, а структурированные данные — таблицы, формы с ключ-значение парами, специализированные документы (удостоверения, медицинские направления, налоговые формы США). Работает через API без обучения под конкретный формат документа.

Базовая интеграция через Boto3

import boto3
import json

class AWSTextractExtractor:
    def __init__(self, region: str = 'us-east-1'):
        self.client = boto3.client('textract', region_name=region)

    def extract_from_file(self, image_path: str,
                           feature_types: list = None) -> dict:
        """Синхронная обработка локального файла (до 10MB, 1 страница)"""
        if feature_types is None:
            feature_types = ['TABLES', 'FORMS']

        with open(image_path, 'rb') as f:
            response = self.client.analyze_document(
                Document={'Bytes': f.read()},
                FeatureTypes=feature_types
            )

        return self._parse_response(response)

    def extract_from_s3(self, bucket: str, key: str) -> str:
        """Асинхронная обработка из S3 (для больших файлов и PDF)"""
        response = self.client.start_document_text_detection(
            DocumentLocation={
                'S3Object': {'Bucket': bucket, 'Name': key}
            }
        )
        job_id = response['JobId']

        # Ожидание завершения
        import time
        while True:
            result = self.client.get_document_text_detection(JobId=job_id)
            if result['JobStatus'] in ['SUCCEEDED', 'FAILED']:
                break
            time.sleep(2)

        if result['JobStatus'] == 'FAILED':
            raise RuntimeError(f"Textract job failed: {result['StatusMessage']}")

        # Объединяем страницы
        pages = [result]
        while 'NextToken' in result:
            result = self.client.get_document_text_detection(
                JobId=job_id, NextToken=result['NextToken']
            )
            pages.append(result)

        return self._extract_text_from_pages(pages)

    def _parse_response(self, response: dict) -> dict:
        blocks = {block['Id']: block for block in response['Blocks']}

        # Извлечение форм (KEY_VALUE_SET)
        forms = {}
        for block in response['Blocks']:
            if block['BlockType'] == 'KEY_VALUE_SET' and 'KEY' in block.get('EntityTypes', []):
                key_text = self._get_text(block, blocks)
                value_block = self._get_value_block(block, blocks)
                if value_block:
                    value_text = self._get_text(value_block, blocks)
                    forms[key_text] = value_text

        # Извлечение таблиц
        tables = self._extract_tables(response['Blocks'], blocks)

        # Весь текст
        lines = [b['Text'] for b in response['Blocks']
                 if b['BlockType'] == 'LINE']

        return {
            'text': '\n'.join(lines),
            'forms': forms,
            'tables': tables
        }

Специализированные модели: Analyze ID

def extract_id_document(self, image_path: str) -> dict:
    """Специализированное извлечение из удостоверений личности"""
    with open(image_path, 'rb') as f:
        response = self.client.analyze_id(
            DocumentPages=[{'Bytes': f.read()}]
        )

    result = {}
    for doc in response['IdentityDocuments']:
        for field in doc['IdentityDocumentFields']:
            field_type = field['Type']['Text']
            field_value = field['ValueDetection']['Text']
            confidence = field['ValueDetection']['Confidence']
            result[field_type] = {
                'value': field_value,
                'confidence': confidence
            }

    return result

# Пример возвращаемого результата:
# {
#   'FIRST_NAME': {'value': 'John', 'confidence': 99.5},
#   'LAST_NAME': {'value': 'Doe', 'confidence': 99.2},
#   'DATE_OF_BIRTH': {'value': '01/15/1990', 'confidence': 98.7},
#   'DOCUMENT_NUMBER': {'value': 'A12345678', 'confidence': 99.8}
# }

Стоимость и лимиты

Лимиты: синхронный API — до 10MB, 1 страница JPEG/PNG. Асинхронный (StartDocumentAnalysis) — до 500MB PDF.

Textract Queries: извлечение кастомных полей

response = self.client.analyze_document(
    Document={'Bytes': content},
    FeatureTypes=['QUERIES'],
    QueriesConfig={
        'Queries': [
            {'Text': 'Какова сумма к оплате?', 'Alias': 'total_due'},
            {'Text': 'Каков номер счёта?', 'Alias': 'invoice_number'},
            {'Text': 'Кто является поставщиком?', 'Alias': 'vendor'}
        ]
    }
)

Queries работают на естественном языке и позволяют извлекать поля не в стандартных форматах.