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

"""Парсер постановлений об изъятии/резервировании ЗУ из PDF/текста (#1091, #1062, #1067).

Источник — PDF-постановления Правительства Свердловской области (pravo.gov66.ru / ПАГЕ).
Геометрии зон изъятия/резервирования нет ни в одном spatial-источнике; перечень ЗУ существует
ТОЛЬКО текстом. Этот модуль — чистая функция от bytes/text: извлекает кад-номера, тип
(изъятие/резервирование), реквизиты акта, цель. Без сети; тестируемо без PDF-файлов.

Seed-документ: 509-ПП (Правительство СО, изъятие под реконструкцию ул. Татищева, ЕКБ).

Зависимость pdfplumber: добавлена в backend/pyproject.toml. Если недоступна — модуль падает
с ImportError на уровне extract_from_pdf (parse-функции работают без неё).
"""

from __future__ import annotations

import logging
import re
from dataclasses import dataclass
from datetime import date
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    pass

logger = logging.getLogger(__name__)

# ── Константы / регулярки ──────────────────────────────────────────────────────

# Полный кадастровый номер ЗУ: 66:41:1234567:12 (6 или 7 цифр в квартале, любой номер ЗУ).
_RE_CAD_NUM = re.compile(r"\b(\d{2}:\d{2}:\d{6,7}:\d+)\b")

# Номер постановления: 509-ПП, 1234-ПА, 55-ПП, NNN-РП и т.п.
# Паттерн намеренно широкий — постановления РФ имеют разные суффиксы.
_RE_ACT_NUMBER = re.compile(
    r"№\s*(\d+[\s\-–—]?(?:[А-Яа-яA-Za-z]+[\-–—]?)*(?:ПП|ПА|РП|ПГ|ГП|МО))\b",
    re.UNICODE,
)

# Дата акта: «от 12.03.2024» (приоритетно) или «12 марта 2024 г.»
_RE_DATE_DMY = re.compile(r"от\s+(\d{2})\.(\d{2})\.(\d{4})")
_RE_DATE_WORD = re.compile(
    r"(\d{1,2})\s+"
    r"(январ[яь]|феврал[яь]|март[аа]?|апрел[яь]|ма[йя]|июн[яь]|"
    r"июл[яь]|август[аа]?|сентябр[яь]|октябр[яь]|ноябр[яь]|декабр[яь])"
    r"\s+(\d{4})",
    re.UNICODE | re.IGNORECASE,
)

_MONTH_MAP: dict[str, int] = {
    "январ": 1,
    "феврал": 2,
    "март": 3,
    "апрел": 4,
    "ма": 5,
    "июн": 6,
    "июл": 7,
    "август": 8,
    "сентябр": 9,
    "октябр": 10,
    "ноябр": 11,
    "декабр": 12,
}

# Слова-маркеры типа документа.
_RE_REZERV = re.compile(r"резервир", re.IGNORECASE | re.UNICODE)
_RE_IZYAT = re.compile(r"изъят", re.IGNORECASE | re.UNICODE)

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

_EXCERPT_RADIUS = 120  # символов в каждую сторону от первого кад-номера


# ── Dataclass результата ───────────────────────────────────────────────────────


@dataclass(frozen=True)
class ReservationRecord:
    """Одна запись кад-номера ЗУ, извлечённая из постановления."""

    cad_num: str
    reservation_kind: str  # 'изъятие' | 'резервирование'
    act_number: str | None
    act_date: date | None
    purpose: str | None
    raw_excerpt: str | None


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


def _parse_act_number(text: str) -> str | None:
    """Извлекает номер постановления из текста (первое вхождение)."""
    m = _RE_ACT_NUMBER.search(text)
    if not m:
        return None
    # Нормализуем пробелы внутри номера.
    return re.sub(r"\s+", "", m.group(1))


def _parse_act_date(text: str) -> date | None:
    """Извлекает дату акта: сначала «от DD.MM.YYYY», потом «DD <месяц-словом> YYYY»."""
    m = _RE_DATE_DMY.search(text)
    if m:
        try:
            return date(int(m.group(3)), int(m.group(2)), int(m.group(1)))
        except ValueError:
            logger.debug("_parse_act_date: некорректная числовая дата %s", m.group(0))

    m2 = _RE_DATE_WORD.search(text)
    if m2:
        day_s, month_s, year_s = m2.group(1), m2.group(2).lower(), m2.group(3)
        month_num = next(
            (v for k, v in _MONTH_MAP.items() if month_s.startswith(k)),
            None,
        )
        if month_num is not None:
            try:
                return date(int(year_s), month_num, int(day_s))
            except ValueError:
                logger.debug("_parse_act_date: некорректная словесная дата %s", m2.group(0))

    return None


