feat(nspd): denorm nspd_parcels + nspd_buildings tables (#94 PR3 of 4)

- data/sql/99_nspd_entities_denorm.sql: 2 tables (cad_num PK) + 6 indexes
  (quarter_cad, GIST geom, partial WHERE purpose ILIKE '%многокв%')
- nspd_denorm.py: denorm_parcel_feature / denorm_building_feature / denorm_dump
  - SAVEPOINT per row via with db.begin_nested()
  - ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:geom), 3857), 4326)
  - json.dumps(geom, ensure_ascii=False) for psycopg bind
  - All CAST(:x AS type) — no ❌:type
- nspd_sync.py: inline denorm after _upsert_dump in harvest_quarter (non-fatal)
- nspd_denorm_backfill.py: Celery one-shot task (per-quarter commit)
- admin_etl.py: POST /api/v1/admin/etl/nspd-denorm-backfill (AdminTokenAuth)
- 19 tests: coerce helpers, denorm parcel/building, aggregate counts

Foundation for downstream features (МКД neighbors lookup, parcel attrs).

Part of #94

backend/app/api/v1/admin_etl.py

@@ -4,6 +4,9 @@ POST /api/v1/admin/etl/objective-backfill
     Запустить fuzzy-match backfill objective_complex_mapping.
     Опциональный REFRESH mv_layout_velocity после успешного backfill.
 
