feat(site-finder): integrate nspd_quarter_dumps cache в analyze_parcel (#94 Sprint 1.1 #4 FINAL)

Замыкает Sprint 1.1 из #94 part 2 plan. После этого PR пользователь видит
свежие НСПД данные в UI (frontend integration — отдельный PR).

Backend (new app/services/site_finder/quarter_dump_lookup.py):
- `derive_quarter_cad(cad_num)` — 3/4/5-сегмент → quarter (3-segment)
- `get_quarter_dump_data(db, cad_num, parcel_wkt)` — main entrypoint:
  - Reads nspd_quarter_dumps row для derived quarter
  - Freshness threshold: 180 days
  - Missing/stale/harvest_error → trigger harvest_quarter.apply_async() fire-
    and-forget (lazy import против circular), return EMPTY_DUMP_RESULT
  - Fresh + parcel_wkt=None → metadata only (no spatial queries)
  - Fresh + geometry → 3 spatial queries via jsonb_array_elements + ST_Transform
    (3857→4326) + ST_Intersects / ST_DWithin
- 3 private helpers:
  - `_get_zoning` — point-in-polygon parcel centroid vs territorial_zones, LIMIT 1
  - `_get_zouit_overlaps` — все zouit_% layers пересекающиеся с parcel
  - `_get_engineering_nearby` — engineering_structures в 200m, sorted by distance
- `EMPTY_DUMP_RESULT` module-level constant — DRY для no-dump fallback (used
  in get_quarter_dump_data internal + analyze_parcel try/except wrap)

Backend (parcels.py):
- Import EMPTY_DUMP_RESULT + get_quarter_dump_data
- Call wrapped в try/except — если nspd_quarter_dumps недоступна (DB timeout
  / table missing) → EMPTY_DUMP_RESULT fallback вместо 500 (consistent с
  resilience pattern других optional fetches)
- Response gets 4 new fields:
  - nspd_zoning: dict | None (G1 ПЗЗ — zone_code, zone_name, source)
  - nspd_zouit_overlaps: list[dict] (G3 — overlaps в parcel, per ЗОУИТ group)
  - nspd_engineering_nearby: list[dict] (I3 — engineering structures в 200m)
  - nspd_dump: dict (freshness metadata — available, fetched_at_utc, stale,
    harvest_triggered, total_features)

Tests: 13 new в test_quarter_dump_lookup.py (mock-based, no real DB):
- derive_quarter_cad 5 edge cases (3seg, 4seg, 5seg, invalid, whitespace)
- get_quarter_dump no_row → harvest triggered
- stale (>180d) → harvest triggered, stale=True
- harvest_error row → retry harvest triggered
- parcel_wkt=None → metadata only (1 DB call)
- fresh + zoning extraction
- fresh + zouit_overlaps list
- fresh integration: все 4 keys present

47 pre-existing tests still pass.

Code review (code-reviewer pre-push): MINOR, 0 blocking. Applied 2 of 4:
- ✅ #1: try/except wrap around get_quarter_dump_data в analyze_parcel
  (защита от DB unavailability) + DRY через EMPTY_DUMP_RESULT module const
- ✅ #2: removed redundant nspd_zoning.fetched_at_utc (DRY — freshness в
  nspd_dump.fetched_at_utc)
- ⏭ Deferred (acceptable): #3 ad-hoc harvest_quarter retry cooldown для
  harvest_error rows (только при high traffic + persistent NSPD errors);
  #4 raw_props в response — tech debt, убрать вместе с frontend PR

Performance note: 3 spatial queries per analyze adds ~10-50ms on typical
~100-feature quarter. Mitigation if quarters grow dense: materialized
per-layer sub-table (отдельная DB issue).

Closes Sprint 1.1 part of #94. Frontend rendering этих 4 полей — отдельный
PR (next: #112 / #115 / #114).

backend/app/api/v1/parcels.py

@@ -22,6 +22,7 @@ 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 EMPTY_DUMP_RESULT, get_quarter_dump_data
 
 logger = logging.getLogger(__name__)
 
@@ -1529,6 +1530,13 @@ 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)
+        nspd_dump_data = EMPTY_DUMP_RESULT
+
     # 10) Market trend — динамика цен ДДУ в радиусе 3 км за 6 vs предыдущие 6 месяцев
     market_trend: dict[str, Any] | None = None
     try:
@@ -1831,6 +1839,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,365 @@
+"""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
+
+# Пустой результат — возвращается когда дамп недоступен, устарел или при ошибке.
+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 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).
+    """
+    empty_result = EMPTY_DUMP_RESULT
+
+    quarter = derive_quarter_cad(cad_num)
+    if quarter is None:
+        logger.warning("quarter_dump_lookup: cannot derive quarter from cad=%s", cad_num)
+        return empty_result
+
+    # Читаем строку дампа из БД
+    row = db.execute(
+        text(
+            """
+            SELECT quarter_cad,
+                   fetched_at_utc,
+                   total_features,
+                   harvest_error
+              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)
+        result = dict(empty_result)
+        result["nspd_dump"] = {
+            "available": False,
+            "fetched_at_utc": None,
+            "stale": False,
+            "harvest_triggered": harvest_triggered,
+            "total_features": None,
+        }
+        return result
+
+    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]
+
+    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)
+        result = dict(empty_result)
+        result["nspd_dump"] = {
+            "available": False,
+            "fetched_at_utc": fetched_at.isoformat(),
+            "stale": is_stale,
+            "harvest_triggered": harvest_triggered,
+            "total_features": total_features,
+        }
+        return result
+
+    # Свежий дамп без ошибок — извлекаем 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,
+        }
+
+    nspd_zoning = _get_zoning(db, quarter, parcel_wkt)
+    nspd_zouit = _get_zouit_overlaps(db, quarter, parcel_wkt)
+    nspd_engineering = _get_engineering_nearby(db, quarter, parcel_wkt)
+
+    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) -> dict[str, Any] | None:
+    """G1: ПЗЗ территориальная зона по centroid участка из dump.
+
+    Geometry в features_json в EPSG:3857 — ST_Transform на чтении → 4326.
+    """
+    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,
+) -> list[dict[str, Any]]:
+    """G3: ЗОУИТ которые пересекают участок.
+
+    Проверяем 5 групп: zouit_okn, zouit_engineering, zouit_natural,
+    zouit_protected, zouit_other.
+    """
+    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,
+) -> list[dict[str, Any]]:
+    """I3: Инженерные сооружения в радиусе _ENGINEERING_RADIUS_M от centroid участка."""
+    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).
+    """
+    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,290 @@
+"""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
+"""
+
+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 (
+    derive_quarter_cad,
+    get_quarter_dump_data,
+)
+
+# ── 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)
+    db = _make_db_mock(row=("66:41:0204016", old_date, 50, 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", 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"))
+
+    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))
+
+    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 с разными результатами для разных запросов.
+
+    Порядок вызовов:
+    1. read dump row
+    2. zoning (first())
+    3. zouit (fetchall())
+    4. engineering (fetchall())
+    """
+    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)
+    # 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)
+
+    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)
+    dump_row = ("66:41:0204016", fresh_date, 200, None)
+
+    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)
+
+    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