feat(#263 sub-A): cad_* overlay schemas + cadastre territorial_zones (ПЗЗ)

Phase 1 (migration 101): создаёт 5 новых таблиц cad_territorial_zones,
cad_red_lines, cad_engineering_structures, cad_risk_zones,
cad_opportunity_parcels. Idempotent (IF NOT EXISTS), GIST-индексы.

Phase 2 partial (ПЗЗ only):
- NSPDBulkClient.get_features_in_bbox_grid — async grid-walk WMS GetFeatureInfo
- NSPDBulkClient.get_territorial_zones_in_bbox — convenience wrapper layer 875838
- harvest_quarter Phase 2.5 — вызов get_territorial_zones_in_bbox после grid-walk
- _save_territorial_zones — UPSERT loop с SAVEPOINT per-row, ON CONFLICT zone_id

9 unit tests (mock DB): insert/upsert/no-geom/no-id/error-isolation/savepoint pattern.

backend/app/scrapers/nspd_bulk_client.py

@@ -374,7 +374,107 @@ class NSPDBulkClient:
         raw_features: list[dict[str, Any]] = (data or {}).get("features") or []
         return [NSPDBulkFeature.model_validate(f) for f in raw_features]
 
-    # ── 3. list_objects_in_building ───────────────────────────────────────────
+    # ── 3. get_features_in_bbox_grid ─────────────────────────────────────────
+
+    async def get_features_in_bbox_grid(
+        self,
+        layer_id: int,
+        bbox: tuple[float, float, float, float],
+        *,
+        grid_n: int = 7,
+        tile_size: int = 512,
+    ) -> list[dict]:
+        """Grid-walk WMS GetFeatureInfo по layer_id в bbox.
+
+        Разбивает bbox на grid_n × grid_n ячеек, кликает центр каждой.
+        Дедуплицирует результаты по feature id.
+
+        Args:
+            layer_id: NSPD layer ID (например 875838 для ПЗЗ).
+            bbox: (xmin, ymin, xmax, ymax) в EPSG:3857 (метры).
+            grid_n: количество ячеек по каждой оси (7 → 49 запросов).
+            tile_size: размер виртуального WMS тайла в пикселях.
+
+        Returns:
+            Список raw feature dict'ов с полями id, geometry, properties.
+            Дедуплицированы по feature id.
+        """
+        xmin, ymin, xmax, ymax = bbox
+        x_step = (xmax - xmin) / grid_n
+        y_step = (ymax - ymin) / grid_n
+        click_xy = (tile_size // 2, tile_size // 2)
+
+        tasks = []
+        for i in range(grid_n):
+            for j in range(grid_n):
+                cell_bbox = (
+                    xmin + i * x_step,
+                    ymin + j * y_step,
+                    xmin + (i + 1) * x_step,
+                    ymin + (j + 1) * y_step,
+                )
+                tasks.append(
+                    self.wms_feature_info(
+                        layer_id=layer_id,
+                        bbox=cell_bbox,
+                        click_xy=click_xy,
+                        width=tile_size,
+                        height=tile_size,
+                    )
+                )
+
+        # Запускаем все ячейки конкурентно (semaphore ограничивает до 3 concurrent)
+        cell_results = await asyncio.gather(*tasks, return_exceptions=True)
+
+        seen_ids: set[str] = set()
+        results: list[dict] = []
+
+        for idx, cell_result in enumerate(cell_results):
+            if isinstance(cell_result, Exception):
+                logger.debug(
+                    "get_features_in_bbox_grid: layer=%d cell=%d error: %s",
+                    layer_id,
+                    idx,
+                    cell_result,
+                )
+                continue
+            for feature in cell_result:
+                fid = str(feature.id) if feature.id is not None else ""
+                if fid and fid in seen_ids:
+                    continue
+                if fid:
+                    seen_ids.add(fid)
+                results.append(
+                    {
+                        "id": feature.id,
+                        "geometry": feature.geometry,
+                        "properties": feature.properties,
+                    }
+                )
+
+        logger.info(
+            "get_features_in_bbox_grid: layer=%d grid=%dx%d unique_features=%d",
+            layer_id,
+            grid_n,
+            grid_n,
+            len(results),
+        )
+        return results
+
+    async def get_territorial_zones_in_bbox(
+        self,
+        bbox: tuple[float, float, float, float],
+        *,
+        grid_n: int = 7,
+    ) -> list[dict]:
+        """Grid-walk WMS GetFeatureInfo для layer 875838 (ПЗЗ территориальные зоны).
+
+        Returns: list of feature dicts с полями id, geometry, properties.
+        Дедуплицирует по feature id.
+        """
+        return await self.get_features_in_bbox_grid(875838, bbox, grid_n=grid_n)
+
+    # ── 4. list_objects_in_building ───────────────────────────────────────────
     # Q3 deferred — метод реализован, но не вызывается в bulk_harvest_quarter MVP.
     # Готов для per-building помещения/парковка фазы.
 

backend/app/services/cadastre/bulk_harvest.py

@@ -236,6 +236,20 @@ async def harvest_quarter(
             }
         )
 
+    # ── Phase 2.5: grid-walk для territorial_zones (ПЗЗ, layer 875838) ────────
+    # Выполняем после основного grid-walk (Phase 2-3). Требует bbox квартала.
+    quarter_bbox = quarter_bbox_3857(db, quarter)
+    if quarter_bbox is not None:
+        update_progress({"phase": "territorial_zones_started", "quarter": quarter})
+        try:
+            tz_features = await client.get_territorial_zones_in_bbox(quarter_bbox)
+            tz_count = _save_territorial_zones(db, quarter, tz_features)
+            logger.info(
+                "harvest_quarter: territorial_zones quarter=%s upserted=%d", quarter, tz_count
+            )
+        except Exception as e:
+            logger.warning("harvest_quarter: territorial_zones failed quarter=%s: %s", quarter, e)
+
     # ── Phase 4: quarter stats + auto-heal geom из snapshot ─────────────────
     stats_features = [f for f in snapshot.features if f.category_id == CAT_QUARTER_STATS]
     if stats_features:
@@ -1176,6 +1190,85 @@ def upsert_quarter_stats(
     )
 
 
+# ── cad_territorial_zones upsert ────────────────────────────────────────────
+
+
+def _save_territorial_zones(db: Session, quarter_cad: str, features: list[dict]) -> int:
+    """UPSERT territorial_zones features в cad_territorial_zones по zone_id.
+
+    Args:
+        db: SQLAlchemy session.
+        quarter_cad: кадастровый номер квартала (3 сегмента).
+        features: list of raw feature dicts от get_features_in_bbox_grid.
+
+    Returns:
+        Количество успешно upserted строк.
+    """
+    inserted = 0
+    for f in features:
+        props: dict = f.get("properties") or {}
+        geom = f.get("geometry")
+        geom_geojson: str | None = json.dumps(geom) if geom else None
+
+        # zone_id — NSPD feature id или синтетический fallback
+        zone_id = str(
+            props.get("id") or props.get("zone_id") or f.get("id") or f"{quarter_cad}_{inserted}"
+        )
+        zone_code = (
+            props.get("zone_code") or props.get("zone_index") or props.get("reg_numb_border")
+        )
+        zone_name = props.get("zone_name") or props.get("zone_type_name") or props.get("type_zone")
+        permitted_use = props.get("permitted_use") or props.get("vri")
+
+        try:
+            with db.begin_nested():
+                db.execute(
+                    text("""
+                        INSERT INTO cad_territorial_zones
+                            (quarter_cad, zone_id, zone_code, zone_name,
+                             permitted_use, raw_props, geom)
+                        VALUES (
+                            CAST(:quarter_cad AS text),
+                            CAST(:zone_id AS text),
+                            CAST(:zone_code AS text),
+                            CAST(:zone_name AS text),
+                            CAST(:permitted_use AS text),
+                            CAST(:raw_props AS jsonb),
+                            CASE WHEN CAST(:geom AS text) IS NOT NULL
+                                THEN ST_GeogFromGeoJSON(CAST(:geom AS text))
+                                ELSE NULL
+                            END
+                        )
+                        ON CONFLICT (zone_id) DO UPDATE SET
+                            zone_code = EXCLUDED.zone_code,
+                            zone_name = EXCLUDED.zone_name,
+                            permitted_use = EXCLUDED.permitted_use,
+                            raw_props = EXCLUDED.raw_props,
+                            geom = EXCLUDED.geom,
+                            fetched_at = NOW()
+                    """),
+                    {
+                        "quarter_cad": quarter_cad,
+                        "zone_id": zone_id,
+                        "zone_code": zone_code,
+                        "zone_name": zone_name,
+                        "permitted_use": permitted_use,
+                        "raw_props": json.dumps(props, ensure_ascii=False),
+                        "geom": geom_geojson,
+                    },
+                )
+            inserted += 1
+        except Exception as e:
+            logger.warning(
+                "_save_territorial_zones: upsert failed zone_id=%s quarter=%s: %s",
+                zone_id,
+                quarter_cad,
+                e,
+            )
+    db.commit()
+    return inserted
+
+
 # ── Утилиты ──────────────────────────────────────────────────────────────────
 
 

backend/tests/services/cadastre/test_bulk_harvest_territorial.py

@@ -0,0 +1,188 @@
+"""Тесты для _save_territorial_zones (bulk_harvest.py) — mock-based.
+
+Проверяет:
+- Успешный UPSERT 3 features → 3 строки вставлены
+- Повторный вызов → ON CONFLICT обновляет, не дублирует
+- Feature без geometry → строка вставлена с geom=NULL, без краша
+- Feature без zone_id → синтетический fallback zone_id используется
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock
+
+from app.services.cadastre.bulk_harvest import _save_territorial_zones
+
+
+def _make_feature(
+    feature_id: Any = "zone_1",
+    zone_code: str = "Ж-1",
+    zone_name: str = "Жилая смешанная",
+    permitted_use: str = "ИЖС",
+    has_geometry: bool = True,
+) -> dict:
+    """Создать raw feature dict в формате get_features_in_bbox_grid."""
+    geom = (
+        {
+            "type": "Polygon",
+            "coordinates": [
+                [
+                    [6090000.0, 7590000.0],
+                    [6090100.0, 7590000.0],
+                    [6090100.0, 7590100.0],
+                    [6090000.0, 7590100.0],
+                    [6090000.0, 7590000.0],
+                ]
+            ],
+        }
+        if has_geometry
+        else None
+    )
+    return {
+        "id": feature_id,
+        "geometry": geom,
+        "properties": {
+            "zone_code": zone_code,
+            "zone_name": zone_name,
+            "permitted_use": permitted_use,
+        },
+    }
+
+
+def _make_db_mock() -> MagicMock:
+    """Mock SQLAlchemy Session с begin_nested() savepoint support."""
+    db = MagicMock()
+    # begin_nested() используется как context manager
+    savepoint_ctx = MagicMock()
+    savepoint_ctx.__enter__ = MagicMock(return_value=savepoint_ctx)
+    savepoint_ctx.__exit__ = MagicMock(return_value=False)
+    db.begin_nested.return_value = savepoint_ctx
+    return db
+
+
+class TestSaveTerritorialZones:
+    """Тесты для _save_territorial_zones."""
+
+    def test_three_features_inserted(self) -> None:
+        """3 features → returned count == 3, execute вызван 3 раза."""
+        db = _make_db_mock()
+        features = [
+            _make_feature("z1", "Ж-1"),
+            _make_feature("z2", "ОД-1"),
+            _make_feature("z3", "П-1"),
+        ]
+
+        result = _save_territorial_zones(db, "66:41:0204016", features)
+
+        assert result == 3
+        assert db.execute.call_count == 3
+        db.commit.assert_called_once()
+
+    def test_empty_features_list(self) -> None:
+        """Пустой список → 0 inserted, commit всё равно вызван."""
+        db = _make_db_mock()
+
+        result = _save_territorial_zones(db, "66:41:0204016", [])
+
+        assert result == 0
+        db.execute.assert_not_called()
+        db.commit.assert_called_once()
+
+    def test_feature_without_geometry_no_crash(self) -> None:
+        """Feature без geometry → geom=NULL, строка вставлена без краша."""
+        db = _make_db_mock()
+        features = [_make_feature("zone_no_geom", has_geometry=False)]
+
+        result = _save_territorial_zones(db, "66:41:0204016", features)
+
+        assert result == 1
+        # Проверяем что geom параметр передан как None
+        call_kwargs: dict = db.execute.call_args[0][1]
+        assert call_kwargs["geom"] is None
+
+    def test_feature_without_zone_id_uses_fallback(self) -> None:
+        """Feature без id → синтетический fallback zone_id не вызывает краш."""
+        db = _make_db_mock()
+        features = [
+            {
+                "id": None,
+                "geometry": None,
+                "properties": {"zone_code": "Ж-2"},
+            }
+        ]
+
+        result = _save_territorial_zones(db, "66:41:0204016", features)
+
+        assert result == 1
+        call_kwargs = db.execute.call_args[0][1]
+        # fallback zone_id содержит quarter_cad
+        assert "66:41:0204016" in call_kwargs["zone_id"]
+
+    def test_zone_id_from_props_id(self) -> None:
+        """Если feature.id=None, но props['id'] есть — используется props['id']."""
+        db = _make_db_mock()
+        features = [
+            {
+                "id": None,
+                "geometry": None,
+                "properties": {"id": "props_id_42", "zone_code": "Ж-3"},
+            }
+        ]
+
+        result = _save_territorial_zones(db, "66:41:0204016", features)
+
+        assert result == 1
+        call_kwargs = db.execute.call_args[0][1]
+        assert call_kwargs["zone_id"] == "props_id_42"
+
+    def test_execute_error_logged_not_raised(self) -> None:
+        """Exception в execute → строка не вставлена, warning залогирован, не re-raise."""
+        db = _make_db_mock()
+        db.execute.side_effect = RuntimeError("DB error")
+        features = [_make_feature("z_err")]
+
+        # Не должен бросить исключение
+        result = _save_territorial_zones(db, "66:41:0204016", features)
+
+        assert result == 0
+        db.commit.assert_called_once()
+
+    def test_savepoint_used_per_row(self) -> None:
+        """begin_nested() вызывается для каждой строки (SAVEPOINT паттерн)."""
+        db = _make_db_mock()
+        features = [_make_feature(f"z{i}") for i in range(3)]
+
+        _save_territorial_zones(db, "66:41:0204016", features)
+
+        assert db.begin_nested.call_count == 3
+
+    def test_quarter_cad_param_passed(self) -> None:
+        """quarter_cad правильно передаётся в SQL параметры."""
+        db = _make_db_mock()
+        features = [_make_feature("zone_check")]
+
+        _save_territorial_zones(db, "66:41:9999999", features)
+
+        call_kwargs = db.execute.call_args[0][1]
+        assert call_kwargs["quarter_cad"] == "66:41:9999999"
+
+    def test_raw_props_serialized(self) -> None:
+        """raw_props — JSON строка из properties dict."""
+        import json
+
+        db = _make_db_mock()
+        features = [
+            {
+                "id": "z_props",
+                "geometry": None,
+                "properties": {"zone_code": "ОД-2", "extra": "value"},
+            }
+        ]
+
+        _save_territorial_zones(db, "66:41:0204016", features)
+
+        call_kwargs = db.execute.call_args[0][1]
+        raw = json.loads(call_kwargs["raw_props"])
+        assert raw["zone_code"] == "ОД-2"
+        assert raw["extra"] == "value"

data/sql/101_cad_nspd_overlay_tables.sql

@@ -0,0 +1,91 @@
+-- Migration 101: cad_* overlay tables for NSPD layers (epic #263 sub-A)
+--
+-- Creates 5 new tables for NSPD overlay data:
+--   cad_territorial_zones   -- layer 875838 (ПЗЗ территориальные зоны)
+--   cad_red_lines           -- layer 879243 (красные линии)
+--   cad_engineering_structures -- layer 36328 (инженерные сооружения)
+--   cad_risk_zones          -- layers 872xxx (зоны рисков)
+--   cad_opportunity_parcels -- layers 37294, 37298, 37299, 36473, 875845
+--
+-- All tables are idempotent (CREATE TABLE IF NOT EXISTS + CREATE INDEX IF NOT EXISTS).
+-- Migration auto-applies via deploy.yml.
+
+BEGIN;
+
+-- ПЗЗ территориальные зоны (NSPD layer 875838)
+CREATE TABLE IF NOT EXISTS cad_territorial_zones (
+    id BIGSERIAL PRIMARY KEY,
+    quarter_cad TEXT NOT NULL,
+    zone_id TEXT NOT NULL UNIQUE,
+    zone_code TEXT,
+    zone_name TEXT,
+    permitted_use TEXT,
+    raw_props JSONB,
+    geom GEOGRAPHY(POLYGON, 4326),
+    fetched_at TIMESTAMPTZ DEFAULT NOW()
+);
+CREATE INDEX IF NOT EXISTS cad_territorial_zones_geom_gist
+    ON cad_territorial_zones USING GIST(geom);
+CREATE INDEX IF NOT EXISTS cad_territorial_zones_quarter
+    ON cad_territorial_zones (quarter_cad);
+
+-- Red lines (NSPD layer 879243) — placeholder schema
+CREATE TABLE IF NOT EXISTS cad_red_lines (
+    id BIGSERIAL PRIMARY KEY,
+    quarter_cad TEXT NOT NULL,
+    feature_id TEXT UNIQUE,
+    name TEXT,
+    raw_props JSONB,
+    geom GEOGRAPHY(GEOMETRY, 4326),
+    fetched_at TIMESTAMPTZ DEFAULT NOW()
+);
+CREATE INDEX IF NOT EXISTS cad_red_lines_geom_gist
+    ON cad_red_lines USING GIST(geom);
+
+-- Engineering structures (NSPD layer 36328) — для #115 точки подключения
+CREATE TABLE IF NOT EXISTS cad_engineering_structures (
+    id BIGSERIAL PRIMARY KEY,
+    quarter_cad TEXT NOT NULL,
+    cad_num TEXT,
+    name TEXT,
+    structure_type TEXT,
+    raw_props JSONB,
+    geom GEOGRAPHY(GEOMETRY, 4326),
+    fetched_at TIMESTAMPTZ DEFAULT NOW()
+);
+CREATE INDEX IF NOT EXISTS cad_engineering_structures_geom_gist
+    ON cad_engineering_structures USING GIST(geom);
+CREATE INDEX IF NOT EXISTS cad_engineering_structures_quarter
+    ON cad_engineering_structures (quarter_cad);
+
+-- Risk zones (NSPD layers 872xxx)
+CREATE TABLE IF NOT EXISTS cad_risk_zones (
+    id BIGSERIAL PRIMARY KEY,
+    quarter_cad TEXT NOT NULL,
+    feature_id TEXT UNIQUE,
+    risk_type TEXT,
+    layer_id INT,
+    severity TEXT,
+    raw_props JSONB,
+    geom GEOGRAPHY(GEOMETRY, 4326),
+    fetched_at TIMESTAMPTZ DEFAULT NOW()
+);
+CREATE INDEX IF NOT EXISTS cad_risk_zones_geom_gist
+    ON cad_risk_zones USING GIST(geom);
+
+-- Opportunity parcels (NSPD layers 37294, 37298, 37299, 36473, 875845)
+CREATE TABLE IF NOT EXISTS cad_opportunity_parcels (
+    id BIGSERIAL PRIMARY KEY,
+    quarter_cad TEXT NOT NULL,
+    cad_num TEXT,
+    opportunity_type TEXT,
+    layer_id INT,
+    name TEXT,
+    raw_props JSONB,
+    geom GEOGRAPHY(GEOMETRY, 4326),
+    fetched_at TIMESTAMPTZ DEFAULT NOW()
+);
+CREATE INDEX IF NOT EXISTS cad_opportunity_parcels_geom_gist
+    ON cad_opportunity_parcels USING GIST(geom);
+
+COMMIT;