+POST /api/v1/admin/etl/nspd-denorm-backfill
+    Запустить backfill nspd_parcels/nspd_buildings из всех nspd_quarter_dumps.
+
 Header: X-Admin-Token: <SCRAPE_ADMIN_TOKEN>
 """
 
@@ -68,3 +71,26 @@ def run_objective_backfill(
 
     result["mv_rows_after_refresh"] = mv_rows
     return result
+
+
+@router.post("/nspd-denorm-backfill")
+def run_nspd_denorm_backfill(
+    _: AdminTokenAuth,
+    limit: Annotated[
+        int | None,
+        Query(description="Максимум кварталов для обработки (None = все)"),
+    ] = None,
+) -> dict[str, object]:
+    """Запустить Celery backfill: denormalize nspd_quarter_dumps → nspd_parcels/nspd_buildings.
+
+    Задача идемпотентна (ON CONFLICT DO UPDATE). Безопасно запускать повторно.
+    Возвращает Celery task_id — статус через /flower или celery inspect.
+
+    Args:
+        limit: если задан — обработать только первые N кварталов ORDER BY quarter_cad.
+    """
+    from app.workers.tasks.nspd_denorm_backfill import backfill_all_dumps
+
+    task = backfill_all_dumps.apply_async(kwargs={"limit": limit})
+    logger.info("nspd-denorm-backfill enqueued: task_id=%s limit=%s", task.id, limit)
+    return {"task_id": task.id, "status": "enqueued", "limit": limit}

backend/app/services/scrapers/nspd_denorm.py

@@ -0,0 +1,290 @@
+"""Denormalize nspd_quarter_dumps.features_json → nspd_parcels / nspd_buildings.
+
+Используется:
+  - Inline в harvest_quarter task после UPSERT dump'а
+  - Backfill task для existing dumps (backfill_all_dumps)
+
+Property mapping (NSPD WMS GetFeatureInfo responses):
+  Parcels (layer="parcels", layer id 36048):
+    cad_num          — кадастровый номер ЗУ
+    permitted_use    — ВРИ
+    land_category    — категория земель
+    cost_value       — кадастровая стоимость, руб.
+    area             — площадь, м²
+    address          — адрес
+
+  Buildings (layer="buildings", layer id 36049):
+    cad_num                — кадастровый номер ОКС
+    purpose                — назначение ("Многоквартирный дом" / "Нежилое здание" / ...)
+    floors_above_ground    — надземных этажей
+    floors_underground     — подземных этажей
+    year_built             — год постройки
+    cost_value             — кадастровая стоимость, руб.
+    build_record_area      — площадь по ГКН, м²
+    address                — адрес
+
+Геометрия в features_json хранится в EPSG:3857 (как возвращает NSPD WMS) —
+трансформация 3857→4326 делается в SQL через ST_Transform.
+"""
+
+from __future__ import annotations
+
+import datetime
+import json
+import logging
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+# ── Type coercions ─────────────────────────────────────────────────────────────
+
+
+def _coerce_int(v: Any) -> int | None:
+    """NSPD properties могут быть str / int / None."""
+    if v is None:
+        return None
+    try:
+        return int(v)
+    except (ValueError, TypeError):
+        return None
+
+
+def _coerce_numeric(v: Any) -> float | None:
+    """NSPD properties могут быть str с запятой (европейский формат) или None."""
+    if v is None:
+        return None
+    try:
+        return float(str(v).replace(",", ".").strip())
+    except (ValueError, TypeError):
+        return None
+
+
+# ── Parcel UPSERT ──────────────────────────────────────────────────────────────
+
+_PARCEL_UPSERT_SQL = text(
+    """
+    INSERT INTO nspd_parcels (
+        cad_num, quarter_cad, permitted_use, land_category,
+        cost_value, cost_per_m2, area_sqm, address, geom, snapshot_date
+    ) VALUES (
+        CAST(:cad_num AS text),
+        CAST(:quarter_cad AS text),
+        CAST(:permitted_use AS text),
+        CAST(:land_category AS text),
+        CAST(:cost_value AS numeric),
+        CAST(:cost_per_m2 AS numeric),
+        CAST(:area_sqm AS numeric),
+        CAST(:address AS text),
+        CASE WHEN :geom_json IS NULL THEN NULL
+             ELSE ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:geom_json), 3857), 4326)
+        END,
+        CAST(:snapshot_date AS date)
+    )
+    ON CONFLICT (cad_num) DO UPDATE SET
+        quarter_cad   = EXCLUDED.quarter_cad,
+        permitted_use = EXCLUDED.permitted_use,
+        land_category = EXCLUDED.land_category,
+        cost_value    = EXCLUDED.cost_value,
+        cost_per_m2   = EXCLUDED.cost_per_m2,
+        area_sqm      = EXCLUDED.area_sqm,
+        address       = EXCLUDED.address,
+        geom          = EXCLUDED.geom,
+        snapshot_date = EXCLUDED.snapshot_date,
+        updated_at    = NOW()
+    """
+)
+
+
+def denorm_parcel_feature(
+    db: Session,
+    *,
+    feature: dict[str, Any],
+    quarter_cad: str,
+    snapshot_date: str,
+) -> bool:
+    """UPSERT one parcel feature → nspd_parcels.
+
+    Returns True если строка вставлена/обновлена, False при пропуске (нет cad_num).
+    """
+    props = feature.get("properties") or {}
+    cad_num = props.get("cad_num") or props.get("cadastral_number")
+    if not cad_num:
+        return False
+
+    area = _coerce_numeric(props.get("area"))
+    cost_value = _coerce_numeric(props.get("cost_value"))
+    cost_per_m2: float | None = None
+    if cost_value is not None and area is not None and area > 0:
+        cost_per_m2 = cost_value / area
+
+    geom = feature.get("geometry")
+    geom_json: str | None = json.dumps(geom, ensure_ascii=False) if geom else None
+
+    try:
+        with db.begin_nested():
+            db.execute(
+                _PARCEL_UPSERT_SQL,
+                {
+                    "cad_num": cad_num,
+                    "quarter_cad": quarter_cad,
+                    "permitted_use": props.get("permitted_use"),
+                    "land_category": props.get("land_category"),
+                    "cost_value": cost_value,
+                    "cost_per_m2": cost_per_m2,
+                    "area_sqm": area,
+                    "address": props.get("address"),
+                    "geom_json": geom_json,
+                    "snapshot_date": snapshot_date,
+                },
+            )
+        return True
+    except Exception as e:
+        logger.warning("denorm parcel %s failed: %s", cad_num, e)
+        return False
+
+
+# ── Building UPSERT ────────────────────────────────────────────────────────────
+
+_BUILDING_UPSERT_SQL = text(
+    """
+    INSERT INTO nspd_buildings (
+        cad_num, quarter_cad, purpose, floors, floors_underground,
+        year_built, cost_value, build_record_area, address, geom, snapshot_date
+    ) VALUES (
+        CAST(:cad_num AS text),
+        CAST(:quarter_cad AS text),
+        CAST(:purpose AS text),
+        CAST(:floors AS integer),
+        CAST(:floors_underground AS integer),
+        CAST(:year_built AS integer),
+        CAST(:cost_value AS numeric),
+        CAST(:build_record_area AS numeric),
+        CAST(:address AS text),
+        CASE WHEN :geom_json IS NULL THEN NULL
+             ELSE ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:geom_json), 3857), 4326)
+        END,
+        CAST(:snapshot_date AS date)
+    )
+    ON CONFLICT (cad_num) DO UPDATE SET
+        quarter_cad       = EXCLUDED.quarter_cad,
+        purpose           = EXCLUDED.purpose,
+        floors            = EXCLUDED.floors,
+        floors_underground = EXCLUDED.floors_underground,
+        year_built        = EXCLUDED.year_built,
+        cost_value        = EXCLUDED.cost_value,
+        build_record_area = EXCLUDED.build_record_area,
+        address           = EXCLUDED.address,
+        geom              = EXCLUDED.geom,
+        snapshot_date     = EXCLUDED.snapshot_date,
+        updated_at        = NOW()
+    """
+)
+
+
+def denorm_building_feature(
+    db: Session,
+    *,
+    feature: dict[str, Any],
+    quarter_cad: str,
+    snapshot_date: str,
+) -> bool:
+    """UPSERT one building feature → nspd_buildings.
+
+    Returns True если строка вставлена/обновлена, False при пропуске (нет cad_num).
+    """
+    props = feature.get("properties") or {}
+    cad_num = props.get("cad_num") or props.get("cadastral_number")
+    if not cad_num:
+        return False
+
+    geom = feature.get("geometry")
+    geom_json: str | None = json.dumps(geom, ensure_ascii=False) if geom else None
+
+    try:
+        with db.begin_nested():
+            db.execute(
+                _BUILDING_UPSERT_SQL,
+                {
+                    "cad_num": cad_num,
+                    "quarter_cad": quarter_cad,
+                    "purpose": props.get("purpose"),
+                    "floors": _coerce_int(props.get("floors_above_ground")),
+                    "floors_underground": _coerce_int(props.get("floors_underground")),
+                    "year_built": _coerce_int(props.get("year_built")),
+                    "cost_value": _coerce_numeric(props.get("cost_value")),
+                    "build_record_area": _coerce_numeric(props.get("build_record_area")),
+                    "address": props.get("address"),
+                    "geom_json": geom_json,
+                    "snapshot_date": snapshot_date,
+                },
+            )
+        return True
+    except Exception as e:
+        logger.warning("denorm building %s failed: %s", cad_num, e)
+        return False
+
+
+# ── Batch helper ───────────────────────────────────────────────────────────────
+
+
+def denorm_dump(
+    db: Session,
+    *,
+    quarter_cad: str,
+    features: list[dict[str, Any]],
+) -> dict[str, int]:
+    """Denorm all features из одного quarter dump → nspd_parcels / nspd_buildings.
+
+    Итерирует features_json плоский список. Записи с layer="parcels" идут в
+    nspd_parcels, layer="buildings" → nspd_buildings. Остальные layers пропускаются.
+
+    Каждая строка UPSERT'ится в отдельном SAVEPOINT (begin_nested) — ошибка
+    одной строки не откатывает весь batch.
+
+    Args:
+        db: SQLAlchemy Session. Caller отвечает за commit/close после вызова.
+        quarter_cad: 3-сегментный кадастровый квартал.
+        features: плоский list из features_json JSONB (уже декодированный Python list).
+
+    Returns:
+        dict {"parcels": N, "buildings": M, "errors": K} — количество обработанных строк.
+    """
+    snapshot_date = datetime.date.today().isoformat()
+    parcels_n = 0
+    buildings_n = 0
+    errors_n = 0
+
+    for feat in features:
+        layer = feat.get("layer", "")
+        try:
+            if layer == "parcels":
+                if denorm_parcel_feature(
+                    db, feature=feat, quarter_cad=quarter_cad, snapshot_date=snapshot_date
+                ):
+                    parcels_n += 1
+                else:
+                    errors_n += 1
+            elif layer == "buildings":
+                if denorm_building_feature(
+                    db, feature=feat, quarter_cad=quarter_cad, snapshot_date=snapshot_date
+                ):
+                    buildings_n += 1
+                else:
+                    errors_n += 1
+            # Остальные layers (territorial_zones, zouit_*, risk_*, ...) — пропускаем
+        except Exception as e:
+            logger.warning("denorm feature (layer=%s quarter=%s) failed: %s", layer, quarter_cad, e)
+            errors_n += 1
+
+    db.commit()
+    logger.info(
+        "denorm_dump quarter=%s parcels=%d buildings=%d errors=%d",
+        quarter_cad,
+        parcels_n,
+        buildings_n,
+        errors_n,
+    )
+    return {"parcels": parcels_n, "buildings": buildings_n, "errors": errors_n}

backend/app/workers/tasks/nspd_denorm_backfill.py

@@ -0,0 +1,96 @@
+"""One-shot backfill: denormalize all existing nspd_quarter_dumps → nspd_parcels/buildings.
+
+Запускается вручную через admin endpoint POST /api/v1/admin/etl/nspd-denorm-backfill
+или через Celery CLI:
+    celery -A app.workers.celery_app call nspd_denorm.backfill_all_dumps
+
+Идемпотентен: повторный запуск обновляет строки через ON CONFLICT DO UPDATE.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from celery import shared_task
+from sqlalchemy import text
+
+from app.core.db import SessionLocal
+from app.services.scrapers.nspd_denorm import denorm_dump
+
+logger = logging.getLogger(__name__)
+
+
+@shared_task(
+    name="nspd_denorm.backfill_all_dumps",
+    bind=True,
+    soft_time_limit=3600,  # 1 час — много кварталов × ~10ms per parcel
+    max_retries=0,  # One-shot, не retry
+)
+def backfill_all_dumps(self: object, *, limit: int | None = None) -> dict[str, int]:  # type: ignore[misc]
+    """Backfill all existing nspd_quarter_dumps → nspd_parcels / nspd_buildings.
+
+    Читает все строки nspd_quarter_dumps (или limit строк), и для каждой вызывает
+    denorm_dump. Возвращает агрегированные счётчики по всем кварталам.
+
+    Args:
+        limit: если задан — обработать только первые N кварталов (для тестирования).
+
+    Returns:
+        dict {
+            "parcels": суммарно вставлено/обновлено parcel строк,
+            "buildings": суммарно вставлено/обновлено building строк,
+            "errors": суммарно пропусков (нет cad_num, DB ошибки),
+            "quarters_processed": количество обработанных кварталов,
+        }
+    """
+    totals: dict[str, int] = {
+        "parcels": 0,
+        "buildings": 0,
+        "errors": 0,
+        "quarters_processed": 0,
+    }
+
+    db = SessionLocal()
+    try:
+        sql = (
+            "SELECT quarter_cad, features_json "
+            "FROM nspd_quarter_dumps "
+            "WHERE total_features > 0 "
+            "ORDER BY quarter_cad"
+        )
+        if limit is not None:
+            # Используем параметр через text() чтобы избежать SQL-injection.
+            rows = db.execute(
+                text(sql + " LIMIT CAST(:lim AS integer)"), {"lim": int(limit)}
+            ).fetchall()
+        else:
+            rows = db.execute(text(sql)).fetchall()
+
+        logger.info("backfill_all_dumps: starting, quarters=%d limit=%s", len(rows), limit)
+
+        for row in rows:
+            quarter_cad: str = row[0]
+            features_json = row[1]
+            # features_json — уже list[dict] через psycopg v3 JSON decoding.
+            features: list[dict] = features_json if isinstance(features_json, list) else []
+
+            try:
+                counts = denorm_dump(db, quarter_cad=quarter_cad, features=features)
+                totals["parcels"] += counts["parcels"]
+                totals["buildings"] += counts["buildings"]
+                totals["errors"] += counts["errors"]
+                totals["quarters_processed"] += 1
+            except Exception as e:
+                logger.warning("backfill_all_dumps: quarter=%s failed: %s", quarter_cad, e)
+                totals["errors"] += 1
+                # Продолжаем с остальными кварталами
+                try:
+                    db.rollback()
+                except Exception:
+                    pass
+
+        logger.info("backfill_all_dumps: done totals=%s", totals)
+    finally:
+        db.close()
+
+    return totals

