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

"""Точный радиус-запрос разрешений на строительство/ввод рядом с участком (ГИСОГД-66).

Заменяет TODO-прокси quarter-prefix из analyze_parcel (#105 Phase 5): после того как
Phase 3 (geocoding → geom) наполнила `gisogd_permits.geom`, соседние РНС/РВЭ ищутся
точным `ST_DWithin(..., radius_m)` по геометрии, а не грубым совпадением префикса
кадастрового квартала. Старый quarter-prefix путь (`ekburg_construction_permits` →
`recent_permits_in_quarter` / `permits_summary`) НЕ трогается — он остаётся параллельно
как отдельный, более богатый полями источник; этот модуль лишь добавляет второй,
геометрически честный ключ payload.

`gisogd_permits`: id / doc_group ('RS' — разрешение на строительство, 'RV' — ввод в
эксплуатацию) / doc_num / doc_name / date_doc / date_reg / approved_organization /
cad_nums TEXT[] / geom geometry(MultiPolygon, 4326), GIST-индекс на geom.

PURE-по-данным (никаких выдуманных чисел): пустой радиус — честный ноль
(total_count=0, items=[]), НЕ ошибка и НЕ None. Вызывается из синхронного analyze_parcel
с той же non-fatal семантикой.
"""

from __future__ import annotations

import logging
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)

# Максимум записей в списке items. Счётчики (total_count/rs_count/rv_count/
# nearest_distance_m) считаются по ПОЛНОЙ выборке радиуса — БЕЗ LIMIT в SQL, чтобы не было
# тихого капа (проектная конвенция «без тихих капов»: тот же паттерн чинили в дедупе
# объявлений и фрешнес-мониторе). LIMIT капает лишь выдачу items → items_truncated=True.
_ITEMS_LIMIT = 30

# Радиус-запрос по образцу POI-в-радиусе (osm_poi_ekb) из analyze_parcel: ST_DWithin по
# geography-центроиду участка. `::geography` приклеено к `)`, НЕ к bind-name → psycopg v3
# его не съедает (backend.md: исключение из :name::type-запрета). БЕЗ LIMIT: агрегаты честны
# по всей выборке; в радиусе 500м GIST-индекс делает выборку дешёвой, тысяч строк не будет.
_PERMITS_NEARBY_SQL = text("""
    SELECT
        doc_group,
        doc_name,
        doc_num,
        date_doc,
        date_reg,
        approved_organization,
        ST_Distance(
            geom::geography,
            ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography
        ) AS distance_m
      FROM gisogd_permits
     -- Только РНС/РВЭ: агрегат обещает total_count = rs_count + rv_count, а с
     -- #2986 в таблице появилась третья группа 'IZ' (изменения в разрешение).
     -- Она попадала бы в total и не попадала ни в один из счётчиков — молчаливое
     -- расхождение. Показывать ли изменения отдельной строкой в §6 — вопрос
     -- продуктовый (см. #2986); до его решения выборка сужена явно, а не молча.
     WHERE doc_group IN ('RS', 'RV')
       AND geom IS NOT NULL
       AND ST_DWithin(
               geom::geography,
               ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography,
               :radius_m
           )
     ORDER BY distance_m ASC
""")


def _empty_result(radius_m: int) -> dict[str, Any]:
    """Честный нулевой агрегат: 0 разрешений в радиусе (НЕ ошибка, НЕ None)."""
    return {
        "radius_m": radius_m,
        "total_count": 0,
        "rs_count": 0,
        "rv_count": 0,
        "nearest_distance_m": None,
        "items": [],
        "items_truncated": False,
        "source": "gisogd66",
    }


def get_permits_nearby(db: Session, geom_wkt: str, radius_m: int = 500) -> dict[str, Any]:
    """РНС/РВЭ из ГИСОГД-66 в радиусе `radius_m` метров от участка (точный ST_DWithin).

    Args:
        db: активная сессия SQLAlchemy.
        geom_wkt: WKT-геометрия анализируемого участка (та же переменная, что для POI).
        radius_m: радиус поиска в метрах (по умолчанию 500).

    Returns:
        Агрегат с ЧЕСТНЫМИ счётчиками (см. `_empty_result` для пустой формы):
        radius_m, total_count / rs_count ('RS') / rv_count ('RV') — по ВСЕЙ выборке радиуса
        (без тихого капа), nearest_distance_m (round 1, None если пусто),
        items (≤30 ближайших), items_truncated (True если total_count > len(items)),
        source="gisogd66". Пустой радиус → честный ноль.
    """
    rows = [
        dict(r)
        for r in db.execute(
            _PERMITS_NEARBY_SQL,
            {"wkt": geom_wkt, "radius_m": radius_m},
        )
        .mappings()
        .all()
    ]

    if not rows:
        return _empty_result(radius_m)

    # Счётчики — по ПОЛНОЙ выборке (все rows), список items — только первые _ITEMS_LIMIT.
    rs_count = sum(1 for r in rows if r["doc_group"] == "RS")
    rv_count = sum(1 for r in rows if r["doc_group"] == "RV")

    items: list[dict[str, Any]] = []
    for r in rows[:_ITEMS_LIMIT]:
        date_doc = r["date_doc"]
        distance_m = r["distance_m"]
        items.append(
            {
                "doc_group": r["doc_group"],
                "doc_name": r["doc_name"],
                "doc_num": r["doc_num"],
                "date_doc": date_doc.isoformat() if date_doc is not None else None,
                "approved_organization": r["approved_organization"],
                "distance_m": round(distance_m, 1) if distance_m is not None else None,
            }
        )

    nearest = items[0]["distance_m"] if items else None
    total_count = len(rows)
    return {
        "radius_m": radius_m,
        "total_count": total_count,
        "rs_count": rs_count,
        "rv_count": rv_count,
        "nearest_distance_m": nearest,
        "items": items,
        "items_truncated": total_count > len(items),
        "source": "gisogd66",
    }