gendesign/backend/app/services/scrapers/izyatie_ocr.py

"""OCR-пайплайн извлечения кадастровых номеров из PDF-сканов изъятия ЕКБ (#1062).

PDF-документы «Сообщений о планируемом изъятии…» с екатеринбург.рф — растровые сканы:
pdfplumber даёт пустой текст. Используем PyMuPDF для растеризации (без poppler) +
Tesseract rus для распознавания.

OCR-шум: сканы содержат артефакты распознавания:
  - «О» вместо «0» в кад-номере: «66:41:07О2048:26»
  - пробелы внутри кад-паттерна: «66: 41 :0702048 :26»
  - «б» вместо «6» в первых цифрах: «6б:41:...»
Нормализация выполняется ДО поиска регуляркой кад-номеров.
"""

from __future__ import annotations

import io
import logging
import re
from typing import Any

logger = logging.getLogger(__name__)

# ── Константы ──────────────────────────────────────────────────────────────────

_DPI_DEFAULT = 300
_MAX_PAGES_DEFAULT = 30
_EXCERPT_RADIUS = 100  # символов вокруг первого кад-номера

# Строгий паттерн кад-номера ПОСЛЕ нормализации текста.
# 66:41:NNNNNNN:NN — регион:район:квартал:номер. Квартал 6 или 7 цифр.
_RE_CAD_NUM = re.compile(r"\b(66:\d{2}:\d{6,7}:\d+)\b")

# Паттерн номера акта: «№ NNN от DD.MM.YYYY» или «№ NNN-ПП».
_RE_ACT_NUMBER = re.compile(
    r"№\s*(\d[\d\s]*[\-–—]?\s*(?:ПП|ПА|РП|ПГ|ГП|МО))\b",
    re.UNICODE | re.IGNORECASE,
)
_RE_ACT_DATE = re.compile(r"от\s+(\d{2})\.(\d{2})\.(\d{4})")

# Слова, по которым дата опознаётся как дата САМОГО акта-основания, а не
# ссылки на другой документ (#2464). «Сообщение о планируемом изъятии»
# открывается списком оснований, где первой строкой почти всегда стоит
# «Решение Екатеринбургской городской Думы от 06.07.2004 № 60/1 «Об
# утверждении Генерального плана города»» — Генплан, а не акт об изъятии.
# Брать первую дату подряд означало ставить всем участкам дату Генплана.
#
# Действующее основание изъятия — постановление (Администрации города об
# утверждении проекта планировки/межевания либо Правительства области),
# поэтому дата принимается, только если слово стоит в предшествующем контексте.
# Ссылки-помехи в этих документах — «Решение … Думы» и «Приказ Министерства»,
# и ни одна из них слова «постановление» не содержит.
#
# Пробовал требовать ещё и «администраци»: на пяти прод-документах результат
# тот же 5 из 5, но правило ломает законный случай «Постановление № 509-ПП»
# (областное постановление без слова «администрация») — он уже закреплён
# тестом test_act_date_extracted_from_text. Взято более широкое условие:
# на живых данных оно не хуже, а лишнего не отсекает.
_ACT_CONTEXT_WORDS = ("постановлени",)
# Ширина окна контекста. OCR перемешивает колонки таблицы, и между словами
# «Постановление Администрации города» и «от DD.MM.YYYY» вклинивается текст
# соседней колонки («…Администрации города документами) Екатеринбурга от
# 19.04.2019…»), поэтому окно шире самой фразы (~50 символов). Откалибровано
# на пяти прод-документах land_reservation: 80, 120 и 160 дают одинаковые
# 5 из 5, выбрана середина.
_ACT_CONTEXT_WINDOW = 120

# Паттерн цели: «в целях…», «для …», «под строительство …» — best-effort.
_RE_PURPOSE = re.compile(
    r"(?:для|в целях?|под)\s+([^.;,\n]{10,120})",
    re.IGNORECASE | re.UNICODE,
)


# ── Нормализация OCR-шума ──────────────────────────────────────────────────────