backend/app/workers/tasks/nspd_sync.py

@@ -33,6 +33,7 @@ from app.services.scrapers.nspd_client import (
     NspdLiteWafError,
     QuarterDump,
 )
+from app.services.scrapers.nspd_denorm import denorm_dump
 from app.workers.celery_app import celery_app
 
 logger = logging.getLogger(__name__)
@@ -301,6 +302,32 @@ def harvest_quarter(
 
         _upsert_dump(quarter_cad, region_code, dump, features_json, duration_ms, None)
 
+        # Inline denorm: разложить parcels/buildings из dump в nspd_parcels/nspd_buildings.
+        # Ошибка denorm не должна фейлить весь harvest — только warning.
+        try:
+            denorm_db = SessionLocal()
+            try:
+                denorm_counts = denorm_dump(
+                    denorm_db,
+                    quarter_cad=quarter_cad,
+                    features=features_json or [],
+                )
+                logger.info(
+                    "harvest_quarter denorm: cad=%s parcels=%d buildings=%d errors=%d",
+                    quarter_cad,
+                    denorm_counts["parcels"],
+                    denorm_counts["buildings"],
+                    denorm_counts["errors"],
+                )
+            finally:
+                denorm_db.close()
+        except Exception as denorm_exc:
+            logger.warning(
+                "harvest_quarter denorm failed (non-fatal): cad=%s error=%s",
+                quarter_cad,
+                denorm_exc,
+            )
+
         logger.info(
             "harvest_quarter done: cad=%s region=%d duration=%dms total=%d",
             quarter_cad,

backend/tests/services/test_nspd_denorm.py

@@ -0,0 +1,322 @@
+"""Тесты для nspd_denorm.py — coerce helpers + denorm_parcel/building/dump.
+
+Не требует реального PostgreSQL — мокает db.execute() и db.begin_nested().
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+
+from app.services.scrapers.nspd_denorm import (
+    _coerce_int,
+    _coerce_numeric,
+    denorm_building_feature,
+    denorm_dump,
+    denorm_parcel_feature,
+)
+
+# ── _coerce_int ────────────────────────────────────────────────────────────────
+
+
+def test_coerce_int_str_number() -> None:
+    assert _coerce_int("5") == 5
+
+
+def test_coerce_int_int_passthrough() -> None:
+    assert _coerce_int(17) == 17
+
+
+def test_coerce_int_none_returns_none() -> None:
+    assert _coerce_int(None) is None
+
+
+def test_coerce_int_invalid_returns_none() -> None:
+    assert _coerce_int("не_число") is None
+
+
+def test_coerce_int_float_truncates() -> None:
+    assert _coerce_int(3.9) == 3
+
+
+# ── _coerce_numeric ────────────────────────────────────────────────────────────
+
+
+def test_coerce_numeric_float_passthrough() -> None:
+    assert _coerce_numeric(12.5) == 12.5
+
+
+def test_coerce_numeric_str_dot() -> None:
+    assert _coerce_numeric("1234567.89") == pytest.approx(1234567.89)
+
+
+def test_coerce_numeric_comma_decimal() -> None:
+    """Европейский формат с запятой → корректный float."""
+    assert _coerce_numeric("1234567,89") == pytest.approx(1234567.89)
+
+
+def test_coerce_numeric_none_returns_none() -> None:
+    assert _coerce_numeric(None) is None
+
+
+def test_coerce_numeric_invalid_returns_none() -> None:
+    assert _coerce_numeric("N/A") is None
+
+
+# ── Helpers для моков ──────────────────────────────────────────────────────────
+
+
+def _make_mock_session() -> MagicMock:
+    """Мок Session: begin_nested() — context manager, execute() → MagicMock."""
+    db = MagicMock()
+    # begin_nested() должен быть context manager
+    cm = MagicMock()
+    cm.__enter__ = MagicMock(return_value=None)
+    cm.__exit__ = MagicMock(return_value=False)
+    db.begin_nested.return_value = cm
+    return db
+
+
+def _parcel_feature(
+    cad_num: str = "66:41:0204016:10",
+    area: Any = "500.0",
+    cost_value: Any = "1500000",
+    **extra_props: Any,
+) -> dict[str, Any]:
+    props: dict[str, Any] = {
+        "cad_num": cad_num,
+        "area": area,
+        "cost_value": cost_value,
+        "permitted_use": "Для ИЖС",
+        "land_category": "Земли населённых пунктов",
+        "address": "г. Екатеринбург",
+        **extra_props,
+    }
+    return {
+        "layer": "parcels",
+        "feature_id": "123",
+        "geometry": {"type": "Polygon", "coordinates": [[[0, 0], [1, 0], [1, 1], [0, 0]]]},
+        "properties": props,
+    }
+
+
+def _building_feature(
+    cad_num: str = "66:41:0204016:10:1",
+    purpose: str = "Многоквартирный дом",
+    **extra_props: Any,
+) -> dict[str, Any]:
+    props: dict[str, Any] = {
+        "cad_num": cad_num,
+        "purpose": purpose,
+        "floors_above_ground": 9,
+        "floors_underground": 1,
+        "year_built": 1985,
+        "cost_value": "50000000",
+        "build_record_area": "3200.0",
+        "address": "ул. Ленина 1",
+        **extra_props,
+    }
+    return {
+        "layer": "buildings",
+        "feature_id": "456",
+        "geometry": {"type": "Polygon", "coordinates": [[[0, 0], [1, 0], [1, 1], [0, 0]]]},
+        "properties": props,
+    }
+
+
+# ── denorm_parcel_feature ──────────────────────────────────────────────────────
+
+
+def test_denorm_parcel_feature_inserts() -> None:
+    """Полный feature с cad_num → вызывает db.execute с правильными параметрами."""
+    db = _make_mock_session()
+
+    result = denorm_parcel_feature(
+        db,
+        feature=_parcel_feature(),
+        quarter_cad="66:41:0204016",
+        snapshot_date="2026-05-16",
+    )
+
+    assert result is True
+    db.begin_nested.assert_called_once()
+    db.execute.assert_called_once()
+    # Проверяем что параметры содержат нужные ключи
+    call_kwargs = db.execute.call_args[0][1]  # positional dict
+    assert call_kwargs["cad_num"] == "66:41:0204016:10"
+    assert call_kwargs["quarter_cad"] == "66:41:0204016"
+    assert call_kwargs["permitted_use"] == "Для ИЖС"
+    assert call_kwargs["land_category"] == "Земли населённых пунктов"
+    assert call_kwargs["area_sqm"] == 500.0
+    assert call_kwargs["cost_value"] == 1500000.0
+    assert call_kwargs["cost_per_m2"] == pytest.approx(3000.0)
+    assert call_kwargs["snapshot_date"] == "2026-05-16"
+
+
+def test_denorm_parcel_no_cad_num_skipped() -> None:
+    """Feature без cad_num → возвращает False, db не вызывается."""
+    db = _make_mock_session()
+    feature: dict[str, Any] = {
+        "layer": "parcels",
+        "feature_id": None,
+        "geometry": None,
+        "properties": {"area": "100"},
+    }
+    result = denorm_parcel_feature(
+        db, feature=feature, quarter_cad="66:41:0204016", snapshot_date="2026-05-16"
+    )
+    assert result is False
+    db.execute.assert_not_called()
+
+
+def test_denorm_parcel_zero_area_cost_per_m2_none() -> None:
+    """area=0 → cost_per_m2 = None (нет деления на ноль)."""
+    db = _make_mock_session()
+    result = denorm_parcel_feature(
+        db,
+        feature=_parcel_feature(area="0", cost_value="1000"),
+        quarter_cad="66:41:0204016",
+        snapshot_date="2026-05-16",
+    )
+    assert result is True
+    call_kwargs = db.execute.call_args[0][1]
+    assert call_kwargs["cost_per_m2"] is None
+
+
+def test_denorm_parcel_null_geometry() -> None:
+    """feature без geometry → geom_json=None, INSERT всё равно вызывается."""
+    db = _make_mock_session()
+    feature = _parcel_feature()
+    feature["geometry"] = None
+
+    result = denorm_parcel_feature(
+        db, feature=feature, quarter_cad="66:41:0204016", snapshot_date="2026-05-16"
+    )
+    assert result is True
+    call_kwargs = db.execute.call_args[0][1]
+    assert call_kwargs["geom_json"] is None
+
+
+def test_denorm_parcel_db_exception_returns_false() -> None:
+    """DB execute raises → возвращает False (не propagate)."""
+    db = _make_mock_session()
+    db.execute.side_effect = Exception("DB constraint violation")
+
+    result = denorm_parcel_feature(
+        db,
+        feature=_parcel_feature(),
+        quarter_cad="66:41:0204016",
+        snapshot_date="2026-05-16",
+    )
+    assert result is False
+
+
+# ── denorm_building_feature ────────────────────────────────────────────────────
+
+
+def test_denorm_building_feature_inserts() -> None:
+    """Полный building feature → INSERT с правильными параметрами."""
+    db = _make_mock_session()
+
+    result = denorm_building_feature(
+        db,
+        feature=_building_feature(),
+        quarter_cad="66:41:0204016",
+        snapshot_date="2026-05-16",
+    )
+
+    assert result is True
+    db.execute.assert_called_once()
+    call_kwargs = db.execute.call_args[0][1]
+    assert call_kwargs["cad_num"] == "66:41:0204016:10:1"
+    assert call_kwargs["purpose"] == "Многоквартирный дом"
+    assert call_kwargs["floors"] == 9
+    assert call_kwargs["floors_underground"] == 1
+    assert call_kwargs["year_built"] == 1985
+    assert call_kwargs["cost_value"] == 50000000.0
+    assert call_kwargs["build_record_area"] == 3200.0
+
+
+def test_denorm_building_no_cad_num_skipped() -> None:
+    """Building feature без cad_num → False."""
+    db = _make_mock_session()
+    feature: dict[str, Any] = {
+        "layer": "buildings",
+        "feature_id": None,
+        "geometry": None,
+        "properties": {"purpose": "Нежилое здание"},
+    }
+    result = denorm_building_feature(
+        db, feature=feature, quarter_cad="66:41:0204016", snapshot_date="2026-05-16"
+    )
+    assert result is False
+    db.execute.assert_not_called()
+
+
+def test_denorm_building_str_floors_coerced() -> None:
+    """floors_above_ground строкой → корректно парсится в int."""
+    db = _make_mock_session()
+    result = denorm_building_feature(
+        db,
+        feature=_building_feature(floors_above_ground="12"),
+        quarter_cad="66:41:0204016",
+        snapshot_date="2026-05-16",
+    )
+    assert result is True
+    call_kwargs = db.execute.call_args[0][1]
+    assert call_kwargs["floors"] == 12
+
+
+# ── denorm_dump ────────────────────────────────────────────────────────────────
+
+
+def test_denorm_dump_aggregates() -> None:
+    """3 parcels + 2 buildings + 1 unknown layer → правильные счётчики."""
+    db = _make_mock_session()
+
+    features: list[dict[str, Any]] = [
+        _parcel_feature("66:41:0101001:1"),
+        _parcel_feature("66:41:0101001:2"),
+        _parcel_feature("66:41:0101001:3"),
+        _building_feature("66:41:0101001:1:1"),
+        _building_feature("66:41:0101001:1:2"),
+        {
+            "layer": "territorial_zones",
+            "feature_id": "tz-1",
+            "geometry": None,
+            "properties": {"type_zone": "Ж-1"},
+        },
+    ]
+
+    counts = denorm_dump(db, quarter_cad="66:41:0101001", features=features)
+
+    assert counts["parcels"] == 3
+    assert counts["buildings"] == 2
+    # territorial_zones не считается как ошибка — просто пропускается
+    assert counts["errors"] == 0
+    db.commit.assert_called_once()
+
+
+def test_denorm_dump_empty_features() -> None:
+    """Пустой список features → нули, commit всё равно вызывается."""
+    db = _make_mock_session()
+    counts = denorm_dump(db, quarter_cad="66:41:0101001", features=[])
+
+    assert counts == {"parcels": 0, "buildings": 0, "errors": 0}
+    db.commit.assert_called_once()
+
+
+def test_denorm_dump_no_cad_num_counted_as_error() -> None:
+    """Parcel без cad_num → denorm_parcel_feature returns False → errors += 1."""
+    db = _make_mock_session()
+    feature: dict[str, Any] = {
+        "layer": "parcels",
+        "feature_id": None,
+        "geometry": None,
+        "properties": {},
+    }
+    counts = denorm_dump(db, quarter_cad="66:41:0101001", features=[feature])
+    assert counts["parcels"] == 0
+    assert counts["errors"] == 1

data/sql/99_nspd_entities_denorm.sql

@@ -0,0 +1,171 @@
+-- 99_nspd_entities_denorm.sql
+-- Context : Denormalized entity tables для NSPD parcel и building features.
+--           nspd_quarter_dumps.features_json хранит все feature'ы в JSONB; эти
+--           таблицы дают быстрый lookup по cad_num без json-unpacking на query time.
+--           Источник: layer="parcels" (NSPD layer 36048) и layer="buildings" (36049)
+--           из features_json. Геометрия трансформируется 3857→4326 при UPSERT.
+--
+-- Dependencies: 88_nspd_quarter_dumps.sql (logical; таблица должна существовать до backfill)
+-- Deploy order: после 98_nspd_quarter_dumps_opportunity_flag.sql
+-- Idempotent: yes — IF NOT EXISTS на всех объектах
+-- Related:
+--   - backend/app/services/scrapers/nspd_denorm.py :: denorm_parcel_feature / denorm_building_feature
+--   - backend/app/workers/tasks/nspd_denorm_backfill.py :: backfill_all_dumps
+--   - backend/app/workers/tasks/nspd_sync.py :: harvest_quarter (inline denorm)
+-- Issue: #94 PR3
+
+BEGIN;
+
+-- ── nspd_parcels ──────────────────────────────────────────────────────────────
+-- Denormalized parcel features из NSPD quarter dumps (layer="parcels", id 36048).
+-- Каждая строка = один земельный участок из любого квартала.
+-- PK cad_num: один участок может пересекать несколько кварталов, берём
+-- последний harvest (ON CONFLICT DO UPDATE).
+
+CREATE TABLE IF NOT EXISTS nspd_parcels (
+    -- Кадастровый номер участка (4-сегментный, e.g. '66:41:0204016:10').
+    -- Первичный ключ — natural, стабильный идентификатор в ЕГРН.
+    cad_num             TEXT PRIMARY KEY,
+
+    -- Квартал из которого была взята последняя версия этого участка.
+    quarter_cad         TEXT NOT NULL,
+
+    -- ВРИ (вид разрешённого использования) — properties.permitted_use из NSPD.
+    -- Nullable: НСПД не всегда возвращает ВРИ для старых записей.
+    permitted_use       TEXT,
+
+    -- Категория земель (properties.land_category).
+    -- Nullable: может отсутствовать в ответе.
+    land_category       TEXT,
+
+    -- Кадастровая стоимость ЗУ (properties.cost_value), руб.
+    cost_value          NUMERIC,
+
+    -- Кадастровая стоимость за м² = cost_value / area_sqm.
+    -- Вычисляется при UPSERT; NULL если area_sqm = 0 или cost_value = NULL.
+    cost_per_m2         NUMERIC,
+
+    -- Площадь участка (properties.area), м².
+    area_sqm            NUMERIC,
+
+    -- Адрес участка (properties.address).
+    address             TEXT,
+
+    -- Геометрия участка в WGS84 (трансформируется из EPSG:3857 при UPSERT).
+    -- Geometry не Polygon т.к. НСПД иногда возвращает MultiPolygon или Point.
+    geom                GEOMETRY(Geometry, 4326),
+
+    -- Дата snapshot'а (день harvest'а).
+    snapshot_date       DATE NOT NULL DEFAULT CURRENT_DATE,
+
+    -- Timestamp последнего UPSERT.
+    updated_at          TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+COMMENT ON TABLE nspd_parcels IS
+    'Denormalized parcel features из NSPD quarter dumps (layer=parcels, id 36048). '
+    'PK = cad_num (последний harvest wins). Заполняется Celery task backfill_all_dumps '
+    'или inline в harvest_quarter. Источник: features_json JSONB колонка nspd_quarter_dumps. '
+    'Геометрия трансформирована EPSG:3857→4326. (#94 PR3)';
+
+COMMENT ON COLUMN nspd_parcels.cad_num IS
+    'Кадастровый номер участка (4-сегментный). Natural PK из НСПД.';
+
+COMMENT ON COLUMN nspd_parcels.quarter_cad IS
+    '3-сегментный квартал последнего harvest'а. Используется для debug/audit.';
+
+COMMENT ON COLUMN nspd_parcels.cost_per_m2 IS
+    'Кадастровая стоимость за м² (cost_value / area_sqm). '
+    'NULL если area_sqm = 0 или cost_value отсутствует. Вычисляется в Python при UPSERT.';
+
+-- Индекс для lookup по кварталу (join с nspd_quarter_dumps).
+CREATE INDEX IF NOT EXISTS idx_nspd_parcels_quarter_cad
+    ON nspd_parcels (quarter_cad);
+
+-- GIST индекс на геометрию для spatial queries (соседние участки, ST_DWithin).
+CREATE INDEX IF NOT EXISTS idx_nspd_parcels_geom_gist
+    ON nspd_parcels USING GIST (geom);
+
+-- Partial B-tree на cost_per_m2 для ценовых аналитических запросов.
+CREATE INDEX IF NOT EXISTS idx_nspd_parcels_cost_per_m2
+    ON nspd_parcels (cost_per_m2)
+    WHERE cost_per_m2 IS NOT NULL;
+
+
+-- ── nspd_buildings ────────────────────────────────────────────────────────────
+-- Denormalized building features из NSPD quarter dumps (layer="buildings", id 36049).
+-- Каждая строка = одно здание/сооружение ЕГРН.
+
+CREATE TABLE IF NOT EXISTS nspd_buildings (
+    -- Кадастровый номер здания (5-сегментный, e.g. '66:41:0204016:10:1').
+    cad_num             TEXT PRIMARY KEY,
+
+    -- Квартал последнего harvest'а.
+    quarter_cad         TEXT NOT NULL,
+
+    -- Назначение здания (properties.purpose).
+    -- Ключевые значения: 'Многоквартирный дом', 'Нежилое здание', etc.
+    purpose             TEXT,
+
+    -- Количество надземных этажей (properties.floors_above_ground).
+    floors              INTEGER,
+
+    -- Количество подземных этажей (properties.floors_underground).
+    floors_underground  INTEGER,
+
+    -- Год постройки (properties.year_built).
+    year_built          INTEGER,
+
+    -- Кадастровая стоимость здания (properties.cost_value), руб.
+    cost_value          NUMERIC,
+
+    -- Площадь по данным ГКН (properties.build_record_area), м².
+    build_record_area   NUMERIC,
+
+    -- Адрес здания (properties.address).
+    address             TEXT,
+
+    -- Геометрия здания в WGS84 (трансформируется из EPSG:3857 при UPSERT).
+    geom                GEOMETRY(Geometry, 4326),
+
+    -- Дата snapshot'а.
+    snapshot_date       DATE NOT NULL DEFAULT CURRENT_DATE,
+
+    -- Timestamp последнего UPSERT.
+    updated_at          TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+COMMENT ON TABLE nspd_buildings IS
+    'Denormalized building features из NSPD quarter dumps (layer=buildings, id 36049). '
+    'PK = cad_num (последний harvest wins). Заполняется Celery task backfill_all_dumps '
+    'или inline в harvest_quarter. Ключевой use-case: МКД-соседи для P2-scoring. '
+    'Геометрия трансформирована EPSG:3857→4326. (#94 PR3)';
+
+COMMENT ON COLUMN nspd_buildings.purpose IS
+    'Назначение здания (properties.purpose). Для МКД = "Многоквартирный дом". '
+    'Используется в partial index idx_nspd_buildings_mkd для быстрой фильтрации.';
+
+COMMENT ON COLUMN nspd_buildings.floors IS
+    'Количество надземных этажей (properties.floors_above_ground из НСПД). '
+    'Nullable: старые ОКС в ЕГРН не всегда имеют этажность.';
+
+-- Индекс для lookup по кварталу.
+CREATE INDEX IF NOT EXISTS idx_nspd_buildings_quarter_cad
+    ON nspd_buildings (quarter_cad);
+
+-- GIST индекс на геометрию (МКД-соседи через ST_DWithin).
+CREATE INDEX IF NOT EXISTS idx_nspd_buildings_geom_gist
+    ON nspd_buildings USING GIST (geom);
+
+-- Partial index для МКД (основной use-case P2-scoring).
+-- ILIKE не работает в partial index WHERE clause — используем точный строковый prefixmatch.
+-- Здания с purpose IS NULL или другим purpose — в idx_nspd_buildings_geom_gist.
+CREATE INDEX IF NOT EXISTS idx_nspd_buildings_mkd
+    ON nspd_buildings (quarter_cad)
+    WHERE purpose = 'Многоквартирный дом';
+
+COMMENT ON INDEX idx_nspd_buildings_mkd IS
+    'Partial index для быстрого lookup МКД по кварталу (P2-scoring neighbors). '
+    'WHERE purpose = exact строка из НСПД. (#94 PR3)';
+
+COMMIT;