Light1YT
86e9ea2937
Многоагентный аудит + имплементация: один воркер на файл, точечные правки. Верификация: py_compile (47/47 .py) + tsc --noEmit (0 ошибок). Unit-тесты не прогонялись (окружение не поднято: rollup native dep / нет pytest-venv). Полностью исправлено (169): #1336, #1337, #1339, #1340, #1341, #1342, #1343, #1345, #1346, #1348, #1349, #1350, #1351, #1354, #1356, #1358, #1359, #1360, #1362, #1364, #1365, #1366, #1367, #1368, #1369, #1370, #1371, #1372, #1373, #1374, #1375, #1376, #1377, #1378, #1379, #1380, #1381, #1382, #1384, #1385, #1386, #1387, #1388, #1389, #1390, #1391, #1392, #1394, #1395, #1396, #1397, #1399, #1400, #1401, #1402, #1403, #1404, #1408, #1409, #1410, #1411, #1412, #1413, #1414, #1415, #1416, #1417, #1418, #1420, #1423, #1425, #1426, #1427, #1428, #1429, #1430, #1431, #1432, #1433, #1434, #1435, #1437, #1438, #1439, #1440, #1441, #1442, #1443, #1444, #1445, #1446, #1447, #1448, #1449, #1450, #1451, #1452, #1453, #1454, #1455, #1456, #1457, #1458, #1459, #1460, #1461, #1462, #1463, #1464, #1465, #1466, #1467, #1468, #1469, #1471, #1472, #1473, #1474, #1476, #1478, #1479, #1481, #1482, #1483, #1484, #1485, #1487, #1488, #1489, #1490, #1491, #1492, #1493, #1494, #1495, #1496, #1497, #1499, #1500, #1501, #1502, #1504, #1505, #1506, #1507, #1510, #1514, #1515, #1516, #1517, #1518, #1519, #1521, #1522, #1523, #1524, #1525, #1526, #1527, #1528, #1529, #1531, #1532, #1533, #1534, #1535, #1536, #1537, #1538 Частично (9, in-file часть, остаток cross-file): #1361, #1419, #1422, #1424, #1470, #1475, #1477, #1480, #1498 Требуют cross-file (3, не тронуты): #1338, #1363, #1421 Пропущено (1): #1539 Не входило в партию: 22 needs-Leha issue (нужны решения владельца). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
257 lines
11 KiB
Python
257 lines
11 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()
|
||
# СЗЗ-маркер специфичнее широких blocker-подстрок ('газ'/'электр'),
|
||
# поэтому проверяем warning-keywords ПЕРВЫМИ (#1341). СЗЗ газораспределительной
|
||
# станции/электроподстанции — ограничение, а не hard-blocker.
|
||
if 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', '')}"
|
||
),
|
||
)
|
||
)
|
||
elif 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', '')}"
|
||
),
|
||
)
|
||
)
|
||
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).
|
||
# subcategory из _get_zouit_overlaps = `subcategory or type_zone` → может быть
|
||
# строкой ('17') или fallback на type_zone (#1360). Нормализуем к int как
|
||
# _get_zouit_engineering_overlaps, иначе blocker subcategory-17 тихо
|
||
# деградирует до warning.
|
||
sub_raw = overlap.get("subcategory")
|
||
sub: int | None
|
||
if isinstance(sub_raw, bool):
|
||
sub = None
|
||
else:
|
||
try:
|
||
sub = int(sub_raw) # type: ignore[arg-type]
|
||
except (ValueError, TypeError):
|
||
sub = None
|
||
if sub is not None and sub in BLOCKER_SUBCATEGORIES:
|
||
blockers.append(
|
||
Blocker(
|
||
code=f"ZOUIT_OVERLAP_SUB{sub}",
|
||
detail=f"{BLOCKER_SUBCATEGORIES[sub]}: {overlap.get('name', '')}",
|
||
)
|
||
)
|
||
else:
|
||
# Сохраняем исходную метку (str/int) в warning-коде если нормализация
|
||
# не дала int — иначе теряется человекочитаемый ярлык subcategory.
|
||
if sub is not None:
|
||
sub_label: int | str = sub
|
||
elif sub_raw is not None:
|
||
sub_label = sub_raw
|
||
else:
|
||
sub_label = "unknown"
|
||
warnings.append(
|
||
Warning(
|
||
code=f"ZOUIT_SUB{sub_label}",
|
||
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,
|
||
)
|