def _normalize_ocr_text(text: str) -> str:
    """Нормализует OCR-шум в кадастровых номерах перед поиском по регулярке.

    Стратегия:
    1. Убираем пробелы внутри цифро-двоеточных последовательностей вида «66: 41 :».
       Паттерн: пробелы после цифры или двоеточия перед цифрой или двоеточием.
    2. Заменяем «О» (кириллическая) → «0» (ноль) в позициях, окружённых цифрами/«:».
    3. Заменяем «б» → «6» только в начале кад-паттерна (если предшествует граница слова
       и после стоит «6:» или цифра с двоеточием).
    4. Убираем мягкий дефис (U+00AD) и другие невидимые разделители.

    Нормализация намеренно консервативна: применяется только в контексте,
    где риск ложной замены минимален.
    """
    # Шаг 1: убрать пробелы/переносы строк внутри «66: 41 :0702048 :26» →
    # используем итеративную замену пробелов между [\d:] и [\d:].
    # Итерируем до схождения (максимум 5 проходов — запас против вложенных пробелов).
    for _ in range(5):
        new_text = re.sub(r"(?<=[\d:])[ \t]+(?=[\d:])", "", text)
        if new_text == text:
            break
        text = new_text

    # Шаг 2: «О» (кириллическая, U+041E) → «0» между цифрами/двоеточиями.
    # Контекст: цифра/«:» ПЕРЕД и цифра/«:» ПОСЛЕ замены.
    text = re.sub(r"(?<=[\d:])О(?=[\d:])", "0", text)
    # Также в начале кад-паттерна «6б:41» → «66:41»:
    # б (кирилл.) → 6, если справа стоит «6:» (т.е. первые два символа кад-номера).
    text = re.sub(r"\bб(?=6:)", "6", text, flags=re.IGNORECASE)
    # Симметрично «6б:» → «66:» (OCR съел второй символ кад-номера МСК-66). См. #1228.
    text = re.sub(r"(?<=6)б(?=:)", "6", text, flags=re.IGNORECASE)

    # Шаг 3: мягкий дефис и нулевая ширина.
    text = text.replace("­", "").replace("​", "")

    return text


# ── OCR PDF ────────────────────────────────────────────────────────────────────


def ocr_pdf_text(
    pdf_bytes: bytes,
    *,
    dpi: int = _DPI_DEFAULT,
    max_pages: int = _MAX_PAGES_DEFAULT,
) -> str:
    """Растеризует PDF и выполняет OCR каждой страницы через Tesseract rus.

    Использует PyMuPDF (fitz) для рендера без poppler.
    При отсутствии tesseract или битом PDF возвращает пустую строку (graceful).

    Args:
        pdf_bytes: сырые байты PDF-файла.
        dpi: разрешение рендера (300 dpi — баланс качество/скорость для OCR).
        max_pages: максимальное число страниц для обработки.

    Returns:
        Конкатенированный OCR-текст всех страниц (разделитель «\\n\\n»).
        Пустая строка при ошибке.
    """
    try:
        import fitz  # type: ignore[import-untyped]  # PyMuPDF
    except ImportError:
        logger.error("ocr_pdf_text: PyMuPDF не установлен — добавь pymupdf в pyproject.toml")
        return ""

    try:
        import pytesseract  # type: ignore[import-untyped]
        from PIL import Image
    except ImportError:
        logger.error(
            "ocr_pdf_text: pytesseract или PIL не установлен — "
            "добавь pytesseract, pillow в pyproject.toml"
        )
        return ""

    try:
        doc = fitz.open(stream=pdf_bytes, filetype="pdf")
    except Exception as exc:
        logger.warning("ocr_pdf_text: не удалось открыть PDF (%d байт): %s", len(pdf_bytes), exc)
        return ""

    pages_total = min(len(doc), max_pages)
    logger.info(
        "ocr_pdf_text: PDF %d стр. → обрабатываем %d стр. при dpi=%d",
        len(doc),
        pages_total,
        dpi,
    )

    text_parts: list[str] = []
    mat = fitz.Matrix(dpi / 72, dpi / 72)  # масштаб: 72 dpi → target dpi

    for page_num in range(pages_total):
        page = doc[page_num]
        try:
            pix = page.get_pixmap(matrix=mat, alpha=False)
            img_bytes = pix.tobytes("png")
            img = Image.open(io.BytesIO(img_bytes))
        except Exception as exc:
            logger.warning("ocr_pdf_text: страница %d — ошибка рендера: %s", page_num + 1, exc)
            text_parts.append("")
            continue

        try:
            page_text: str = pytesseract.image_to_string(img, lang="rus")
        except pytesseract.pytesseract.TesseractNotFoundError:
            logger.error(
                "ocr_pdf_text: Tesseract не найден в системе. "
                "Установи tesseract-ocr + tesseract-ocr-rus (apt-get)."
            )
            return ""
        except Exception as exc:
            logger.warning("ocr_pdf_text: страница %d — OCR ошибка: %s", page_num + 1, exc)
            page_text = ""

        text_parts.append(page_text)

    full_text = "\n\n".join(text_parts)
    logger.info(
        "ocr_pdf_text: OCR завершён: %d стр. → %d символов",
        pages_total,
        len(full_text),
    )
    return full_text


# ── Извлечение записей из OCR-текста ──────────────────────────────────────────


