Документация Proton OCR API¶

Добро пожаловать в документацию Протона — высокопроизводительного OCR движка для автоматизации распознавания текста в ваших приложениях и сервисах.

Что это такое¶

Протон предоставляет RESTful API для интеграции функций распознавания текста (OCR) в цифровые продукты, корпоративные системы и исследовательские проекты. Сервис поддерживает как печатный, так и рукописный текст, обеспечивая высокую точность и скорость обработки.

Содержание¶

Быстрый старт¶

Ниже приведён базовый пример работы с API на Python. Предполагается, что у вас уже есть API-токен.

import base64
import pathlib
import time
from typing import Literal

import requests


API_KEY = "ВАШ-API-КЛЮЧ"
IMAGES = ['ПУТЬ(И) К ИЗОБРАЖЕНИЮ(ЯМ)']

POLL_INTERVAL_S = 0.5
MAX_POLLS = 10

BASE_URL = "https://ocrbot.ru/api/v1"
Headers = {"X-Api-Key": API_KEY}


def encode_image(path: str | pathlib.Path) -> str:
    with open(path, "rb") as f:
        return base64.b64encode(f.read()).decode()


def decode_result(data: str) -> bytes:
    return base64.b64decode(data)


def save_result(data: bytes, path: str | pathlib.Path) -> None:
    with open(path, "wb") as f:
        f.write(data)


def create_task(images_b64: str) -> str:
    resp = requests.post(
        f"{BASE_URL}/tasks",
        headers=Headers,
        json={"image": images_b64, "return_type": "json"},
        timeout=30,
    )
    resp.raise_for_status()
    task_id = resp.json().get("task_id")
    if not task_id:
        raise RuntimeError("Не удалось получить task_id")
    print(f"Создана задача: {task_id}")
    return task_id


def wait_for_task(
    task_id: str,
    max_polls: int = MAX_POLLS,
    interval_s: float = POLL_INTERVAL_S,
) -> Literal["success", "error"]:
    for _ in range(max_polls):
        status_resp = requests.get(
            f"{BASE_URL}/tasks/{task_id}/status",
            headers=Headers,
            timeout=10,
        )
        status_resp.raise_for_status()
        status = status_resp.json().get("task_status", "unknown")
        print(f"статус: {status}")

        if status in ("pending"):
            time.sleep(interval_s)
        elif status in ("success", "error"):
            return status
        else:
            raise RuntimeError(f"{status}")
    raise TimeoutError("Лимит ожидания статуса исчерпан")


def fetch_result(task_id: str) -> str:
    resp = requests.get(
        f"{BASE_URL}/tasks/{task_id}/result",
        headers=Headers,
        timeout=30,
    )
    resp.raise_for_status()
    print(resp.json().keys())
    return resp.json().get("recognition_result", "")


def main() -> None:
    images_b64 = [encode_image(img) for img in IMAGES]
    task_id = create_task(images_b64)

    status = wait_for_task(task_id)
    if status == "success":
        data = fetch_result(task_id)
        decoded = [decode_result(elem) for elem in data]
        for i, page in enumerate(decoded):
            save_result(page, f"page_{i}.json")
    else:
        print("Распознавание завершилось ошибкой")


if __name__ == "__main__":
    main()