Google Cloud Vision OCR Integration

Integration Google Cloud Vision OCR

Google Cloud Vision API — управляемый OCR-сервис с высоким качеством распознавания и поддержкой 50+ языков. Два режима: TEXT_DETECTION (OCR для изображений) и DOCUMENT_TEXT_DETECTION (оптимизирован для плотных документов с сохранением структуры). Не требует инфраструктуры, платная модель на основе количества запросов.

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

from google.cloud import vision
from google.oauth2 import service_account
import io

class GoogleVisionOCR:
    def __init__(self, credentials_path: str):
        credentials = service_account.Credentials.from_service_account_file(
            credentials_path
        )
        self.client = vision.ImageAnnotatorClient(credentials=credentials)

    def extract_text(self, image_path: str) -> str:
        with io.open(image_path, 'rb') as image_file:
            content = image_file.read()

        image = vision.Image(content=content)
        response = self.client.text_detection(image=image)

        if response.error.message:
            raise RuntimeError(f'Vision API error: {response.error.message}')

        return response.text_annotations[0].description if response.text_annotations else ''

    def extract_document(self, image_path: str) -> dict:
        """DOCUMENT_TEXT_DETECTION для структурированных документов"""
        with io.open(image_path, 'rb') as f:
            content = f.read()

        image = vision.Image(content=content)
        response = self.client.document_text_detection(image=image)
        document = response.full_text_annotation

        pages_data = []
        for page in document.pages:
            page_text = ''
            blocks = []
            for block in page.blocks:
                block_text = ''
                for paragraph in block.paragraphs:
                    para_text = ' '.join(
                        ''.join(s.text for s in word.symbols)
                        for word in paragraph.words
                    )
                    block_text += para_text + '\n'
                blocks.append({'text': block_text.strip()})
                page_text += block_text

            pages_data.append({'text': page_text, 'blocks': blocks})

        return {'full_text': document.text, 'pages': pages_data}

Батчевая обработка и оптимизация стоимости

import base64
from google.cloud import vision_v1

def batch_process_gcs(gcs_uris: list[str],
                       output_gcs_prefix: str,
                       credentials_path: str):
    """Асинхронная batch обработка через Google Cloud Storage — дешевле"""
    client = vision_v1.ImageAnnotatorClient.from_service_account_file(
        credentials_path
    )

    requests = []
    for uri in gcs_uris:
        source = vision_v1.ImageSource(gcs_image_uri=uri)
        image = vision_v1.Image(source=source)
        feature = vision_v1.Feature(type_=vision_v1.Feature.Type.DOCUMENT_TEXT_DETECTION)
        requests.append(vision_v1.AnnotateImageRequest(
            image=image, features=[feature]
        ))

    # Batch request — обрабатывает до 2000 изображений асинхронно
    gcs_dest = vision_v1.GcsDestination(uri=output_gcs_prefix)
    output_config = vision_v1.OutputConfig(
        gcs_destination=gcs_dest,
        batch_size=100  # файлы результатов по 100 страниц
    )

    operation = client.async_batch_annotate_images(
        requests=requests[:2000],
        output_config=output_config
    )
    return operation

Работа с PDF через Cloud Vision

def process_pdf(pdf_gcs_uri: str, output_gcs_prefix: str, client):
    """OCR PDF-файлов через Cloud Vision"""
    feature = vision_v1.Feature(
        type_=vision_v1.Feature.Type.DOCUMENT_TEXT_DETECTION
    )
    gcs_source = vision_v1.GcsSource(uri=pdf_gcs_uri)
    input_config = vision_v1.InputConfig(
        gcs_source=gcs_source,
        mime_type='application/pdf'
    )
    gcs_dest = vision_v1.GcsDestination(uri=output_gcs_prefix)
    output_config = vision_v1.OutputConfig(
        gcs_destination=gcs_dest, batch_size=10
    )

    request = vision_v1.AsyncAnnotateFileRequest(
        features=[feature],
        input_config=input_config,
        output_config=output_config
    )
    operation = client.async_batch_annotate_files(requests=[request])
    return operation

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

Optimization стоимости: использовать DOCUMENT_TEXT_DETECTION только для документов (качественнее, но та же цена), батчевая обработка через GCS для больших объёмов.

Когда выбирать Cloud Vision vs PaddleOCR