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

"""КРТ-реквизиты для analyze_parcel (#1060, эпик #1067).

Связывает геометрию КРТ-площадок из ``planning_projects`` (ППТ/ПМТ, м.134) с реестром
реквизитов ``ekb_krt_sites`` (застройщик / аукционная цена / градпотенциал).

Ключ связки: номер постановления ПАГЕ из ``ekb_krt_sites.paga_decisions`` + год совпадает с
заголовком ``planning_projects.full_name``. Даёт ``confidence='exact'`` для 3 из 15 площадок
(по состоянию на 2026-06). Для остальных КРТ-геом-пересечений (opportunity_krt в ird_overlays —
это ККР, не КРТ-площадки застройки) надёжного ключа нет → ``matched=False, confidence='none'``.

Graceful: нет таблицы / пустой WKT / исключение БД → [].
Зеркалит стиль ``ird_overlay_lookup.py`` + ``planning_lookup.py``.
"""

from __future__ import annotations

import logging
import re
from typing import Any

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

logger = logging.getLogger(__name__)

# Участок → planning_projects, которые пересекают его И имеют в full_name «ПАГЕ № N от»
# (первичный документ, не «Внесение изменений»). Geom — 4326. CAST psycopg v3.
_PLANNING_KRT_SQL = text(
    """
    SELECT
        pp.id           AS pp_id,
        pp.full_name    AS pp_full_name,
        pp.dmd_actual_year,
        pp.doc_status_name,
        k.id            AS krt_id,
        k.site_name,
        k.developer_name,
        k.contract_number,
        k.auction_final_price,
        k.gradpotential_sqm,
        k.area_ha,
        k.krt_kind
    FROM planning_projects pp
    -- только первичные документы (не «Внесение изменений»)
    CROSS JOIN ekb_krt_sites k
    -- пересечение участка с PP-геометрией
    WHERE ST_Intersects(
            pp.geom,
            ST_GeomFromText(CAST(:parcel_wkt AS text), 4326)
          )
    -- матч по ПАГЕ-номеру: в full_name есть «ПАГЕ № {paga_num} от», год совпадает
    AND EXISTS (
        SELECT 1
        FROM jsonb_array_elements(k.paga_decisions) d
        WHERE pp.full_name ~ ('^[[:space:]]*ПАГЕ № ' || (d->>'number') || ' от')
          AND pp.dmd_actual_year::text = split_part(d->>'date', '.', 3)
          AND (d->>'title' ILIKE '%планировки%' OR d->>'title' ILIKE '%межевания%')
    )
    ORDER BY pp.dmd_actual_year DESC, k.id
    """
)

# Вторичный запрос: planning_projects без матча с ekb_krt_sites — geom-only без реквизитов.
# Запускается только если первый не дал результатов, чтобы не путать пользователя
# двумя overlapping-источниками.
#
# Regex-топонимы для geom-only КРТ-fallback — собран как ОДИН непрерывный паттерн.
# Раньше literal был разорван переносом строки внутри ОДНОЙ пары '...' внутри SQL,
# и ветки 'Донбасской\n        ' / 'Лумумбы\n        ' содержали NL+8 пробелов отступа,
# никогда не матчили однострочные full_name (PG `~` — POSIX ARE без (?x)) — #1230.
# Python склеивает соседние строковые литералы на этапе парсинга, поэтому regex попадает
# в SQL одной строкой. Не SQL-injection: статическая константа, без user input.
# При добавлении новых топонимов держим строки <100 chars и НЕ ломаем regex переносом.
_KRT_TOPONYM_REGEX = (
    "[кК][рР][тТ]|комплексн.*развити|Краснофлотцев|Донбасской|Шефской|Азина"
    "|Черкасской|Кварцевой|Летчиков|Вилонова|Лумумбы|Студенческой|Васильева"
    "|Гурзуфск|Бакинских|Шадринский|Бабушкина"
)

_PLANNING_NO_MATCH_SQL = text(
    """
    SELECT
        pp.id          AS pp_id,
        pp.full_name   AS pp_full_name,
        pp.dmd_actual_year,
        pp.doc_status_name
    FROM planning_projects pp
    WHERE ST_Intersects(
            pp.geom,
            ST_GeomFromText(CAST(:parcel_wkt AS text), 4326)
          )
      AND pp.full_name ~ '"""
    + _KRT_TOPONYM_REGEX
    + """'
    ORDER BY pp.dmd_actual_year DESC
    """
)


