gendesign/backend/app/services/site_finder/granddoc_lookup.py

"""Lookup разработанной граддокументации (ППТ/ПМТ) из planning_projects (#1061).

Читает ``planning_projects`` (м.134) и отдаёт для участка: тип документа, статус «разработан»
и реквизиты ПАГЕ (номер + дата + предмет). Используется в ИРД-блоке analyze.

Отличие от ``planning_lookup.parcel_planning_overlaps``: тот отдаёт сырые поля
(future-supply обзор); этот — нормализованные поля с regex-парсингом ПАГЕ и is_active-флагом
(для UI-отображения статуса разработки).

Graceful: нет wkt / таблицы (OperationalError/ProgrammingError) → [].
Зеркалит стиль ird_overlay_lookup.py.
"""

from __future__ import annotations

import logging
import re
from datetime import date
from typing import Any

from sqlalchemy import text
from sqlalchemy.exc import OperationalError, ProgrammingError
from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)

# Реквизиты ПАГЕ из full_name: «ПАГЕ № 2792 от 16.12.2021\n<предмет>»
_PAGA_RE = re.compile(r"ПАГЕ\s*№\s*(\d+)\s+от\s+(\d{2})\.(\d{2})\.(\d{4})", re.IGNORECASE)

# ST_Intersects через GIST idx_planning_projects_geom; geom уже 4326 (WFS harvest).
# Участок — WKT 4326. CAST psycopg v3 (никогда :param::type).
_GRANDDOC_SQL = text(
    """
    SELECT project_type, doc_status_name, full_name, doc_full_name
    FROM planning_projects
    WHERE ST_Intersects(geom, ST_GeomFromText(CAST(:parcel_wkt AS text), 4326))
    ORDER BY project_type, doc_status_name
    """
)


def _parse_paga(full_name: str | None) -> tuple[str | None, date | None, str | None]:
    """Распарсить реквизиты ПАГЕ из full_name.

    Returns:
        (paga_number, paga_date, subject). Любое поле может быть None если не распарсилось.
        subject — текст после строки с реквизитами ПАГЕ (или None).
    """
    if not full_name:
        return None, None, None
    m = _PAGA_RE.search(full_name)
    if m is None:
        return None, None, None

    paga_number = m.group(1)
    try:
        paga_date: date | None = date(int(m.group(4)), int(m.group(3)), int(m.group(2)))
    except ValueError:
        paga_date = None

    # Предмет — текст после первой строки с ПАГЕ-реквизитами; обрезаем незначащие символы.
    separators = "\n\r ,;— "
    after = full_name[m.end() :]
    for ch in separators:
        after = after.lstrip(ch)
    after = after.strip()
    subject: str | None = after if after else None

    return paga_number, paga_date, subject


def parcel_granddoc(db: Session, parcel_wkt: str | None) -> list[dict[str, Any]]:
    """Разработанная граддокументация (ППТ/ПМТ) на участок из planning_projects.

    Args:
        db: сессия SQLAlchemy.
        parcel_wkt: геометрия участка в WKT (EPSG:4326). None → пустой список.

    Returns:
        list[{project_type, doc_status_name, is_active, paga_number, paga_date,
        subject, doc_full_name}].
        - ``is_active``: doc_status_name == 'действующий' (регистронезависимо).
        - ``paga_number`` / ``paga_date``: regex из full_name; None если не распарсилось.
        - ``subject``: текст full_name после строки с ПАГЕ-реквизитами; None если нет.
        Отсортировано: сначала действующие, внутри группы — по paga_date desc (свежие первыми).
        Пустой список при отсутствии участка / таблицы / пересечений (analyze не падает).
    """
    if not parcel_wkt:
        return []
    try:
        rows = db.execute(_GRANDDOC_SQL, {"parcel_wkt": parcel_wkt}).mappings().all()
    except (OperationalError, ProgrammingError) as exc:
        logger.warning("parcel_granddoc: planning_projects недоступна, skip: %s", exc)
        return []

    result: list[dict[str, Any]] = []
    for r in rows:
        full_name: str | None = r["full_name"]
        paga_number, paga_date, subject = _parse_paga(full_name)
        status_name: str | None = r["doc_status_name"]
        # Регистронезависимо: harvest пишет doc_status_name из WFS без нормализации,
        # DDL/COMMENT документируют хранимое значение как lowercase ('действующий').
        is_active = (status_name or "").strip().lower() == "действующий"
        result.append(
            {
                "project_type": r["project_type"],
                "doc_status_name": r["doc_status_name"],
                "is_active": is_active,
                "paga_number": paga_number,
                "paga_date": paga_date,
                "subject": subject,
                "doc_full_name": r["doc_full_name"],
            }
        )

    # Сначала действующие (is_active=True), внутри группы — по paga_date desc (свежие первыми).
    # None-дата уходит в конец (None < date → sentinel через _neg_date).
    result.sort(
        key=lambda d: (not d["is_active"], d["paga_date"] is None, _neg_date(d["paga_date"]))
    )
    return result


def _neg_date(d: date | None) -> tuple[int, int, int]:
    """Ключ для обратной сортировки по дате (None → максимум = уходит в хвост)."""
    if d is None:
        return (0, 0, 0)
    # Отрицание для desc: чем больше дата — тем меньше кортеж → выше в сортировке.
    return (-d.year, -d.month, -d.day)


__all__ = ["parcel_granddoc"]