Поделиться
Поделиться

Каждый день компании получают тысячи документов: счета-фактуры, акты, договоры, паспорта, медицинские справки. Ручной ввод данных — дорого и медленно. OCR (Optical Character Recognition) решает эту задачу автоматически, но «просто поставить Tesseract» недостаточно: без правильной предобработки точность распознавания падает до неприемлемых значений, а таблицы превращаются в нечитаемую мешанину.

В этой статье мы построим промышленный пайплайн: от сырого скана до структурированного JSON с обработкой ошибок и мониторингом качества.

Сравнение движков: Tesseract, PaddleOCR и облачные сервисы

Прежде чем писать код, выберем инструмент. Ни один движок не выигрывает по всем параметрам.

| Критерий | Tesseract 5 | PaddleOCR | Google Vision API | AWS Textract | |---|---|---|---|---| | Русский язык | Хорошо | Отлично | Отлично | Хорошо | | Таблицы | Плохо | Средне | Хорошо | Отлично | | Рукописный текст | Нет | Средне | Хорошо | Средне | | Стоимость | Бесплатно | Бесплатно | $1.5/1000 стр. | $1.5–15/1000 стр. | | Self-hosted | Да | Да | Нет | Нет | | Latency | ~1 с | ~0.3 с (GPU) | ~0.8 с | ~1.5 с |

Вывод: для прототипов и небольших объёмов — Tesseract. Для production на русском языке без GPU — PaddleOCR CPU. Для сложных таблиц и финансовых документов — AWS Textract.

Предобработка: от чего зависит 80% качества

Плохо отсканированный документ не спасёт ни один движок. Нужна препроцессинговая цепочка.

import cv2
import numpy as np
from PIL import Image
import io

def preprocess_document(image_path: str) -> np.ndarray:
    """
    Полный пайплайн предобработки: выравнивание, шумоподавление, бинаризация.
    """
    img = cv2.imread(image_path)
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)

    # 1. Deskew — выравнивание наклона
    coords = np.column_stack(np.where(gray < 128))
    angle = cv2.minAreaRect(coords)[-1]
    if angle < -45:
        angle = -(90 + angle)
    else:
        angle = -angle

    (h, w) = gray.shape[:2]
    center = (w // 2, h // 2)
    M = cv2.getRotationMatrix2D(center, angle, 1.0)
    rotated = cv2.warpAffine(
        gray, M, (w, h),
        flags=cv2.INTER_CUBIC,
        borderMode=cv2.BORDER_REPLICATE
    )

    # 2. Denoise — подавление шума
    denoised = cv2.fastNlMeansDenoising(rotated, h=10)

    # 3. Adaptive binarization — лучше глобального порога
    binary = cv2.adaptiveThreshold(
        denoised, 255,
        cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
        cv2.THRESH_BINARY, 31, 15
    )

    # 4. Морфологическая очистка мелких артефактов
    kernel = np.ones((1, 1), np.uint8)
    cleaned = cv2.morphologyEx(binary, cv2.MORPH_OPEN, kernel)

    return cleaned

Upscaling для низкого разрешения

Tesseract даёт плохие результаты при разрешении ниже 200 DPI. Увеличение через cv2.resize с INTER_CUBIC до 300 DPI перед распознаванием часто поднимает точность на 15–30%.

def ensure_dpi(img: np.ndarray, target_dpi: int = 300, source_dpi: int = 72) -> np.ndarray:
    if source_dpi >= target_dpi:
        return img
    scale = target_dpi / source_dpi
    h, w = img.shape[:2]
    return cv2.resize(img, (int(w * scale), int(h * scale)), interpolation=cv2.INTER_CUBIC)

Распознавание с Tesseract

import pytesseract
from pytesseract import Output

def ocr_with_tesseract(img: np.ndarray, lang: str = "rus+eng") -> dict:
    # Конфиг: psm 6 = единый блок текста, oem 3 = LSTM + legacy
    config = "--psm 6 --oem 3 -c preserve_interword_spaces=1"
    
    data = pytesseract.image_to_data(
        img, lang=lang, config=config,
        output_type=Output.DICT
    )
    
    # Фильтруем слова с низкой уверенностью
    words = []
    for i, conf in enumerate(data["conf"]):
        if int(conf) > 60:  # порог уверенности
            words.append({
                "text": data["text"][i],
                "conf": conf,
                "bbox": (
                    data["left"][i], data["top"][i],
                    data["width"][i], data["height"][i]
                )
            })
    
    full_text = pytesseract.image_to_string(img, lang=lang, config=config)
    avg_conf = np.mean([int(c) for c in data["conf"] if int(c) > 0])
    
    return {
        "text": full_text,
        "words": words,
        "avg_confidence": round(avg_conf, 2)
    }

PaddleOCR: быстрее и точнее для кириллицы

from paddleocr import PaddleOCR
import numpy as np

# Инициализация один раз, не при каждом вызове
paddle_ocr = PaddleOCR(
    use_angle_cls=True,
    lang="ru",
    use_gpu=False,          # True если есть GPU
    enable_mkldnn=True,     # ускорение на CPU
    det_db_score_mode="slow"  # точнее, но медленнее
)

def ocr_with_paddle(image_path: str) -> dict:
    result = paddle_ocr.ocr(image_path, cls=True)
    
    lines = []
    confidences = []
    
    for line in result[0]:
        bbox, (text, conf) = line
        lines.append({"text": text, "confidence": conf, "bbox": bbox})
        confidences.append(conf)
    
    full_text = "\n".join(item["text"] for item in lines)
    
    return {
        "text": full_text,
        "lines": lines,
        "avg_confidence": round(np.mean(confidences), 4) if confidences else 0
    }

Извлечение таблиц

Таблицы — самая сложная часть. Tesseract и PaddleOCR дают текст без структуры; для структурированного извлечения нужен отдельный подход.

Подход 1: Детекция линий через OpenCV

def extract_table_structure(img: np.ndarray) -> list[list[str]]:
    """
    Находит ячейки таблицы по горизонтальным и вертикальным линиям.
    """
    # Бинаризация
    _, thresh = cv2.threshold(img, 128, 255, cv2.THRESH_BINARY_INV)
    
    # Горизонтальные линии
    h_kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (40, 1))
    h_lines = cv2.morphologyEx(thresh, cv2.MORPH_OPEN, h_kernel)
    
    # Вертикальные линии
    v_kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (1, 40))
    v_lines = cv2.morphologyEx(thresh, cv2.MORPH_OPEN, v_kernel)
    
    # Комбинируем
    grid = cv2.add(h_lines, v_lines)
    
    # Находим контуры ячеек
    contours, _ = cv2.findContours(
        grid, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE
    )
    
    cells = []
    for c in contours:
        x, y, w, h = cv2.boundingRect(c)
        if w > 30 and h > 15:  # фильтруем мусор
            cell_img = img[y:y+h, x:x+w]
            text = pytesseract.image_to_string(
                cell_img, lang="rus+eng",
                config="--psm 7"  # одна строка
            ).strip()
            cells.append({"bbox": (x, y, w, h), "text": text})
    
    return cells

