TASK A (#29 G2): add parcel_meta to analyze response - New ParcelMeta Pydantic schema in app/schemas/parcel.py - SELECT from cad_parcels WHERE cad_num=:c in analyze_parcel() (step 9f) - Returns permitted_use_established_by_document, land_record_category_type, land_record_subtype, cost_value; None when row absent - Tests: test_analyze_parcel_meta.py (found + not-found cases) TASK B (#232 G3): cad_zouit fallback in _get_zouit_overlaps - When nspd_quarter_dumps has zouit_count==0, fall back to ST_Intersects query on cad_zouit (3483 rows, GIST indexed) - Overlaps tagged with source='cad_zouit'; format compatible with NSPD path - gate_verdict.py: BLOCKER_TYPE_ZONE_KEYWORDS tuple for keyword-based classification (охранная зона / трубопровод / электр / газ -> blocker; СЗЗ -> warning); NSPD subcategory path preserved backward-compat - Tests: 6 new test cases in test_gate_verdict.py covering cad_zouit path and backward-compat for NSPD subcategory path Updated db.execute call sequence in test_analyze_*.py (index shift +1 at pos 10).
234 lines
9.2 KiB
Python
234 lines
9.2 KiB
Python
"""Gate Verdict aggregator — собирает can-build-MKD signal из NSPD lookups.
|
||
|
||
Per #32 G5: user-facing answer на главный вопрос девелопера "Можно ли тут МКД?"
|
||
Foundation: nspd_zoning, nspd_zouit_overlaps, nspd_engineering_nearby (Sprint 1.1).
|
||
Pure function — no DB dependencies, consumes already-fetched data from analyze_parcel.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Literal, TypedDict
|
||
|
||
# ── Zone matching ─────────────────────────────────────────────────────────────
|
||
|
||
# Residential zone codes — точные ПЗЗ Свердловска (Ж-1..Ж-5)
|
||
RESIDENTIAL_ZONE_PREFIXES = ("Ж-", "Ж1", "Ж2", "Ж3", "Ж4", "Ж5")
|
||
# Fallback по zone_name (lowercase substring)
|
||
RESIDENTIAL_KEYWORDS = ("жил",)
|
||
|
||
# ── ЗОУИТ taxonomy ────────────────────────────────────────────────────────────
|
||
|
||
# Subcategory codes которые блокируют МКД (охранные зоны ЛЭП/газа/трубопровода).
|
||
# Используется для overlaps из NSPD dump (subcategory INT там заполнен).
|
||
BLOCKER_SUBCATEGORIES: dict[int, str] = {
|
||
17: "Инженерные коммуникации (охранная зона ЛЭП/газа/трубопровода)",
|
||
}
|
||
|
||
# Keyword-based blockers для cad_zouit fallback (#232).
|
||
# cad_zouit.subcategory = 100% NULL, поэтому классификация по type_zone substring.
|
||
# Порядок: lowercase, substring match (any wins → blocker).
|
||
BLOCKER_TYPE_ZONE_KEYWORDS: tuple[str, ...] = (
|
||
"охранная зона",
|
||
"трубопровод",
|
||
"электр",
|
||
"газ",
|
||
)
|
||
|
||
# type_zone substrings, которые дают warning (не blocker) из cad_zouit.
|
||
WARNING_TYPE_ZONE_KEYWORDS: tuple[str, ...] = ("сзз", "санитарно-защитная")
|
||
|
||
# Engineering nearby search radius (метры) — совпадает с quarter_dump_lookup.py
|
||
ENGINEERING_NEARBY_THRESHOLD_M = 200
|
||
|
||
|
||
# ── TypedDicts ────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class Blocker(TypedDict):
|
||
code: str
|
||
detail: str
|
||
|
||
|
||
class Warning(TypedDict):
|
||
code: str
|
||
detail: str
|
||
|
||
|
||
class GateVerdict(TypedDict):
|
||
can_build_mkd: bool | Literal["unknown"]
|
||
verdict_label: str # "Можно" | "Нельзя" | "С ограничениями" | "Нужна проверка"
|
||
blockers: list[Blocker]
|
||
warnings: list[Warning]
|
||
checks_performed: list[str]
|
||
source: Literal["nspd_dump", "nspd_dump_partial", "no_data"]
|
||
|
||
|
||
# ── Helpers ───────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def is_residential_zone(zone_code: str | None, zone_name: str | None) -> bool:
|
||
"""Check if ПЗЗ zone permits residential (МКД) building.
|
||
|
||
Checks zone_code prefix first (exact ПЗЗ codes), then zone_name lowercase
|
||
substring match as fallback for non-standard zone labels.
|
||
"""
|
||
if zone_code:
|
||
if any(zone_code.startswith(p) for p in RESIDENTIAL_ZONE_PREFIXES):
|
||
return True
|
||
if zone_name:
|
||
zn_lower = zone_name.lower()
|
||
if any(kw in zn_lower for kw in RESIDENTIAL_KEYWORDS):
|
||
return True
|
||
return False
|
||
|
||
|
||
# ── Main aggregator ───────────────────────────────────────────────────────────
|
||
|
||
|
||
def compute_gate_verdict(
|
||
nspd_zoning: dict | None,
|
||
nspd_zouit_overlaps: list[dict] | None,
|
||
nspd_engineering_nearby: list[dict] | None,
|
||
nspd_dump: dict | None,
|
||
) -> GateVerdict:
|
||
"""Aggregate NSPD lookups into can-build-MKD verdict.
|
||
|
||
Checks performed (in order):
|
||
1. ПЗЗ residential zone match (blocker if non-residential)
|
||
2. ЗОУИТ subcategory 17 overlap (blocker); other subcategories → warnings
|
||
3. Engineering nearby presence check (warning-only, not a blocker)
|
||
|
||
Returns GateVerdict с blockers/warnings/checks_performed и source label.
|
||
"""
|
||
blockers: list[Blocker] = []
|
||
warnings: list[Warning] = []
|
||
checks: list[str] = []
|
||
|
||
# Guard: no NSPD data at all
|
||
dump_available = (nspd_dump or {}).get("available", False)
|
||
dump_stale = (nspd_dump or {}).get("stale", False)
|
||
if not dump_available:
|
||
return GateVerdict(
|
||
can_build_mkd="unknown",
|
||
verdict_label="Нужна проверка",
|
||
blockers=[],
|
||
warnings=[
|
||
Warning(
|
||
code="NO_NSPD_DUMP",
|
||
detail="Данные NSPD ещё не подгружены",
|
||
)
|
||
],
|
||
checks_performed=[],
|
||
source="no_data",
|
||
)
|
||
|
||
source: Literal["nspd_dump", "nspd_dump_partial"] = (
|
||
"nspd_dump_partial" if dump_stale else "nspd_dump"
|
||
)
|
||
|
||
# Check 1 — ПЗЗ residential zone
|
||
checks.append("ПЗЗ зональность")
|
||
if nspd_zoning:
|
||
if not is_residential_zone(nspd_zoning.get("zone_code"), nspd_zoning.get("zone_name")):
|
||
zone_label = (
|
||
nspd_zoning.get("zone_name") or nspd_zoning.get("zone_code") or "неизвестна"
|
||
)
|
||
blockers.append(
|
||
Blocker(
|
||
code="PZZ_NOT_RESIDENTIAL",
|
||
detail=f"Зона: {zone_label}",
|
||
)
|
||
)
|
||
else:
|
||
warnings.append(
|
||
Warning(
|
||
code="PZZ_UNKNOWN",
|
||
detail="ПЗЗ зона не определена",
|
||
)
|
||
)
|
||
|
||
# Check 2 — ЗОУИТ overlaps
|
||
checks.append("ЗОУИТ пересечения")
|
||
for overlap in nspd_zouit_overlaps or []:
|
||
src = overlap.get("source", "nspd-quarter-dump")
|
||
if src == "cad_zouit":
|
||
# cad_zouit fallback path: classify by type_zone keywords (#232).
|
||
# subcategory = NULL в cad_zouit, поэтому subcategory-based logic не применяется.
|
||
type_zone_lower = (overlap.get("type_zone") or overlap.get("layer") or "").lower()
|
||
if any(kw in type_zone_lower for kw in BLOCKER_TYPE_ZONE_KEYWORDS):
|
||
blockers.append(
|
||
Blocker(
|
||
code="ZOUIT_CAD_BLOCKER",
|
||
detail=(
|
||
f"Охранная зона ({overlap.get('type_zone', '')}): "
|
||
f"{overlap.get('name', '')}"
|
||
),
|
||
)
|
||
)
|
||
elif any(kw in type_zone_lower for kw in WARNING_TYPE_ZONE_KEYWORDS):
|
||
warnings.append(
|
||
Warning(
|
||
code="ZOUIT_CAD_SZZ",
|
||
detail=(
|
||
f"СЗЗ ({overlap.get('type_zone', '')}): " f"{overlap.get('name', '')}"
|
||
),
|
||
)
|
||
)
|
||
else:
|
||
warnings.append(
|
||
Warning(
|
||
code="ZOUIT_CAD_OTHER",
|
||
detail=(
|
||
f"ЗОУИТ cad ({overlap.get('type_zone', '')}): "
|
||
f"{overlap.get('name', '')}"
|
||
),
|
||
)
|
||
)
|
||
else:
|
||
# NSPD dump path: subcategory-based logic (backward-compat).
|
||
sub = overlap.get("subcategory")
|
||
if isinstance(sub, int) and sub in BLOCKER_SUBCATEGORIES:
|
||
blockers.append(
|
||
Blocker(
|
||
code=f"ZOUIT_OVERLAP_SUB{sub}",
|
||
detail=f"{BLOCKER_SUBCATEGORIES[sub]}: {overlap.get('name', '')}",
|
||
)
|
||
)
|
||
else:
|
||
warnings.append(
|
||
Warning(
|
||
code=f"ZOUIT_SUB{sub if sub is not None else 'unknown'}",
|
||
detail=f"ЗОУИТ {overlap.get('layer', '')}: {overlap.get('name', '')}",
|
||
)
|
||
)
|
||
|
||
# Check 3 — Engineering nearby (warning only, not a blocker)
|
||
checks.append(f"Инженерные сети в радиусе {ENGINEERING_NEARBY_THRESHOLD_M}м")
|
||
if not (nspd_engineering_nearby or []):
|
||
warnings.append(
|
||
Warning(
|
||
code="NO_ENGINEERING_NEARBY",
|
||
detail=f"Сетей не найдено в радиусе {ENGINEERING_NEARBY_THRESHOLD_M}м",
|
||
)
|
||
)
|
||
|
||
# Final verdict
|
||
can_build: bool | Literal["unknown"]
|
||
if blockers:
|
||
can_build = False
|
||
label = "Нельзя"
|
||
elif warnings:
|
||
can_build = True
|
||
label = "С ограничениями"
|
||
else:
|
||
can_build = True
|
||
label = "Можно"
|
||
|
||
return GateVerdict(
|
||
can_build_mkd=can_build,
|
||
verdict_label=label,
|
||
blockers=blockers,
|
||
warnings=warnings,
|
||
checks_performed=checks,
|
||
source=source,
|
||
)
|