56 lines
2.5 KiB
Python
56 lines
2.5 KiB
Python
"""ППТ/ПМТ planning-overlay lookup для analyze_parcel (#1085 analyze-wiring).
|
|
|
|
Читает ``planning_projects`` (м.134, наполняется planning_harvest #1085) и отдаёт проекты
|
|
планировки/межевания, пересекающие участок: что/где утверждено к застройке (future-supply).
|
|
Комплементарно ИРД-overlay (``ird_overlay_lookup``): тот — ограничения участка, этот — граддоки.
|
|
|
|
Graceful: если таблица ``planning_projects`` ещё не задеплоена (pre-migration) или БД-ошибка —
|
|
возвращает пустой список (analyze не падает). Wiring в analyze — за тем же флагом, что ИРД (D9b).
|
|
"""
|
|
|
|
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__)
|
|
|
|
# ППТ/ПМТ overlap по участку: ST_Intersects через GIST idx_planning_projects_geom. geom уже 4326
|
|
# (harvest стор WFS geoloc как 4326). Участок — WKT в 4326. CAST psycopg v3.
|
|
_PLANNING_OVERLAP_SQL = text(
|
|
"""
|
|
SELECT project_type, doc_status_name, full_name, project_name, dmd_actual_year
|
|
FROM planning_projects
|
|
WHERE ST_Intersects(geom, ST_GeomFromText(CAST(:parcel_wkt AS text), 4326))
|
|
ORDER BY project_type, dmd_actual_year DESC NULLS LAST
|
|
"""
|
|
)
|
|
|
|
|
|
def parcel_planning_overlaps(db: Session, parcel_wkt: str | None) -> list[dict[str, Any]]:
|
|
"""ППТ/ПМТ, пересекающие участок.
|
|
|
|
Args:
|
|
db: сессия.
|
|
parcel_wkt: геометрия участка в WKT (EPSG:4326). None → пустой список.
|
|
|
|
Returns:
|
|
list[{project_type, doc_status_name, full_name, project_name, dmd_actual_year}].
|
|
Пусто при отсутствии участка/таблицы/пересечений (analyze не падает).
|
|
"""
|
|
if not parcel_wkt:
|
|
return []
|
|
try:
|
|
rows = db.execute(_PLANNING_OVERLAP_SQL, {"parcel_wkt": parcel_wkt}).mappings().all()
|
|
except (OperationalError, ProgrammingError) as exc:
|
|
# Таблица ещё не задеплоена / БД-ошибка — graceful degrade.
|
|
logger.warning("parcel_planning_overlaps: planning_projects недоступна, skip: %s", exc)
|
|
return []
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
__all__ = ["parcel_planning_overlaps"]
|