feat(analyze): surface real ПЗЗ regulation (КСИТ/height) into nspd_zoning

Resolve numeric ПЗЗ limits (max_far/КСИТ, max_height_m, max_floors,
max_building_pct, min_parcel_area_m2) by parcel centroid via the coordinate
resolver get_or_fetch_zone_regulation (cache-first; bounded 3s live geoportal)
and merge into the /analyze nspd_zoning object. Powers the ПТИЦА cockpit's
real КСИТ/height/density (was placeholder). Flag-gated
(enable_zoning_regulation_in_analyze, default on) and hot-path-safe: any
exception/timeout → null fields, analyze never 500s or slow-fails (3s cap).
Additive — existing nspd_zoning keys untouched. 198 tests pass.

Note: zone_index_at is a live WFS call per analyze even on warm cache
(~0.3-0.5s healthy / 3s cap degraded) — optimization tracked as follow-up.

backend/app/api/v1/parcels.py

@@ -53,6 +53,7 @@ from app.services.exporters.report_md import (
 from app.services.exporters.report_pdf import export_report_pdf
 from app.services.exporters.report_pptx import render_report_pptx
 from app.services.exporters.snapshot_pdf import generate_snapshot_pdf
+from app.services.scrapers.ekb_geoportal_client import EKBGeoportalClient
 from app.services.site_finder.best_layouts import get_best_layouts
 from app.services.site_finder.cadastre_fetch import (
     cad_exists_in_db,
@@ -96,6 +97,7 @@ from app.services.site_finder.weight_profiles import (
 from app.services.site_finder.weight_profiles import (
     resolve_weights as _resolve_weights,
 )
+from app.services.site_finder.zone_regulation import get_or_fetch_zone_regulation
 from app.services.weather_cache import (
     get_air_quality_cached,
     get_seasonal_weather_cached,
@@ -424,6 +426,11 @@ _GEOM_LABEL_MICRO_HA = 0.05  # ниже → label "микро" (комбинир
 _GEOM_LABEL_GOOD = 0.7
 _GEOM_LABEL_MEDIUM = 0.4
 
+# ПЗЗ-градрегламент в analyze (#1067): bounded timeout (сек) для live-geoportal вызова
+# на cache-miss. Короткий, чтобы медленный геопортал не вешал hot-path analyze —
+# cache-hit (норма для ЕКБ) сетевых вызовов не делает вообще.
+_ZONE_REGULATION_TIMEOUT_S = 3
+
 
 def _polygon_suitability(geom_wkt: str) -> dict[str, Any]:
     """P1 (#45) — physical suitability участка по метрикам shape.
@@ -2052,6 +2059,47 @@ def analyze_parcel(
         # Independent dict per request — never mutate module singleton.
         nspd_dump_data = make_empty_result()
 
+    # 9e-bis) Реальный ПЗЗ-градрегламент зоны → merge в nspd_zoning (Route 2, #1067).
+    # Резолвим числовые предельные параметры (КСИТ/max_far, высота, этажность,
+    # %застройки, min площадь ЗУ) по centroid через get_or_fetch_zone_regulation:
+    # cache-first (кэш прогрет для ЕКБ), live-geoportal только на miss и закапан
+    # коротким timeout'ом (_ZONE_REGULATION_TIMEOUT_S), чтобы медленный геопортал не
+    # вешал analyze. Hot-path-safe: любой сбой/таймаут → поля остаются None, analyze
+    # не падает. Существующие ключи nspd_zoning (zone_code/zone_name/source/raw_props)
+    # не трогаем — только ДОБАВЛЯЕМ regulation-поля.
+    _nspd_zoning = nspd_dump_data.get("nspd_zoning")
+    if (
+        settings.enable_zoning_regulation_in_analyze
+        and _nspd_zoning is not None
+        and centroid_lat is not None
+        and centroid_lon is not None
+    ):
+        try:
+            _client = EKBGeoportalClient(timeout=_ZONE_REGULATION_TIMEOUT_S)
+            _regulation = get_or_fetch_zone_regulation(
+                db, centroid_lon, centroid_lat, client=_client
+            )
+        except Exception as e:
+            # Никогда не роняем analyze из-за регламента — деградируем в None.
+            logger.warning("zone regulation resolve failed for %s: %s", cad_num, e)
+            _regulation = None
+        if _regulation is not None:
+            _nspd_zoning["max_far"] = _regulation.get("max_far")
+            _nspd_zoning["max_height_m"] = _regulation.get("max_height_m")
+            _nspd_zoning["max_floors"] = _regulation.get("max_floors")
+            _nspd_zoning["max_building_pct"] = _regulation.get("max_building_pct")
+            _nspd_zoning["min_parcel_area_m2"] = _regulation.get("min_parcel_area_m2")
+            _nspd_zoning["regulation_zone_index"] = _regulation.get("zone_index")
+            _nspd_zoning["regulation_source"] = "ekb-geoportal-urbancard"
+            logger.debug(
+                "zone regulation hit for %s: zone_index=%s max_far=%s",
+                cad_num,
+                _regulation.get("zone_index"),
+                _regulation.get("max_far"),
+            )
+        else:
+            logger.debug("zone regulation miss for %s (no resolve)", cad_num)
+
     # 9f) Parcel meta — ВРИ и кадастровые метаданные из cad_parcels (#29 G2)
     parcel_meta: ParcelMeta | None = None
     try:

backend/app/core/config.py

@@ -207,6 +207,16 @@ class Settings(BaseSettings):
     # (callable вручную для smoke-тестирования конкретного bbox).
     enable_riasurt_harvest: bool = False
 
+    # Реальный ПЗЗ-градрегламент зоны (КСИТ/max_far, высота, этажность, %застройки,
+    # min площадь ЗУ) в ответе analyze: поле `nspd_zoning` дополняется числовыми
+    # предельными параметрами из zone_regulation_cache (Route 2 — coordinate-based
+    # resolver get_or_fetch_zone_regulation, cache-first + bounded live geoportal на
+    # miss). Кэш прогрет для ЕКБ (33 зоны), поэтому live-вызовы редки и закапаны
+    # коротким timeout'ом. Hot-path-safe: любой сбой/таймаут → поля None, analyze не
+    # падает. По умолчанию ON; выключение (env ENABLE_ZONING_REGULATION_IN_ANALYZE=false)
+    # полностью пропускает резолв — поведение analyze без изменений.
+    enable_zoning_regulation_in_analyze: bool = True
+
     # ── LLM infrastructure (#960) ────────────────────────────────────────────
     # ОПЦИОНАЛЬНЫЙ слой поверх детерминированного движка. Forecasting НИКОГДА не
     # зависит от LLM — при любом сбое/выключенности возвращается детерминированный

backend/tests/api/v1/test_analyze_zoning_regulation.py

@@ -0,0 +1,261 @@
+"""Тесты для реального ПЗЗ-градрегламента в nspd_zoning ответа analyze (Route 2, #1067).
+
+Покрывает merge числовых предельных параметров зоны (КСИТ/max_far, высота, этажность,
+%застройки, min площадь ЗУ) в nspd_zoning через get_or_fetch_zone_regulation:
+
+1. resolver вернул регламент → nspd_zoning несёт max_far/max_height_m/regulation_source
+2. resolver кинул исключение → analyze всё равно 200 и nspd_zoning БЕЗ regulation-полей
+   (hot-path-safe, не падаем)
+3. флаг enable_zoning_regulation_in_analyze=False → resolver НЕ вызывается, полей нет
+
+Стратегия mock: как в test_analyze_parcel_meta.py — DB mock через dependency_overrides,
+тяжёлые сервисы патчим через unittest.mock.patch. get_quarter_dump_data замокан так, чтобы
+вернуть непустой nspd_zoning (иначе merge пропускается по guard `_nspd_zoning is not None`).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+from fastapi.testclient import TestClient
+
+from app.main import app
+
+_CAD = "66:41:0204016:10"
+_WKT = "POLYGON((60.6 56.838, 60.61 56.838, 60.61 56.845, 60.6 56.845, 60.6 56.838))"
+_GEOJSON = '{"type":"Polygon","coordinates":[[[60.6,56.838],[60.61,56.838]]]}'
+
+_NSPD_ZONING = {
+    "zone_code": "Ц-1",
+    "zone_name": "Общественно-деловая зона",
+    "source": "nspd-quarter-dump",
+    "raw_props": {"type_zone": "Общественно-деловая зона"},
+}
+
+_REGULATION = {
+    "zone_index": "Ц-1",
+    "max_far": 4.0,
+    "max_height_m": 25.0,
+    "max_floors": 9,
+    "max_building_pct": 100.0,
+    "min_parcel_area_m2": 1500.0,
+    "source": "ekb_geoportal",
+}
+
+
+def _make_mapping(data: dict[str, Any]) -> MagicMock:
+    m = MagicMock()
+    m.__getitem__ = lambda self, k: data[k]
+    m.get = lambda k, default=None: data.get(k, default)
+    return m
+
+
+def _make_db_for_analyze() -> MagicMock:
+    """Mock DB Session для analyze_parcel (минимальный — фокус на nspd_zoning merge)."""
+    db = MagicMock()
+
+    geom_row = _make_mapping({"geom_geojson": _GEOJSON, "geom_wkb": None, "source": "cad_quarter"})
+    wkt_row = _make_mapping({"wkt": _WKT})
+    district_row = _make_mapping(
+        {
+            "district_name": "Октябрьский",
+            "median_price_per_m2": 120000,
+            "dist_to_center": 1500.0,
+        }
+    )
+    centroid_row = _make_mapping({"lat": 56.84, "lon": 60.605})
+
+    call_idx = [0]
+    responses: list[Any] = [
+        ("first", geom_row),  # 0: geom UNION ALL
+        ("first", wkt_row),  # 1: WKT
+        ("first", district_row),  # 2: district
+        ("all", []),  # 3: POI rows
+        ("all", []),  # 4: competitor rows
+        ("all", []),  # 5: pipeline rows
+        ("first", centroid_row),  # 6: centroid
+        ("all", []),  # 7: noise rows
+        ("all", []),  # 8: hydrology rows
+        ("all", []),  # 9: utilities rows
+        # (далее nspd_zoning merge не ходит в db.execute — резолвер замокан)
+        ("first", None),  # 10: parcel_meta
+        ("first", None),  # 11: market trend
+        ("first", None),  # 12: zoning (begin_nested)
+        ("all", []),  # 13: success recommendation
+        ("first", None),  # 14: market price
+        ("all", []),  # 15: recent permits
+        ("scalar", 0),  # 16: geotech_risk
+        ("all", []),  # 17: neighbors
+        ("first", None),  # 18: overlap
+    ]
+
+    def _execute_side_effect(*args: Any, **kwargs: Any) -> MagicMock:
+        idx = call_idx[0]
+        call_idx[0] += 1
+        if idx >= len(responses):
+            r = MagicMock()
+            r.mappings.return_value.first.return_value = None
+            r.mappings.return_value.all.return_value = []
+            r.scalar.return_value = 0
+            return r
+        kind, data = responses[idx]
+        r = MagicMock()
+        r.mappings.return_value.first.return_value = data
+        r.mappings.return_value.all.return_value = data if isinstance(data, list) else []
+        r.scalar.return_value = data if kind == "scalar" else 0
+        return r
+
+    db.execute.side_effect = _execute_side_effect
+
+    ctx = MagicMock()
+    ctx.__enter__ = MagicMock(return_value=ctx)
+    ctx.__exit__ = MagicMock(return_value=False)
+    db.begin_nested.return_value = ctx
+
+    return db
+
+
+def _override_db(db: MagicMock):
+    def _get_db_override():
+        yield db
+
+    return _get_db_override
+
+
+def _base_patches() -> list[Any]:
+    """Патчи тяжёлых сервисов; get_quarter_dump_data отдаёт непустой nspd_zoning."""
+    return [
+        patch("app.api.v1.parcels.get_air_quality_cached", return_value=None),
+        patch("app.api.v1.parcels.get_weather_cached", return_value=None),
+        patch("app.api.v1.parcels.get_seasonal_weather_cached", return_value=None),
+        patch(
+            "app.api.v1.parcels.get_quarter_dump_data",
+            return_value={
+                # dict(_NSPD_ZONING) — свежая копия per-test, merge мутирует её in-place
+                "nspd_zoning": dict(_NSPD_ZONING),
+                "nspd_zouit_overlaps": [],
+                "nspd_engineering_nearby": [],
+                "nspd_risk_zones": [],
+                "nspd_opportunity_parcels": [],
+                "nspd_red_lines": [],
+                "nspd_dump": {"available": False, "stale": False, "harvest_triggered": False},
+            },
+        ),
+        patch("app.api.v1.parcels.compute_velocity", return_value=None),
+        patch("app.api.v1.parcels.compute_gate_verdict", return_value={"verdict": "unknown"}),
+        patch(
+            "app.api.v1.parcels.build_ird_analyze_block",
+            return_value={
+                "ird_overlaps": [],
+                "ird_by_kind": {},
+                "opportunity_overlaps": [],
+                "planning_projects": [],
+                "functional_zone": None,
+                "krt": [],
+                "zone_regulation": None,
+            },
+        ),
+    ]
+
+
+def test_zoning_regulation_merged_when_resolved() -> None:
+    """resolver вернул регламент → nspd_zoning несёт max_far/max_height_m/regulation_source."""
+    from app.core.db import get_db
+
+    db = _make_db_for_analyze()
+    app.dependency_overrides[get_db] = _override_db(db)
+    patches = [
+        *_base_patches(),
+        patch(
+            "app.api.v1.parcels.get_or_fetch_zone_regulation",
+            return_value=dict(_REGULATION),
+        ),
+    ]
+    for p in patches:
+        p.start()
+    try:
+        client = TestClient(app)
+        resp = client.post(f"/api/v1/parcels/{_CAD}/analyze")
+        assert resp.status_code == 200, resp.text
+        body = resp.json()
+        zoning = body["nspd_zoning"]
+        assert zoning is not None
+        # существующие ключи сохранены
+        assert zoning["zone_code"] == "Ц-1"
+        assert zoning["source"] == "nspd-quarter-dump"
+        # regulation-поля добавлены
+        assert zoning["max_far"] == 4.0
+        assert zoning["max_height_m"] == 25.0
+        assert zoning["max_floors"] == 9
+        assert zoning["max_building_pct"] == 100.0
+        assert zoning["min_parcel_area_m2"] == 1500.0
+        assert zoning["regulation_zone_index"] == "Ц-1"
+        assert zoning["regulation_source"] == "ekb-geoportal-urbancard"
+    finally:
+        app.dependency_overrides.clear()
+        for p in patches:
+            p.stop()
+
+
+def test_zoning_regulation_graceful_on_resolver_exception() -> None:
+    """resolver кинул → analyze 200, nspd_zoning сохранён БЕЗ regulation-полей (не падаем)."""
+    from app.core.db import get_db
+
+    db = _make_db_for_analyze()
+    app.dependency_overrides[get_db] = _override_db(db)
+    patches = [
+        *_base_patches(),
+        patch(
+            "app.api.v1.parcels.get_or_fetch_zone_regulation",
+            side_effect=RuntimeError("geoportal timeout"),
+        ),
+    ]
+    for p in patches:
+        p.start()
+    try:
+        client = TestClient(app)
+        resp = client.post(f"/api/v1/parcels/{_CAD}/analyze")
+        assert resp.status_code == 200, resp.text
+        body = resp.json()
+        zoning = body["nspd_zoning"]
+        assert zoning is not None
+        assert zoning["zone_code"] == "Ц-1"  # существующие ключи целы
+        # regulation-поля НЕ просочились (graceful degrade)
+        assert "max_far" not in zoning
+        assert "max_height_m" not in zoning
+        assert "regulation_source" not in zoning
+    finally:
+        app.dependency_overrides.clear()
+        for p in patches:
+            p.stop()
+
+
+def test_zoning_regulation_skipped_when_flag_off() -> None:
+    """флаг OFF → resolver НЕ вызывается, regulation-полей нет, nspd_zoning без изменений."""
+    from app.core.db import get_db
+
+    db = _make_db_for_analyze()
+    app.dependency_overrides[get_db] = _override_db(db)
+    resolver_mock = MagicMock(return_value=dict(_REGULATION))
+    patches = [
+        *_base_patches(),
+        patch("app.api.v1.parcels.settings.enable_zoning_regulation_in_analyze", False),
+        patch("app.api.v1.parcels.get_or_fetch_zone_regulation", resolver_mock),
+    ]
+    for p in patches:
+        p.start()
+    try:
+        client = TestClient(app)
+        resp = client.post(f"/api/v1/parcels/{_CAD}/analyze")
+        assert resp.status_code == 200, resp.text
+        body = resp.json()
+        zoning = body["nspd_zoning"]
+        assert zoning is not None
+        assert zoning["zone_code"] == "Ц-1"
+        assert "max_far" not in zoning  # резолв пропущен → полей нет
+        resolver_mock.assert_not_called()  # флаг OFF → резолвер не дёрнут
+    finally:
+        app.dependency_overrides.clear()
+        for p in patches:
+            p.stop()