def extract_izyatie_records(
    ocr_text: str,
    doc_title: str,
    doc_url: str,
) -> list[dict[str, Any]]:
    """Извлекает записи об изъятии ЗУ из OCR-текста документа.

    Нормализует OCR-шум (пробелы в кад-номерах, «О»→«0», «б»→«6»), находит
    кад-номера 66:41:NNNNNNN:NN, дедуплицирует, извлекает реквизиты акта и цель.

    Args:
        ocr_text: сырой текст от OCR (может содержать артефакты).
        doc_title: заголовок документа (из HTML-страницы раздела).
        doc_url: URL PDF-файла (сохраняется в поле doc_url).

    Returns:
        Список словарей — по одному на каждый уникальный кад-номер. Поля:
          cad_num, reservation_kind, basis_act, act_number, act_date,
          purpose, doc_url, source, raw_excerpt.
        Пустой список если кад-номеров нет.
    """
    if not ocr_text or not ocr_text.strip():
        return []

    # Нормализуем OCR-шум ДО поиска кад-номеров.
    normalized = _normalize_ocr_text(ocr_text)

    # Реквизиты акта из заголовка или текста.
    act_number = _extract_act_number(doc_title) or _extract_act_number(normalized)
    act_date = _extract_act_date(doc_title) or _extract_act_date(
        normalized, require_act_context=True
    )
    purpose = _extract_purpose(doc_title) or _extract_purpose(normalized)

    # Поиск кад-номеров.
    seen: set[str] = set()
    records: list[dict[str, Any]] = []
    first_pos: int | None = None

    for m in _RE_CAD_NUM.finditer(normalized):
        cad = m.group(1)
        if cad in seen:
            continue
        seen.add(cad)

        if first_pos is None:
            first_pos = m.start()
            raw_excerpt = _build_excerpt(normalized, m.start())
        else:
            raw_excerpt = None

        records.append(
            {
                "cad_num": cad,
                "reservation_kind": "изъятие",
                "basis_act": doc_title,
                "act_number": act_number,
                "act_date": act_date,
                "purpose": purpose,
                "doc_url": doc_url,
                "source": "izyatie_ekb_ocr",
                "raw_excerpt": raw_excerpt,
            }
        )

    if records:
        logger.info(
            "extract_izyatie_records: документ %r → %d кад-номеров (act=%s)",
            doc_title[:60],
            len(records),
            act_number,
        )
    else:
        logger.debug(
            "extract_izyatie_records: документ %r → кад-номеров не найдено",
            doc_title[:60],
        )

    return records


# ── Вспомогательные функции ────────────────────────────────────────────────────


def _extract_act_number(text: str) -> str | None:
    """Извлекает номер акта из текста (первое вхождение «№ NNN-ПП»)."""
    m = _RE_ACT_NUMBER.search(text)
    if not m:
        return None
    return re.sub(r"\s+", "", m.group(1))


def _act_context_matches(text: str, pos: int) -> bool:
    """Стоит ли перед датой упоминание постановления — акта-основания."""
    ctx = text[max(0, pos - _ACT_CONTEXT_WINDOW) : pos].lower()
    return all(word in ctx for word in _ACT_CONTEXT_WORDS)


def _extract_act_date(text: str, *, require_act_context: bool = False) -> str | None:
    """Извлекает дату акта «от DD.MM.YYYY» → строка «YYYY-MM-DD» для SQL DATE.

    require_act_context=True — брать только дату, перед которой стоит
    упоминание постановления (#2464). Нужен для ТЕЛА
    документа, где первой датой почти всегда идёт ссылка на Генплан-2004.
    Для заголовка не нужен: там ссылок на посторонние акты нет.

    Если подходящей даты нет, возвращается None. Это сознательно: отсутствие
    даты честнее, чем дата чужого документа — по ней нельзя ни отфильтровать
    актуальные изъятия, ни сверить срок.
    """
    for m in _RE_ACT_DATE.finditer(text):
        if require_act_context and not _act_context_matches(text, m.start()):
            continue
        day, month, year = m.group(1), m.group(2), m.group(3)
        try:
            # Валидируем диапазоны.
            d, mo, y = int(day), int(month), int(year)
        except ValueError:
            continue
        if 1 <= d <= 31 and 1 <= mo <= 12 and 2000 <= y <= 2100:
            return f"{y:04d}-{mo:02d}-{d:02d}"
    return None


def _extract_purpose(text: str) -> str | None:
    """Извлекает цель изъятия из текста (best-effort)."""
    m = _RE_PURPOSE.search(text)
    if m:
        return re.sub(r"\s+", " ", m.group(1)).strip().rstrip(".,;")
    return None


def _build_excerpt(text: str, pos: int) -> str:
    """Возвращает ±100 символов вокруг позиции первого кад-номера."""
    start = max(0, pos - _EXCERPT_RADIUS)
    end = min(len(text), pos + _EXCERPT_RADIUS)
    return text[start:end]


__all__ = ["_normalize_ocr_text", "extract_izyatie_records", "ocr_pdf_text"]