feat(nspd): denorm nspd_parcels + nspd_buildings (#94 PR3 of 4) #222
6 changed files with 932 additions and 0 deletions
|
|
@ -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}
|
||||
|
|
|
|||
290
backend/app/services/scrapers/nspd_denorm.py
Normal file
290
backend/app/services/scrapers/nspd_denorm.py
Normal file
|
|
@ -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}
|
||||
96
backend/app/workers/tasks/nspd_denorm_backfill.py
Normal file
96
backend/app/workers/tasks/nspd_denorm_backfill.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
322
backend/tests/services/test_nspd_denorm.py
Normal file
322
backend/tests/services/test_nspd_denorm.py
Normal file
|
|
@ -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
|
||||
171
data/sql/99_nspd_entities_denorm.sql
Normal file
171
data/sql/99_nspd_entities_denorm.sql
Normal file
|
|
@ -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;
|
||||
Loading…
Add table
Reference in a new issue