backend/app/api/v1/parcels.py

@@ -22,6 +22,10 @@ from app.services.site_finder.cadastre_fetch import (
 from app.services.site_finder.cadastre_fetch import (
     fetch_status as _fetch_status,
 )
+from app.services.site_finder.quarter_dump_lookup import (
+    get_quarter_dump_data,
+    make_empty_result,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -1529,6 +1533,14 @@ def analyze_parcel(
         logger.warning("utilities query failed for %s: %s", cad_num, e)
         utilities = None
 
+    # 9e) NSPD quarter dump — ПЗЗ зона + ЗОУИТ + инженерные сооружения из кеша
+    try:
+        nspd_dump_data = get_quarter_dump_data(db, cad_num, geom_wkt)
+    except Exception as e:
+        logger.warning("nspd quarter dump lookup failed for %s: %s", cad_num, e)
+        # Independent dict per request — never mutate module singleton.
+        nspd_dump_data = make_empty_result()
+
     # 10) Market trend — динамика цен ДДУ в радиусе 3 км за 6 vs предыдущие 6 месяцев
     market_trend: dict[str, Any] | None = None
     try:
@@ -1831,6 +1843,15 @@ def analyze_parcel(
         "confidence_label": confidence_info["label"],
         "confidence_breakdown": confidence_info["breakdown"],
         "confidence_caveats": confidence_info["caveats"],
+        # Sprint 1.1 item #4 — NSPD quarter dump fields
+        # nspd_zoning: ПЗЗ зона из territorial_zones кеша (G1)
+        # nspd_zouit_overlaps: ЗОУИТ пересечения (G3)
+        # nspd_engineering_nearby: инженерные сооружения в 200м (I3)
+        # nspd_dump: freshness metadata — available, stale, harvest_triggered
+        "nspd_zoning": nspd_dump_data["nspd_zoning"],
+        "nspd_zouit_overlaps": nspd_dump_data["nspd_zouit_overlaps"],
+        "nspd_engineering_nearby": nspd_dump_data["nspd_engineering_nearby"],
+        "nspd_dump": nspd_dump_data["nspd_dump"],
     }
 
 

backend/app/services/site_finder/quarter_dump_lookup.py

@@ -0,0 +1,417 @@
+"""Quarter dump lookup helper for analyze_parcel.
+
+Sprint 1.1 item #4 (feat/analyze-uses-quarter-dump):
+Читает nspd_quarter_dumps кеш и извлекает:
+- nspd_zoning   — территориальная зона ПЗЗ (G1) по centroid участка
+- nspd_zouit_overlaps — список ЗОУИТ (G3) которые пересекаются с участком
+- nspd_engineering_nearby — инженерные сооружения в 200м (I3)
+- nspd_dump     — freshness metadata (доступность, возраст, trigger флаг)
+
+Если дамп отсутствует или устарел (>180 дней) — fire-and-forget
+harvest_quarter.apply_async() и продолжает без dump-derived полей.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+# Порог свежести дампа — 180 дней. Совпадает с beat-расписанием (раз в квартал
+# на практике, но с запасом для редко запрашиваемых кварталов).
+_DUMP_MAX_AGE_DAYS = 180
+
+# Радиус поиска инженерных сооружений (метры) — договорённость #44 I3.
+_ENGINEERING_RADIUS_M = 200
+
+# Sentinel для isinstance-проверок и read-only fallback в parcels.py try/except.
+# НИКОГДА не мутировать — использовать make_empty_result() для новых dict.
+EMPTY_DUMP_RESULT: dict[str, Any] = {
+    "nspd_zoning": None,
+    "nspd_zouit_overlaps": [],
+    "nspd_engineering_nearby": [],
+    "nspd_dump": {
+        "available": False,
+        "fetched_at_utc": None,
+        "stale": False,
+        "harvest_triggered": False,
+        "total_features": None,
+    },
+}
+
+
+def make_empty_result(
+    *,
+    fetched_at_utc: str | None = None,
+    stale: bool = False,
+    harvest_triggered: bool = False,
+    total_features: int | None = None,
+) -> dict[str, Any]:
+    """Создаёт свежую копию empty-dump result с возможностью переопределить поля.
+
+    Вызывать вместо EMPTY_DUMP_RESULT напрямую — чтобы не мутировать singleton.
+    """
+    return {
+        "nspd_zoning": None,
+        "nspd_zouit_overlaps": [],
+        "nspd_engineering_nearby": [],
+        "nspd_dump": {
+            "available": False,
+            "fetched_at_utc": fetched_at_utc,
+            "stale": stale,
+            "harvest_triggered": harvest_triggered,
+            "total_features": total_features,
+        },
+    }
+
+
+def derive_quarter_cad(cad_num: str) -> str | None:
+    """3-сегментный кадастровый квартал из любого кадастрового номера.
+
+    - 3-сегмент (квартал)  '66:41:0204016'          → '66:41:0204016'
+    - 4-сегмент (участок)  '66:41:0204016:10'        → '66:41:0204016'
+    - 5-сегмент (здание)   '66:41:0204016:10:1'      → '66:41:0204016'
+    - невалидный формат                               → None
+    """
+    parts = cad_num.strip().split(":")
+    if len(parts) < 3:
+        return None
+    quarter = ":".join(parts[:3])
+    # Каждый сегмент — только цифры
+    if not all(p.isdigit() and len(p) >= 1 for p in parts[:3]):
+        return None
+    return quarter
+
+
+def get_quarter_dump_data(
+    db: Session,
+    cad_num: str,
+    parcel_wkt: str | None,
+) -> dict[str, Any]:
+    """Читает quarter dump для квартала cad_num и возвращает НСПД-контекст.
+
+    Returns dict с ключами:
+    - nspd_zoning: dict | None  — зона ПЗЗ из territorial_zones
+    - nspd_zouit_overlaps: list[dict]  — ЗОУИТ пересечения
+    - nspd_engineering_nearby: list[dict]  — инженерные сооружения в 200м
+    - nspd_dump: dict  — freshness metadata
+
+    Если дамп отсутствует или устарел — вызывает harvest_quarter.apply_async()
+    (non-blocking) и возвращает пустые spatial поля + nspd_dump.available=False.
+    Если parcel_wkt=None — возвращает только freshness metadata (нет geom для
+    spatial queries).
+    """
+    quarter = derive_quarter_cad(cad_num)
+    if quarter is None:
+        logger.warning("quarter_dump_lookup: cannot derive quarter from cad=%s", cad_num)
+        return make_empty_result()
+
+    # Читаем строку дампа из БД. Денормализованные счётчики слоёв используются
+    # для early-exit в spatial helpers (M2 mitigation).
+    row = db.execute(
+        text(
+            """
+            SELECT quarter_cad,
+                   fetched_at_utc,
+                   total_features,
+                   harvest_error,
+                   territorial_zones_count,
+                   zouit_count,
+                   engineering_count
+              FROM nspd_quarter_dumps
+             WHERE quarter_cad = :q
+            """
+        ),
+        {"q": quarter},
+    ).first()
+
+    now = datetime.now(UTC)
+    max_age = timedelta(days=_DUMP_MAX_AGE_DAYS)
+
+    if row is None:
+        # Дампа нет — ставим harvest в очередь
+        harvest_triggered = _trigger_harvest(quarter)
+        return make_empty_result(harvest_triggered=harvest_triggered)
+
+    fetched_at: datetime = row[1]
+    # Убедимся что timezone-aware для корректного сравнения
+    if fetched_at.tzinfo is None:
+        fetched_at = fetched_at.replace(tzinfo=UTC)
+
+    total_features: int | None = row[2]
+    harvest_error: str | None = row[3]
+    territorial_zones_count: int = row[4] or 0
+    zouit_count: int = row[5] or 0
+    engineering_count: int = row[6] or 0
+
+    is_stale = (now - fetched_at) > max_age
+    has_error = harvest_error is not None
+
+    # Устаревший или с ошибкой — триггерим повторный harvest
+    if is_stale or has_error:
+        harvest_triggered = _trigger_harvest(quarter)
+        return make_empty_result(
+            fetched_at_utc=fetched_at.isoformat(),
+            stale=is_stale,
+            harvest_triggered=harvest_triggered,
+            total_features=total_features,
+        )
+
+    # Свежий дамп без ошибок — извлекаем spatial данные
+    dump_meta: dict[str, Any] = {
+        "available": True,
+        "fetched_at_utc": fetched_at.isoformat(),
+        "stale": False,
+        "harvest_triggered": False,
+        "total_features": total_features,
+    }
+
+    if parcel_wkt is None:
+        # Нет геометрии участка — возвращаем только метаданные
+        return {
+            "nspd_zoning": None,
+            "nspd_zouit_overlaps": [],
+            "nspd_engineering_nearby": [],
+            "nspd_dump": dump_meta,
+        }
+
+    layer_counts = {
+        "territorial_zones_count": territorial_zones_count,
+        "zouit_count": zouit_count,
+        "engineering_count": engineering_count,
+    }
+    nspd_zoning = _get_zoning(db, quarter, parcel_wkt, layer_counts)
+    nspd_zouit = _get_zouit_overlaps(db, quarter, parcel_wkt, layer_counts)
+    nspd_engineering = _get_engineering_nearby(db, quarter, parcel_wkt, layer_counts)
+
+    return {
+        "nspd_zoning": nspd_zoning,
+        "nspd_zouit_overlaps": nspd_zouit,
+        "nspd_engineering_nearby": nspd_engineering,
+        "nspd_dump": dump_meta,
+    }
+
+
+# ── Spatial helpers ───────────────────────────────────────────────────────────
+
+
+def _get_zoning(
+    db: Session,
+    quarter: str,
+    parcel_wkt: str,
+    layer_counts: dict[str, int] | None = None,
+) -> dict[str, Any] | None:
+    """G1: ПЗЗ территориальная зона по centroid участка из dump.
+
+    Geometry в features_json в EPSG:3857 — ST_Transform на чтении → 4326.
+    layer_counts — денормализованные счётчики из строки дампа. Если
+    territorial_zones_count == 0 — пропускаем heavy jsonb_array_elements scan.
+    """
+    if layer_counts is not None and layer_counts.get("territorial_zones_count", 1) == 0:
+        return None
+    try:
+        row = db.execute(
+            text(
+                """
+                SELECT feat.value->'properties' AS zone_props
+                  FROM nspd_quarter_dumps d,
+                       jsonb_array_elements(d.features_json) AS feat(value)
+                 WHERE d.quarter_cad = :q
+                   AND feat.value->>'layer' = 'territorial_zones'
+                   AND (feat.value->'geometry') IS NOT NULL
+                   AND feat.value->>'geometry' != 'null'
+                   AND ST_Intersects(
+                       ST_Transform(
+                           ST_SetSRID(
+                               ST_GeomFromGeoJSON(feat.value->>'geometry'),
+                               3857
+                           ),
+                           4326
+                       ),
+                       ST_Centroid(ST_GeomFromText(:wkt, 4326))
+                   )
+                 LIMIT 1
+                """
+            ),
+            {"q": quarter, "wkt": parcel_wkt},
+        ).first()
+
+        if row is None:
+            return None
+
+        props: dict[str, Any] = row[0] if isinstance(row[0], dict) else {}
+        zone_code = props.get("reg_numb_border") or props.get("zone_code") or props.get("name")
+        zone_name = props.get("type_zone") or props.get("zone_name") or props.get("name")
+        return {
+            "zone_code": zone_code,
+            "zone_name": zone_name,
+            "source": "nspd-quarter-dump",
+            "raw_props": props,
+        }
+    except Exception as e:
+        logger.warning("nspd zoning query failed for quarter=%s: %s", quarter, e)
+        return None
+
+
+def _get_zouit_overlaps(
+    db: Session,
+    quarter: str,
+    parcel_wkt: str,
+    layer_counts: dict[str, int] | None = None,
+) -> list[dict[str, Any]]:
+    """G3: ЗОУИТ которые пересекают участок.
+
+    Проверяем 5 групп: zouit_okn, zouit_engineering, zouit_natural,
+    zouit_protected, zouit_other.
+    layer_counts — денормализованные счётчики. Если zouit_count == 0 —
+    пропускаем heavy jsonb_array_elements scan.
+    """
+    if layer_counts is not None and layer_counts.get("zouit_count", 1) == 0:
+        return []
+    try:
+        rows = db.execute(
+            text(
+                """
+                SELECT feat.value->>'layer'         AS layer,
+                       feat.value->'properties'     AS props
+                  FROM nspd_quarter_dumps d,
+                       jsonb_array_elements(d.features_json) AS feat(value)
+                 WHERE d.quarter_cad = :q
+                   AND feat.value->>'layer' LIKE 'zouit_%'
+                   AND (feat.value->'geometry') IS NOT NULL
+                   AND feat.value->>'geometry' != 'null'
+                   AND ST_Intersects(
+                       ST_Transform(
+                           ST_SetSRID(
+                               ST_GeomFromGeoJSON(feat.value->>'geometry'),
+                               3857
+                           ),
+                           4326
+                       ),
+                       ST_GeomFromText(:wkt, 4326)
+                   )
+                """
+            ),
+            {"q": quarter, "wkt": parcel_wkt},
+        ).fetchall()
+
+        result: list[dict[str, Any]] = []
+        for r in rows:
+            layer: str = r[0] or ""
+            props: dict[str, Any] = r[1] if isinstance(r[1], dict) else {}
+            group_key = layer.removeprefix("zouit_")
+            result.append(
+                {
+                    "group_key": group_key,
+                    "layer": layer,
+                    "subcategory": props.get("subcategory") or props.get("type_zone"),
+                    "name": props.get("name") or props.get("object_name"),
+                    "raw_props": props,
+                }
+            )
+        return result
+    except Exception as e:
+        logger.warning("nspd zouit query failed for quarter=%s: %s", quarter, e)
+        return []
+
+
+def _get_engineering_nearby(
+    db: Session,
+    quarter: str,
+    parcel_wkt: str,
+    layer_counts: dict[str, int] | None = None,
+) -> list[dict[str, Any]]:
+    """I3: Инженерные сооружения в радиусе _ENGINEERING_RADIUS_M от centroid участка.
+
+    layer_counts — денормализованные счётчики. Если engineering_count == 0 —
+    пропускаем heavy jsonb_array_elements + ST_DWithin scan.
+    """
+    if layer_counts is not None and layer_counts.get("engineering_count", 1) == 0:
+        return []
+    try:
+        rows = db.execute(
+            text(
+                """
+                SELECT feat.value->'properties'  AS props,
+                       ST_Distance(
+                           ST_Transform(
+                               ST_SetSRID(
+                                   ST_GeomFromGeoJSON(feat.value->>'geometry'),
+                                   3857
+                               ),
+                               4326
+                           )::geography,
+                           ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography
+                       ) AS distance_m
+                  FROM nspd_quarter_dumps d,
+                       jsonb_array_elements(d.features_json) AS feat(value)
+                 WHERE d.quarter_cad = :q
+                   AND feat.value->>'layer' = 'engineering_structures'
+                   AND (feat.value->'geometry') IS NOT NULL
+                   AND feat.value->>'geometry' != 'null'
+                   AND ST_DWithin(
+                       ST_Transform(
+                           ST_SetSRID(
+                               ST_GeomFromGeoJSON(feat.value->>'geometry'),
+                               3857
+                           ),
+                           4326
+                       )::geography,
+                       ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography,
+                       :radius_m
+                   )
+                 ORDER BY distance_m ASC
+                 LIMIT 20
+                """
+            ),
+            {"q": quarter, "wkt": parcel_wkt, "radius_m": _ENGINEERING_RADIUS_M},
+        ).fetchall()
+
+        result: list[dict[str, Any]] = []
+        for r in rows:
+            props: dict[str, Any] = r[0] if isinstance(r[0], dict) else {}
+            distance_m = float(r[1]) if r[1] is not None else None
+            name = props.get("name") or props.get("object_name")
+            obj_type = props.get("object_type") or props.get("type_zone")
+            result.append(
+                {
+                    "name": name,
+                    "type": obj_type,
+                    "distance_m": round(distance_m) if distance_m is not None else None,
+                    "raw_props": props,
+                }
+            )
+        return result
+    except Exception as e:
+        logger.warning("nspd engineering query failed for quarter=%s: %s", quarter, e)
+        return []
+
+
+# ── Harvest trigger ───────────────────────────────────────────────────────────
+
+
+def _trigger_harvest(quarter: str) -> bool:
+    """Fire-and-forget harvest_quarter.apply_async(). Возвращает True если enqueue OK.
+
+    Ленивый импорт чтобы избежать circular import (tasks → services → tasks).
+
+    Known limitation: burst из N concurrent analyze_parcel запросов на один
+    и тот же ещё не закешированный квартал может поставить N одинаковых задач
+    в очередь (нет дедупликации). UPSERT в harvest_quarter идемпотентен, поэтому
+    данные не портятся, но WAF traffic тратится впустую.
+    TODO: добавить Redis SETNX lock с TTL перед apply_async — отдельная issue/PR.
+    """
+    try:
+        from app.workers.tasks.nspd_sync import harvest_quarter
+
+        harvest_quarter.apply_async(args=[quarter], kwargs={"region_code": 66})
+        logger.info("quarter dump harvest triggered for quarter=%s", quarter)
+        return True
+    except Exception as e:
+        logger.warning("failed to trigger harvest for quarter=%s: %s", quarter, e)
+        return False

backend/tests/test_quarter_dump_lookup.py

@@ -0,0 +1,389 @@
+"""Unit tests для quarter_dump_lookup helper (Sprint 1.1 item #4).
+
+Mock-based — без реальной БД и Celery. Проверяет:
+- derive_quarter_cad: 3/4/5-сегмент → правильный quarter
+- get_quarter_dump_data: отсутствующий дамп → triggers harvest, available=False
+- get_quarter_dump_data: устаревший дамп → stale=True, triggers harvest
+- get_quarter_dump_data: свежий дамп с territorial_zones → nspd_zoning populated
+- get_quarter_dump_data: свежий дамп с 2 zouit → nspd_zouit_overlaps length=2
+- get_quarter_dump_data: parcel_wkt=None → metadata only, no spatial queries
+- make_empty_result: mutation-safety — каждый вызов возвращает независимый dict
+- early-exit: layer_counts == 0 → spatial query не вызывается
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+from app.services.site_finder.quarter_dump_lookup import (
+    EMPTY_DUMP_RESULT,
+    derive_quarter_cad,
+    get_quarter_dump_data,
+    make_empty_result,
+)
+
+# ── derive_quarter_cad ────────────────────────────────────────────────────────
+
+
+def test_derive_quarter_cad_3seg() -> None:
+    """3-сегментный номер — квартал сам себе."""
+    assert derive_quarter_cad("66:41:0204016") == "66:41:0204016"
+
+
+def test_derive_quarter_cad_4seg() -> None:
+    """4-сегментный участок → 3-сегментный квартал."""
+    assert derive_quarter_cad("66:41:0204016:10") == "66:41:0204016"
+
+
+def test_derive_quarter_cad_5seg() -> None:
+    """5-сегментное здание → 3-сегментный квартал."""
+    assert derive_quarter_cad("66:41:0204016:10:1") == "66:41:0204016"
+
+
+def test_derive_quarter_cad_invalid() -> None:
+    """Невалидный формат → None."""
+    assert derive_quarter_cad("66:41") is None
+    assert derive_quarter_cad("abc:def:ghi") is None
+    assert derive_quarter_cad("") is None
+
+
+def test_derive_quarter_cad_strips_whitespace() -> None:
+    """Пробелы обрезаются."""
+    assert derive_quarter_cad(" 66:41:0204016 ") == "66:41:0204016"
+
+
+# ── get_quarter_dump_data — no row ────────────────────────────────────────────
+
+
+def _make_db_mock(row: Any = None) -> MagicMock:
+    """Создаёт mock Session.execute().first() с заданным результатом."""
+    db = MagicMock()
+    execute_result = MagicMock()
+    execute_result.first.return_value = row
+    db.execute.return_value = execute_result
+    return db
+
+
+def test_get_quarter_dump_no_row() -> None:
+    """Дамп отсутствует в БД → harvest triggered, available=False."""
+    db = _make_db_mock(row=None)
+
+    with patch(
+        "app.services.site_finder.quarter_dump_lookup._trigger_harvest",
+        return_value=True,
+    ) as mock_trigger:
+        result = get_quarter_dump_data(db, "66:41:0204016:10", parcel_wkt="POLYGON ((...))")
+
+    mock_trigger.assert_called_once_with("66:41:0204016")
+    assert result["nspd_dump"]["available"] is False
+    assert result["nspd_dump"]["harvest_triggered"] is True
+    assert result["nspd_dump"]["stale"] is False
+    assert result["nspd_zoning"] is None
+    assert result["nspd_zouit_overlaps"] == []
+    assert result["nspd_engineering_nearby"] == []
+
+
+# ── get_quarter_dump_data — stale ─────────────────────────────────────────────
+
+
+def test_get_quarter_dump_stale() -> None:
+    """Дамп старше 180 дней → stale=True, harvest triggered, available=False."""
+    old_date = datetime.now(UTC) - timedelta(days=200)
+    # row: (quarter_cad, fetched_at_utc, total_features, harvest_error,
+    #        territorial_zones_count, zouit_count, engineering_count)
+    db = _make_db_mock(row=("66:41:0204016", old_date, 50, None, 5, 2, 1))
+
+    with patch(
+        "app.services.site_finder.quarter_dump_lookup._trigger_harvest",
+        return_value=True,
+    ) as mock_trigger:
+        result = get_quarter_dump_data(db, "66:41:0204016", parcel_wkt="POLYGON ((...))")
+
+    mock_trigger.assert_called_once_with("66:41:0204016")
+    assert result["nspd_dump"]["available"] is False
+    assert result["nspd_dump"]["stale"] is True
+    assert result["nspd_dump"]["harvest_triggered"] is True
+    assert result["nspd_dump"]["total_features"] == 50
+
+
+def test_get_quarter_dump_with_harvest_error() -> None:
+    """Дамп с harvest_error → harvest_triggered=True, available=False."""
+    fresh_date = datetime.now(UTC) - timedelta(days=10)
+    db = _make_db_mock(row=("66:41:0204016", fresh_date, 0, "WAF 429", 0, 0, 0))
+
+    with patch(
+        "app.services.site_finder.quarter_dump_lookup._trigger_harvest",
+        return_value=True,
+    ) as mock_trigger:
+        result = get_quarter_dump_data(db, "66:41:0204016", parcel_wkt=None)
+
+    mock_trigger.assert_called_once_with("66:41:0204016")
+    assert result["nspd_dump"]["available"] is False
+    assert result["nspd_dump"]["harvest_triggered"] is True
+
+
+# ── get_quarter_dump_data — no parcel_wkt ─────────────────────────────────────
+
+
+def test_get_quarter_dump_no_parcel_geometry() -> None:
+    """parcel_wkt=None → metadata only, нет spatial queries."""
+    fresh_date = datetime.now(UTC) - timedelta(days=5)
+    db = _make_db_mock(row=("66:41:0204016", fresh_date, 120, None, 10, 3, 2))
+
+    result = get_quarter_dump_data(db, "66:41:0204016", parcel_wkt=None)
+
+    # execute вызывался только один раз (для чтения строки дампа)
+    assert db.execute.call_count == 1
+    assert result["nspd_dump"]["available"] is True
+    assert result["nspd_dump"]["stale"] is False
+    assert result["nspd_zoning"] is None
+    assert result["nspd_zouit_overlaps"] == []
+    assert result["nspd_engineering_nearby"] == []
+
+
+# ── get_quarter_dump_data — fresh dump with zoning ───────────────────────────
+
+
+def _make_db_mock_with_spatial(
+    dump_row: Any,
+    zoning_row: Any = None,
+    zouit_rows: list[Any] | None = None,
+    engineering_rows: list[Any] | None = None,
+) -> MagicMock:
+    """Mock DB с разными результатами для разных запросов.
+
+    Порядок вызовов db.execute жёстко привязан к реализации get_quarter_dump_data:
+    0. dump lookup  → .first()
+    1. zoning       → .first()   (пропускается если territorial_zones_count == 0)
+    2. zouit        → .fetchall() (пропускается если zouit_count == 0)
+    3. engineering  → .fetchall() (пропускается если engineering_count == 0)
+
+    Хрупко при реорганизации порядка запросов в production-коде. При изменении
+    порядка в get_quarter_dump_data — обновить индексы call_count здесь.
+    dump_row должен быть 7-элементным tuple:
+    (quarter_cad, fetched_at_utc, total_features, harvest_error,
+     territorial_zones_count, zouit_count, engineering_count).
+    """
+    db = MagicMock()
+    call_count = 0
+
+    def side_effect(*args: Any, **kwargs: Any) -> MagicMock:
+        nonlocal call_count
+        result_mock = MagicMock()
+
+        if call_count == 0:
+            # dump lookup
+            result_mock.first.return_value = dump_row
+        elif call_count == 1:
+            # zoning
+            result_mock.first.return_value = zoning_row
+        elif call_count == 2:
+            # zouit
+            result_mock.fetchall.return_value = zouit_rows or []
+        elif call_count == 3:
+            # engineering
+            result_mock.fetchall.return_value = engineering_rows or []
+
+        call_count += 1
+        return result_mock
+
+    db.execute.side_effect = side_effect
+    return db
+
+
+def test_get_quarter_dump_fresh_zoning() -> None:
+    """Свежий дамп с territorial_zone → nspd_zoning populated."""
+    fresh_date = datetime.now(UTC) - timedelta(days=30)
+    dump_row = ("66:41:0204016", fresh_date, 100, None, 5, 0, 0)
+    # zone_props JSONB в виде dict (SQLAlchemy вернёт dict)
+    zone_props: dict[str, Any] = {
+        "reg_numb_border": "Ж-3",
+        "type_zone": "Зона жилой застройки средней этажности",
+    }
+    zoning_row_mock = MagicMock()
+    zoning_row_mock.__getitem__ = lambda self, i: zone_props if i == 0 else None
+
+    db = _make_db_mock_with_spatial(
+        dump_row=dump_row,
+        zoning_row=zoning_row_mock,
+        zouit_rows=[],
+        engineering_rows=[],
+    )
+
+    wkt = "POLYGON ((60.6 56.8, 60.7 56.8, 60.7 56.9, 60.6 56.9, 60.6 56.8))"
+    result = get_quarter_dump_data(db, "66:41:0204016", parcel_wkt=wkt)
+
+    assert result["nspd_dump"]["available"] is True
+    assert result["nspd_zoning"] is not None
+    assert result["nspd_zoning"]["zone_code"] == "Ж-3"
+    assert result["nspd_zoning"]["zone_name"] == "Зона жилой застройки средней этажности"
+    assert result["nspd_zoning"]["source"] == "nspd-quarter-dump"
+
+
+def test_get_quarter_dump_fresh_zoning_no_match() -> None:
+    """Свежий дамп, centroid вне всех territorial_zones → nspd_zoning=None."""
+    fresh_date = datetime.now(UTC) - timedelta(days=30)
+    dump_row = ("66:41:0204016", fresh_date, 100, None, 3, 0, 0)
+
+    db = _make_db_mock_with_spatial(
+        dump_row=dump_row,
+        zoning_row=None,
+        zouit_rows=[],
+        engineering_rows=[],
+    )
+
+    result = get_quarter_dump_data(db, "66:41:0204016", parcel_wkt="POLYGON ((...))")
+    assert result["nspd_zoning"] is None
+    assert result["nspd_dump"]["available"] is True
+
+
+# ── get_quarter_dump_data — fresh dump with zouit overlaps ────────────────────
+
+
+def test_get_quarter_dump_fresh_zouit_overlaps() -> None:
+    """Свежий дамп с 2 zouit → nspd_zouit_overlaps length=2."""
+    fresh_date = datetime.now(UTC) - timedelta(days=10)
+    # territorial_zones_count > 0 чтобы все 3 spatial запроса вызывались
+    # (иначе call_count в mock смещается и zouit попадает не в тот слот)
+    dump_row = ("66:41:0204016", fresh_date, 200, None, 1, 5, 0)
+
+    props1: dict[str, Any] = {"name": "ОКН Дом купца", "subcategory": "okn"}
+    props2: dict[str, Any] = {"name": "ОЗ газопровода", "subcategory": "engineering"}
+
+    def make_row(layer: str, props: dict[str, Any]) -> MagicMock:
+        r = MagicMock()
+        r.__getitem__ = lambda self, i: layer if i == 0 else props
+        return r
+
+    zouit_rows = [make_row("zouit_okn", props1), make_row("zouit_engineering", props2)]
+
+    db = _make_db_mock_with_spatial(
+        dump_row=dump_row,
+        zoning_row=None,
+        zouit_rows=zouit_rows,
+        engineering_rows=[],
+    )
+
+    result = get_quarter_dump_data(db, "66:41:0204016", parcel_wkt="POLYGON ((...))")
+
+    assert len(result["nspd_zouit_overlaps"]) == 2
+    keys = {r["group_key"] for r in result["nspd_zouit_overlaps"]}
+    assert "okn" in keys
+    assert "engineering" in keys
+
+
+# ── get_quarter_dump_data — integration of nspd_dump in response ──────────────
+
+
+def test_get_quarter_dump_includes_nspd_fields() -> None:
+    """Базовая проверка что все 4 nspd-ключа присутствуют в возвращаемом dict."""
+    fresh_date = datetime.now(UTC) - timedelta(days=1)
+    dump_row = ("66:41:0204016", fresh_date, 55, None, 2, 1, 0)
+
+    db = _make_db_mock_with_spatial(
+        dump_row=dump_row,
+        zoning_row=None,
+        zouit_rows=[],
+        engineering_rows=[],
+    )
+
+    result = get_quarter_dump_data(db, "66:41:0204016:5", parcel_wkt="POINT (60.6 56.8)")
+
+    assert "nspd_zoning" in result
+    assert "nspd_zouit_overlaps" in result
+    assert "nspd_engineering_nearby" in result
+    assert "nspd_dump" in result
+    dump = result["nspd_dump"]
+    assert "available" in dump
+    assert "fetched_at_utc" in dump
+    assert "stale" in dump
+    assert "harvest_triggered" in dump
+    assert "total_features" in dump
+
+
+# ── make_empty_result — mutation safety ─────────────────────────────────────
+
+
+def testmake_empty_result_returns_independent_copies() -> None:
+    """Каждый вызов make_empty_result() возвращает независимый dict.
+
+    Мутация nspd_dump в одном результате не должна влиять на другой
+    и на модульный sentinel EMPTY_DUMP_RESULT.
+    """
+    a = make_empty_result()
+    b = make_empty_result()
+
+    # Вложенный dict — независимый объект
+    assert a["nspd_dump"] is not b["nspd_dump"]
+    assert a["nspd_dump"] is not EMPTY_DUMP_RESULT["nspd_dump"]
+
+    # Мутируем один — другой не трогается
+    a["nspd_dump"]["harvest_triggered"] = True
+    a["nspd_zouit_overlaps"].append({"group_key": "test"})
+
+    assert b["nspd_dump"]["harvest_triggered"] is False
+    assert b["nspd_zouit_overlaps"] == []
+    assert EMPTY_DUMP_RESULT["nspd_dump"]["harvest_triggered"] is False
+    assert EMPTY_DUMP_RESULT["nspd_zouit_overlaps"] == []
+
+
+def testmake_empty_result_overrides() -> None:
+    """make_empty_result принимает override-аргументы."""
+    r = make_empty_result(harvest_triggered=True, stale=True, total_features=42)
+    assert r["nspd_dump"]["harvest_triggered"] is True
+    assert r["nspd_dump"]["stale"] is True
+    assert r["nspd_dump"]["total_features"] == 42
+    assert r["nspd_dump"]["available"] is False
+
+
+# ── early-exit when layer_counts == 0 ────────────────────────────────────────
+
+
+def test_early_exit_all_counts_zero_no_spatial_queries() -> None:
+    """Если все denormalized counts == 0 — spatial queries не вызываются."""
+    fresh_date = datetime.now(UTC) - timedelta(days=5)
+    # territorial_zones_count=0, zouit_count=0, engineering_count=0
+    dump_row = ("66:41:0204016", fresh_date, 50, None, 0, 0, 0)
+
+    db = _make_db_mock_with_spatial(dump_row=dump_row)
+
+    result = get_quarter_dump_data(
+        db, "66:41:0204016", parcel_wkt="POLYGON ((60.6 56.8, 60.7 56.8, 60.7 56.9, 60.6 56.8))"
+    )
+
+    # Только 1 execute-вызов — чтение dump row; spatial helpers пропущены
+    assert db.execute.call_count == 1
+    assert result["nspd_zoning"] is None
+    assert result["nspd_zouit_overlaps"] == []
+    assert result["nspd_engineering_nearby"] == []
+    assert result["nspd_dump"]["available"] is True
+
+
+def test_early_exit_partial_counts() -> None:
+    """territorial_zones_count > 0, zouit_count == 0 → zoning query вызывается,
+    zouit и engineering — нет."""
+    fresh_date = datetime.now(UTC) - timedelta(days=5)
+    dump_row = ("66:41:0204016", fresh_date, 30, None, 3, 0, 0)
+    zone_props: dict[str, Any] = {"reg_numb_border": "Ж-1", "type_zone": "Зона жилая"}
+    zoning_row_mock = MagicMock()
+    zoning_row_mock.__getitem__ = lambda self, i: zone_props if i == 0 else None
+
+    db = _make_db_mock_with_spatial(
+        dump_row=dump_row,
+        zoning_row=zoning_row_mock,
+        zouit_rows=[],
+        engineering_rows=[],
+    )
+
+    result = get_quarter_dump_data(
+        db, "66:41:0204016", parcel_wkt="POLYGON ((60.6 56.8, 60.7 56.8, 60.7 56.9, 60.6 56.8))"
+    )
+
+    # dump lookup (0) + zoning (1) = 2 вызова; zouit и engineering — пропущены
+    assert db.execute.call_count == 2
+    assert result["nspd_zoning"] is not None
+    assert result["nspd_zoning"]["zone_code"] == "Ж-1"
+    assert result["nspd_zouit_overlaps"] == []
+    assert result["nspd_engineering_nearby"] == []