lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

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

Browse source 
Замыкает 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).
This commit is contained in:
lekss361 2026-05-12 19:00:03 +03:00
parent f3b0ce5687
commit 3068a9c531
3 changed files with 672 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
backend/app/api/v1/parcels.py
View file

@ -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"],
}

365
backend/app/services/site_finder/quarter_dump_lookup.py Normal file
View file

@ -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

290
backend/tests/test_quarter_dump_lookup.py Normal file
View file

@ -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