def _detect_kind(text: str, default_kind: str) -> str:
    """Определяет тип операции: 'резервирование' | 'изъятие' | default_kind.

    Побеждает то слово, что встретилось РАНЬШЕ, а не то, что стоит выше в
    коде (#2464). Прежний безусловный приоритет «резервир» переклассифицировал
    ВЕСЬ документ — все участки разом, — если постановление об изъятии хоть
    раз ссылалось на резервирование (типовая формулировка «ранее
    зарезервированных земель», ссылка на утративший силу акт). Тема документа
    стоит в заголовке, поэтому позиция первого упоминания — сигнал сильнее
    порядка проверок, и он симметричен: заголовок «О резервировании» так же
    выигрывает у «изъятия» в теле.
    """
    m_rez = _RE_REZERV.search(text)
    m_izy = _RE_IZYAT.search(text)
    if m_rez and m_izy:
        return "резервирование" if m_rez.start() < m_izy.start() else "изъятие"
    if m_rez:
        return "резервирование"
    if m_izy:
        return "изъятие"
    return default_kind


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:
    """±120 символов вокруг позиции первого кад-номера."""
    start = max(0, pos - _EXCERPT_RADIUS)
    end = min(len(text), pos + _EXCERPT_RADIUS)
    return text[start:end]


# ── Основные публичные функции ─────────────────────────────────────────────────


def extract_reservations(
    text: str,
    *,
    default_kind: str = "изъятие",
) -> list[ReservationRecord]:
    """Извлекает записи ЗУ из текста постановления об изъятии/резервировании.

    Args:
        text: полный текст постановления (конкатенация всех страниц PDF или plain-text).
        default_kind: тип по умолчанию, если ни «изъят», ни «резервир» не встретились.

    Returns:
        Список ReservationRecord, дедуплицированный по cad_num внутри документа.
        Пустой список если кад-номеров нет.
    """
    if not text or not text.strip():
        return []

    kind = _detect_kind(text, default_kind)
    act_number = _parse_act_number(text)
    act_date = _parse_act_date(text)
    purpose = _extract_purpose(text)

    seen: set[str] = set()
    records: list[ReservationRecord] = []
    first_pos: int | None = None

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

        if first_pos is None:
            first_pos = m.start()

        records.append(
            ReservationRecord(
                cad_num=cad,
                reservation_kind=kind,
                act_number=act_number,
                act_date=act_date,
                purpose=purpose,
                raw_excerpt=_build_excerpt(text, m.start()) if first_pos == m.start() else None,
            )
        )

    # Ставим raw_excerpt только первой записи (выше он уже записан через first_pos == m.start()).
    # Для остальных строк оставляем None (достаточно для верификации).
    logger.debug(
        "extract_reservations: kind=%s act=%s date=%s cad_count=%d",
        kind,
        act_number,
        act_date,
        len(records),
    )
    return records


def extract_from_pdf(pdf_bytes: bytes) -> list[ReservationRecord]:
    """Извлекает записи ЗУ из байт PDF-постановления.

    Args:
        pdf_bytes: сырые байты PDF-файла.

    Returns:
        Список ReservationRecord. Пустой список при пустом/нечитаемом PDF.

    Raises:
        ImportError: если pdfplumber не установлен (добавить в backend/pyproject.toml).
    """
    try:
        import pdfplumber  # type: ignore[import-untyped]
    except ImportError as exc:
        raise ImportError(
            "pdfplumber не установлен. Добавь 'pdfplumber>=0.10.0' в backend/pyproject.toml "
            "и выполни 'uv lock'."
        ) from exc

    import io

    text_parts: list[str] = []
    try:
        with pdfplumber.open(io.BytesIO(pdf_bytes)) as pdf:
            for page in pdf.pages:
                page_text = page.extract_text() or ""
                text_parts.append(page_text)
    except Exception as exc:
        logger.warning("extract_from_pdf: не удалось прочитать PDF: %s", exc)
        return []

    full_text = "\n".join(text_parts)
    return extract_reservations(full_text)


__all__ = ["ReservationRecord", "extract_from_pdf", "extract_reservations"]