Подход 2: AWS Textract для сложных таблиц

import boto3
import json

def extract_table_textract(image_bytes: bytes) -> list[dict]:
    client = boto3.client("textract", region_name="eu-west-1")
    
    response = client.analyze_document(
        Document={"Bytes": image_bytes},
        FeatureTypes=["TABLES", "FORMS"]
    )
    
    blocks = {b["Id"]: b for b in response["Blocks"]}
    tables = []
    
    for block in response["Blocks"]:
        if block["BlockType"] != "TABLE":
            continue
        
        table_rows = {}
        for rel in block.get("Relationships", []):
            if rel["Type"] == "CHILD":
                for cell_id in rel["Ids"]:
                    cell = blocks[cell_id]
                    if cell["BlockType"] == "CELL":
                        row = cell["RowIndex"]
                        col = cell["ColumnIndex"]
                        
                        # Извлекаем текст из ячейки
                        cell_text = ""
                        for word_rel in cell.get("Relationships", []):
                            if word_rel["Type"] == "CHILD":
                                for word_id in word_rel["Ids"]:
                                    word = blocks[word_id]
                                    if word["BlockType"] == "WORD":
                                        cell_text += word["Text"] + " "
                        
                        if row not in table_rows:
                            table_rows[row] = {}
                        table_rows[row][col] = cell_text.strip()
        
        tables.append(table_rows)
    
    return tables

PDF → структурированные данные: полный пайплайн

import fitz  # PyMuPDF
from dataclasses import dataclass, asdict
from typing import Optional
import logging

logger = logging.getLogger(__name__)

@dataclass
class DocumentResult:
    source_file: str
    pages: int
    text: str
    tables: list
    avg_confidence: float
    errors: list[str]

