gendesign/backend/app/services/scrapers/page_reservation_parser.py
lekss361 59f2628e0b
All checks were successful
Deploy / changes (push) Successful in 6s
Deploy / build-frontend (push) Has been skipped
Deploy / build-backend (push) Successful in 2m41s
Deploy / build-worker (push) Successful in 3m25s
Deploy / deploy (push) Successful in 1m25s
feat(sf): ПАГЕ-парсер изъятия/резервирования → land_reservation (#1118)
Co-authored-by: lekss361 <lekss361@gendsgn.local>
Co-committed-by: lekss361 <lekss361@gendsgn.local>
2026-06-07 10:15:03 +00:00

252 lines
10 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Парсер постановлений об изъятии/резервировании ЗУ из 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."""
if _RE_REZERV.search(text):
return "резервирование"
if _RE_IZYAT.search(text):
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"]