def _extract_paga_num(full_name: str) -> str | None:
    """Извлечь номер ПАГЕ из заголовка planning_project.full_name."""
    m = re.search(r"ПАГЕ\s*№\s*(\d+)\s+от", full_name)
    return m.group(1) if m else None


def parcel_krt_requisites(db: Session, parcel_wkt: str | None) -> list[dict[str, Any]]:
    """КРТ-реквизиты для участка: застройщик / цена / градпотенциал.

    Алгоритм:
      1. ST_Intersects участка с ``planning_projects`` + матч ПАГЕ-номер+год с
         ``ekb_krt_sites.paga_decisions`` → ``confidence='exact'``.
      2. Если матчей нет, ищем ППТ/ПМТ в ``planning_projects`` с КРТ-топонимами
         → ``matched=False, confidence='none'`` (геом-факт без реквизитов).
      3. Никакой ложной атрибуции: при отсутствии exact-матча — NOT передаём
         developer_name/auction_final_price (остаются None).

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

    Returns:
        list[dict]: каждый элемент содержит поля ниже; список пуст при graceful-деградации.
        Поля: ``matched`` (bool), ``confidence`` ('exact'|'none'),
        ``krt_geom_ref`` (pp_id или None), ``pp_full_name`` (str),
        ``site_name``, ``developer_name``, ``auction_final_price``,
        ``gradpotential_sqm``, ``area_ha``, ``contract_number`` — None при matched=False.
    """
    if not parcel_wkt:
        return []

    try:
        with db.begin_nested():
            rows = db.execute(_PLANNING_KRT_SQL, {"parcel_wkt": parcel_wkt}).mappings().all()
    except (OperationalError, ProgrammingError) as exc:
        logger.warning("parcel_krt_requisites: таблица недоступна, skip: %s", exc)
        return []
    except Exception as exc:
        logger.warning("parcel_krt_requisites: неожиданная ошибка, skip: %s", exc)
        return []

    if rows:
        # Есть exact-матч — возвращаем с реквизитами
        seen: set[tuple[int, int]] = set()
        result: list[dict[str, Any]] = []
        for r in rows:
            key = (r["pp_id"], r["krt_id"])
            if key in seen:
                continue
            seen.add(key)
            result.append(
                {
                    "matched": True,
                    "confidence": "exact",
                    "krt_geom_ref": r["pp_id"],
                    "pp_full_name": r["pp_full_name"],
                    "site_name": r["site_name"],
                    "developer_name": r["developer_name"],
                    "contract_number": r["contract_number"],
                    "auction_final_price": (
                        float(r["auction_final_price"])
                        if r["auction_final_price"] is not None
                        else None
                    ),
                    "gradpotential_sqm": (
                        float(r["gradpotential_sqm"])
                        if r["gradpotential_sqm"] is not None
                        else None
                    ),
                    "area_ha": float(r["area_ha"]) if r["area_ha"] is not None else None,
                }
            )
        return result

    # Нет exact-матча — geom-only поиск по КРТ-топонимам
    try:
        with db.begin_nested():
            no_match_rows = (
                db.execute(_PLANNING_NO_MATCH_SQL, {"parcel_wkt": parcel_wkt}).mappings().all()
            )
    except (OperationalError, ProgrammingError) as exc:
        logger.warning("parcel_krt_requisites: no-match fallback недоступен: %s", exc)
        return []
    except Exception as exc:
        logger.warning("parcel_krt_requisites: no-match fallback ошибка: %s", exc)
        return []

    return [
        {
            "matched": False,
            "confidence": "none",
            "krt_geom_ref": r["pp_id"],
            "pp_full_name": r["pp_full_name"],
            "site_name": None,
            "developer_name": None,
            "contract_number": None,
            "auction_final_price": None,
            "gradpotential_sqm": None,
            "area_ha": None,
        }
        for r in no_match_rows
    ]


__all__ = ["parcel_krt_requisites"]