feat(sf): parcel_ird_overlaps read-side для analyze (#1092)
Some checks failed
Deploy / changes (push) Successful in 7s
Deploy / build-backend (push) Successful in 1m26s
Deploy / build-frontend (push) Has been skipped
Deploy / build-worker (push) Successful in 3m29s
Deploy / deploy (push) Has been cancelled

ird_overlay_lookup.py — read-side ИРД-overlay lookup: ST_Intersects участка (WKT 4326) против ird_overlays (м.132) → ограничения + reg_numb_border + zone_index/type + группировка by_kind. Standalone (parcels.py не тронут), GIST idx_ird_overlays_geom, graceful-degrade при отсутствии таблицы. Wiring в analyze деферится в D9b flag-gated. 4 теста.

Refs #1067.
Co-authored-by: lekss361 <lekss361@gendsgn.local>
Co-committed-by: lekss361 <lekss361@gendsgn.local>
This commit is contained in:
lekss361 2026-06-06 19:01:06 +00:00 committed by bot-reviewer
parent 579ad54202
commit 9b9c6e85d3
2 changed files with 170 additions and 0 deletions

View file

@ -0,0 +1,82 @@
"""ИРД-overlay lookup для analyze_parcel (#1067 D9).
Читает ``ird_overlays`` (м.132, наполняется harvest-task #1067 B5) и отдаёт ИРД-ограничения,
пересекающие участок: ЗОУИТ / ПЗЗ-границы / ОКН / ООПТ-леса-вода с reg_numb_border и (когда
резолвлено, #1067 C8/D9-content) читаемым типом/индексом зоны.
Отличие от ``quarter_dump_lookup`` (nspd_quarter_dumps blob-кеш): тот отдаёт per-quarter дамп
ЗОУИТ/zoning из NSPD QuarterDump; ``ird_overlays`` выделенный per-feature ИРД-store с
reg_numb_border + content-колонками (для геопортал-резолва ВРИ/типа). Комплементарно: канонический
ИРД-слой «волшебной таблицы».
Graceful: если таблица ``ird_overlays`` ещё не задеплоена (pre-migration) или БД-ошибка возвращает
пустой результат (analyze не падает). Wiring в analyze flag-gated (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__)
# Пустой результат — sentinel для analyze try/except (НЕ мутировать).
IRD_EMPTY_RESULT: dict[str, Any] = {"ird_overlaps": [], "ird_by_kind": {}}
# ИРД-overlay по участку: ST_Intersects через GIST idx_ird_overlays_geom. geom уже EPSG:4326
# (harvest стор ST_Transform 3857→4326). Участок передаётся как WKT в 4326. CAST psycopg v3.
_IRD_OVERLAP_SQL = text(
"""
SELECT layer_kind, reg_numb_border, category, subcategory,
zone_index, zone_type_name
FROM ird_overlays
WHERE ST_Intersects(geom, ST_GeomFromText(CAST(:parcel_wkt AS text), 4326))
ORDER BY layer_kind, reg_numb_border
"""
)
def parcel_ird_overlaps(db: Session, parcel_wkt: str | None) -> dict[str, Any]:
"""ИРД-ограничения, пересекающие участок.
Args:
db: сессия.
parcel_wkt: геометрия участка в WKT (EPSG:4326). None пустой результат.
Returns:
{'ird_overlaps': [{layer_kind, reg_numb_border, category, subcategory,
zone_index, zone_type_name}], 'ird_by_kind': {layer_kind: count}}.
Пусто при отсутствии участка/таблицы/пересечений (analyze не падает).
"""
if not parcel_wkt:
return {"ird_overlaps": [], "ird_by_kind": {}}
try:
rows = db.execute(_IRD_OVERLAP_SQL, {"parcel_wkt": parcel_wkt}).mappings().all()
except (OperationalError, ProgrammingError) as exc:
# Таблица ещё не задеплоена / БД-ошибка — graceful degrade.
logger.warning("parcel_ird_overlaps: ird_overlays недоступна, skip: %s", exc)
return {"ird_overlaps": [], "ird_by_kind": {}}
overlaps: list[dict[str, Any]] = []
by_kind: dict[str, int] = {}
for r in rows:
kind = r["layer_kind"]
overlaps.append(
{
"layer_kind": kind,
"reg_numb_border": r["reg_numb_border"],
"category": r["category"],
"subcategory": r["subcategory"],
"zone_index": r["zone_index"],
"zone_type_name": r["zone_type_name"],
}
)
by_kind[kind] = by_kind.get(kind, 0) + 1
return {"ird_overlaps": overlaps, "ird_by_kind": by_kind}
__all__ = ["IRD_EMPTY_RESULT", "parcel_ird_overlaps"]

View file

@ -0,0 +1,88 @@
"""Тесты parcel_ird_overlaps (#1067 D9) — ИРД-overlay lookup для analyze."""
from __future__ import annotations
from typing import Any
from sqlalchemy.exc import OperationalError
from app.services.site_finder.ird_overlay_lookup import parcel_ird_overlaps
class _Result:
def __init__(self, rows: list[dict[str, Any]]) -> None:
self._rows = rows
def mappings(self) -> _Result:
return self
def all(self) -> list[dict[str, Any]]:
return self._rows
class _FakeDB:
def __init__(
self, rows: list[dict[str, Any]] | None = None, raise_exc: Exception | None = None
) -> None:
self._rows = rows or []
self._raise = raise_exc
def execute(self, sql: Any, params: dict[str, Any] | None = None) -> _Result:
if self._raise is not None:
raise self._raise
return _Result(self._rows)
_WKT = "POLYGON((60.5 56.8, 60.6 56.8, 60.6 56.9, 60.5 56.9, 60.5 56.8))"
def test_groups_overlaps_by_kind() -> None:
rows = [
{
"layer_kind": "zouit_okn",
"reg_numb_border": "66:41-6.41",
"category": 36940,
"subcategory": 20,
"zone_index": None,
"zone_type_name": None,
},
{
"layer_kind": "zouit_okn",
"reg_numb_border": "66:41-6.42",
"category": 36940,
"subcategory": 20,
"zone_index": None,
"zone_type_name": None,
},
{
"layer_kind": "territorial_zones",
"reg_numb_border": "66:41-7.10",
"category": 472819,
"subcategory": 1,
"zone_index": "Ц-1",
"zone_type_name": None,
},
]
res = parcel_ird_overlaps(_FakeDB(rows), _WKT)
assert len(res["ird_overlaps"]) == 3
assert res["ird_by_kind"] == {"zouit_okn": 2, "territorial_zones": 1}
tz = next(o for o in res["ird_overlaps"] if o["layer_kind"] == "territorial_zones")
assert tz["reg_numb_border"] == "66:41-7.10"
assert tz["zone_index"] == "Ц-1"
def test_empty_when_no_wkt() -> None:
res = parcel_ird_overlaps(_FakeDB([{"layer_kind": "x"}]), None)
assert res == {"ird_overlaps": [], "ird_by_kind": {}}
def test_empty_when_no_overlap() -> None:
res = parcel_ird_overlaps(_FakeDB([]), _WKT)
assert res == {"ird_overlaps": [], "ird_by_kind": {}}
def test_graceful_when_table_missing() -> None:
"""ird_overlays ещё не задеплоена → пусто, без падения analyze."""
db = _FakeDB(raise_exc=OperationalError("stmt", {}, Exception("no such table")))
res = parcel_ird_overlaps(db, _WKT)
assert res == {"ird_overlaps": [], "ird_by_kind": {}}