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

"""ПАТ-lookup для site-finder analyze (#1065 / #1067).

Читает pat_subzones и возвращает подзоны ПАТ, пересекающие переданный участок.
Graceful: если таблица ещё не задеплоена или БД недоступна — возвращает пустой список.
Wiring в analyze_parcel — follow-up (не в этом PR).
"""

from __future__ import annotations

import logging
from typing import Any

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

logger = logging.getLogger(__name__)

# ST_Intersects к pat_subzones, GIST idx_pat_subzones_geom.
# Исключаем subzone_no=0 (aggregate_outline) — клиент получает конкретные подзоны 1..6.
_PAT_INTERSECTS_SQL = text(
    """
    SELECT subzone_no, name, restriction, aerodrome
    FROM pat_subzones
    WHERE subzone_no > 0
      AND ST_Intersects(geom, ST_GeomFromText(CAST(:parcel_wkt AS text), 4326))
    ORDER BY subzone_no
    """
)


def parcel_pat_subzones(db: Session, parcel_wkt: str | None) -> list[dict[str, Any]]:
    """Подзоны ПАТ, пересекающие участок.

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

    Returns:
        Список dict [{subzone_no, name, restriction, aerodrome}] отсортированный
        по subzone_no. Пустой список при отсутствии пересечений, ошибке БД
        или недеплоенной таблице.
    """
    if not parcel_wkt:
        return []

    try:
        rows = db.execute(_PAT_INTERSECTS_SQL, {"parcel_wkt": parcel_wkt}).mappings().all()
    except (OperationalError, ProgrammingError) as exc:
        # Таблица ещё не задеплоена / структура изменилась — graceful degrade.
        logger.warning("parcel_pat_subzones: pat_subzones недоступна, skip: %s", exc)
        return []

    return [
        {
            "subzone_no": r["subzone_no"],
            "name": r["name"],
            "restriction": r["restriction"],
            "aerodrome": r["aerodrome"],
        }
        for r in rows
    ]


__all__ = ["parcel_pat_subzones"]