def pdf_to_structured(
    pdf_path: str,
    engine: str = "paddle",  # или "tesseract"
    dpi: int = 200
) -> DocumentResult:
    doc = fitz.open(pdf_path)
    all_text = []
    all_tables = []
    confidences = []
    errors = []
    
    for page_num in range(len(doc)):
        try:
            page = doc[page_num]
            
            # Рендерим страницу в растр
            mat = fitz.Matrix(dpi / 72, dpi / 72)
            pix = page.get_pixmap(matrix=mat, alpha=False)
            img_bytes = pix.tobytes("png")
            
            # Предобработка
            img_array = np.frombuffer(img_bytes, np.uint8)
            img = cv2.imdecode(img_array, cv2.IMREAD_GRAYSCALE)
            processed = preprocess_document_from_array(img)
            
            # OCR
            if engine == "paddle":
                result = ocr_with_paddle_bytes(processed)
            else:
                result = ocr_with_tesseract(processed)
            
            all_text.append(f"--- Page {page_num + 1} ---\n{result['text']}")
            confidences.append(result.get("avg_confidence", 0))
            
            # Пытаемся извлечь таблицы
            table_cells = extract_table_structure(processed)
            if table_cells:
                all_tables.append({
                    "page": page_num + 1,
                    "cells": table_cells
                })
                
        except Exception as e:
            error_msg = f"Page {page_num + 1}: {str(e)}"
            logger.error(error_msg)
            errors.append(error_msg)
    
    doc.close()
    
    return DocumentResult(
        source_file=pdf_path,
        pages=len(doc),
        text="\n\n".join(all_text),
        tables=all_tables,
        avg_confidence=round(np.mean(confidences), 4) if confidences else 0,
        errors=errors
    )

Обработка ошибок в production

В реальных проектах качество скана нестабильно. Нужна система оценки качества и fallback-стратегия.

from enum import Enum

class OCRQuality(Enum):
    HIGH = "high"       # > 85% confidence
    MEDIUM = "medium"   # 60–85%
    LOW = "low"         # < 60% — отправляем на ручную проверку

def assess_quality(result: dict) -> OCRQuality:
    conf = result.get("avg_confidence", 0)
    text = result.get("text", "")
    
    # Дополнительные эвристики
    words = text.split()
    if len(words) < 10:
        return OCRQuality.LOW
    
    # Доля "мусорных" символов
    garbage_ratio = sum(1 for c in text if c in "©®†‡§¶") / max(len(text), 1)
    if garbage_ratio > 0.02:
        return OCRQuality.LOW
    
    if conf > 85:
        return OCRQuality.HIGH
    elif conf > 60:
        return OCRQuality.MEDIUM
    return OCRQuality.LOW

async def process_with_fallback(
    image_path: str,
    primary_engine: str = "paddle"
) -> dict:
    # Первичный движок
    result = await run_ocr(image_path, engine=primary_engine)
    quality = assess_quality(result)
    
    if quality == OCRQuality.HIGH:
        return {**result, "quality": quality.value, "engine": primary_engine}
    
    if quality == OCRQuality.MEDIUM:
        # Пробуем второй движок
        result2 = await run_ocr(image_path, engine="tesseract")
        if assess_quality(result2) == OCRQuality.HIGH:
            return {**result2, "quality": "high", "engine": "tesseract_fallback"}
        # Возвращаем лучший из двух
        best = result if result["avg_confidence"] > result2["avg_confidence"] else result2
        return {**best, "quality": quality.value, "needs_review": False}
    
    # LOW quality: отправляем в очередь ручной проверки
    await queue_for_manual_review(image_path, result)
    return {**result, "quality": quality.value, "needs_review": True}

Мониторинг и метрики

Отслеживайте ключевые показатели OCR-пайплайна:

  • Средняя уверенность по движкам и типам документов
  • Доля документов на ручную проверку (цель: < 5%)
  • Время обработки одной страницы
  • Количество ошибок по типам (timeout, низкое DPI, повреждённый файл)

Интегрируйте с Prometheus/Grafana через счётчики:

from prometheus_client import Counter, Histogram, Gauge

ocr_processed = Counter("ocr_documents_total", "Total processed", ["engine", "quality"])
ocr_latency = Histogram("ocr_processing_seconds", "Processing time", ["engine"])
ocr_confidence = Gauge("ocr_avg_confidence", "Average confidence score", ["engine"])

Итог

Хороший OCR-пайплайн строится послойно: сначала надёжная предобработка, затем подходящий движок, потом оценка качества с fallback. Для большинства русскоязычных документов оптимальна связка PaddleOCR + ручная проверка при уверенности ниже 60%. Для финансовых таблиц стоит рассмотреть AWS Textract — его дополнительная стоимость окупается сохранённым временем операторов.

Подробнее про автоматизацию рутинных задач на Python читайте в статье Автоматизация с AI, а про оптимизацию Docker-образов для ML-сервисов — в Docker: оптимизация образов.