diff --git a/backend/app/api/v1/admin_etl.py b/backend/app/api/v1/admin_etl.py index 96b2d51a..047cb68e 100644 --- a/backend/app/api/v1/admin_etl.py +++ b/backend/app/api/v1/admin_etl.py @@ -4,6 +4,9 @@ POST /api/v1/admin/etl/objective-backfill Запустить fuzzy-match backfill objective_complex_mapping. Опциональный REFRESH mv_layout_velocity после успешного backfill. +POST /api/v1/admin/etl/nspd-denorm-backfill + Запустить backfill nspd_parcels/nspd_buildings из всех nspd_quarter_dumps. + Header: X-Admin-Token: """ @@ -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} diff --git a/backend/app/services/scrapers/nspd_denorm.py b/backend/app/services/scrapers/nspd_denorm.py new file mode 100644 index 00000000..de3ee973 --- /dev/null +++ b/backend/app/services/scrapers/nspd_denorm.py @@ -0,0 +1,290 @@ +"""Denormalize nspd_quarter_dumps.features_json → nspd_parcels / nspd_buildings. + +Используется: + - Inline в harvest_quarter task после UPSERT dump'а + - Backfill task для existing dumps (backfill_all_dumps) + +Property mapping (NSPD WMS GetFeatureInfo responses): + Parcels (layer="parcels", layer id 36048): + cad_num — кадастровый номер ЗУ + permitted_use — ВРИ + land_category — категория земель + cost_value — кадастровая стоимость, руб. + area — площадь, м² + address — адрес + + Buildings (layer="buildings", layer id 36049): + cad_num — кадастровый номер ОКС + purpose — назначение ("Многоквартирный дом" / "Нежилое здание" / ...) + floors_above_ground — надземных этажей + floors_underground — подземных этажей + year_built — год постройки + cost_value — кадастровая стоимость, руб. + build_record_area — площадь по ГКН, м² + address — адрес + +Геометрия в features_json хранится в EPSG:3857 (как возвращает NSPD WMS) — +трансформация 3857→4326 делается в SQL через ST_Transform. +""" + +from __future__ import annotations + +import datetime +import json +import logging +from typing import Any + +from sqlalchemy import text +from sqlalchemy.orm import Session + +logger = logging.getLogger(__name__) + +# ── Type coercions ───────────────────────────────────────────────────────────── + + +def _coerce_int(v: Any) -> int | None: + """NSPD properties могут быть str / int / None.""" + if v is None: + return None + try: + return int(v) + except (ValueError, TypeError): + return None + + +def _coerce_numeric(v: Any) -> float | None: + """NSPD properties могут быть str с запятой (европейский формат) или None.""" + if v is None: + return None + try: + return float(str(v).replace(",", ".").strip()) + except (ValueError, TypeError): + return None + + +# ── Parcel UPSERT ────────────────────────────────────────────────────────────── + +_PARCEL_UPSERT_SQL = text( + """ + INSERT INTO nspd_parcels ( + cad_num, quarter_cad, permitted_use, land_category, + cost_value, cost_per_m2, area_sqm, address, geom, snapshot_date + ) VALUES ( + CAST(:cad_num AS text), + CAST(:quarter_cad AS text), + CAST(:permitted_use AS text), + CAST(:land_category AS text), + CAST(:cost_value AS numeric), + CAST(:cost_per_m2 AS numeric), + CAST(:area_sqm AS numeric), + CAST(:address AS text), + CASE WHEN :geom_json IS NULL THEN NULL + ELSE ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:geom_json), 3857), 4326) + END, + CAST(:snapshot_date AS date) + ) + ON CONFLICT (cad_num) DO UPDATE SET + quarter_cad = EXCLUDED.quarter_cad, + permitted_use = EXCLUDED.permitted_use, + land_category = EXCLUDED.land_category, + cost_value = EXCLUDED.cost_value, + cost_per_m2 = EXCLUDED.cost_per_m2, + area_sqm = EXCLUDED.area_sqm, + address = EXCLUDED.address, + geom = EXCLUDED.geom, + snapshot_date = EXCLUDED.snapshot_date, + updated_at = NOW() + """ +) + + +def denorm_parcel_feature( + db: Session, + *, + feature: dict[str, Any], + quarter_cad: str, + snapshot_date: str, +) -> bool: + """UPSERT one parcel feature → nspd_parcels. + + Returns True если строка вставлена/обновлена, False при пропуске (нет cad_num). + """ + props = feature.get("properties") or {} + cad_num = props.get("cad_num") or props.get("cadastral_number") + if not cad_num: + return False + + area = _coerce_numeric(props.get("area")) + cost_value = _coerce_numeric(props.get("cost_value")) + cost_per_m2: float | None = None + if cost_value is not None and area is not None and area > 0: + cost_per_m2 = cost_value / area + + geom = feature.get("geometry") + geom_json: str | None = json.dumps(geom, ensure_ascii=False) if geom else None + + try: + with db.begin_nested(): + db.execute( + _PARCEL_UPSERT_SQL, + { + "cad_num": cad_num, + "quarter_cad": quarter_cad, + "permitted_use": props.get("permitted_use"), + "land_category": props.get("land_category"), + "cost_value": cost_value, + "cost_per_m2": cost_per_m2, + "area_sqm": area, + "address": props.get("address"), + "geom_json": geom_json, + "snapshot_date": snapshot_date, + }, + ) + return True + except Exception as e: + logger.warning("denorm parcel %s failed: %s", cad_num, e) + return False + + +# ── Building UPSERT ──────────────────────────────────────────────────────────── + +_BUILDING_UPSERT_SQL = text( + """ + INSERT INTO nspd_buildings ( + cad_num, quarter_cad, purpose, floors, floors_underground, + year_built, cost_value, build_record_area, address, geom, snapshot_date + ) VALUES ( + CAST(:cad_num AS text), + CAST(:quarter_cad AS text), + CAST(:purpose AS text), + CAST(:floors AS integer), + CAST(:floors_underground AS integer), + CAST(:year_built AS integer), + CAST(:cost_value AS numeric), + CAST(:build_record_area AS numeric), + CAST(:address AS text), + CASE WHEN :geom_json IS NULL THEN NULL + ELSE ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:geom_json), 3857), 4326) + END, + CAST(:snapshot_date AS date) + ) + ON CONFLICT (cad_num) DO UPDATE SET + quarter_cad = EXCLUDED.quarter_cad, + purpose = EXCLUDED.purpose, + floors = EXCLUDED.floors, + floors_underground = EXCLUDED.floors_underground, + year_built = EXCLUDED.year_built, + cost_value = EXCLUDED.cost_value, + build_record_area = EXCLUDED.build_record_area, + address = EXCLUDED.address, + geom = EXCLUDED.geom, + snapshot_date = EXCLUDED.snapshot_date, + updated_at = NOW() + """ +) + + +def denorm_building_feature( + db: Session, + *, + feature: dict[str, Any], + quarter_cad: str, + snapshot_date: str, +) -> bool: + """UPSERT one building feature → nspd_buildings. + + Returns True если строка вставлена/обновлена, False при пропуске (нет cad_num). + """ + props = feature.get("properties") or {} + cad_num = props.get("cad_num") or props.get("cadastral_number") + if not cad_num: + return False + + geom = feature.get("geometry") + geom_json: str | None = json.dumps(geom, ensure_ascii=False) if geom else None + + try: + with db.begin_nested(): + db.execute( + _BUILDING_UPSERT_SQL, + { + "cad_num": cad_num, + "quarter_cad": quarter_cad, + "purpose": props.get("purpose"), + "floors": _coerce_int(props.get("floors_above_ground")), + "floors_underground": _coerce_int(props.get("floors_underground")), + "year_built": _coerce_int(props.get("year_built")), + "cost_value": _coerce_numeric(props.get("cost_value")), + "build_record_area": _coerce_numeric(props.get("build_record_area")), + "address": props.get("address"), + "geom_json": geom_json, + "snapshot_date": snapshot_date, + }, + ) + return True + except Exception as e: + logger.warning("denorm building %s failed: %s", cad_num, e) + return False + + +# ── Batch helper ─────────────────────────────────────────────────────────────── + + +def denorm_dump( + db: Session, + *, + quarter_cad: str, + features: list[dict[str, Any]], +) -> dict[str, int]: + """Denorm all features из одного quarter dump → nspd_parcels / nspd_buildings. + + Итерирует features_json плоский список. Записи с layer="parcels" идут в + nspd_parcels, layer="buildings" → nspd_buildings. Остальные layers пропускаются. + + Каждая строка UPSERT'ится в отдельном SAVEPOINT (begin_nested) — ошибка + одной строки не откатывает весь batch. + + Args: + db: SQLAlchemy Session. Caller отвечает за commit/close после вызова. + quarter_cad: 3-сегментный кадастровый квартал. + features: плоский list из features_json JSONB (уже декодированный Python list). + + Returns: + dict {"parcels": N, "buildings": M, "errors": K} — количество обработанных строк. + """ + snapshot_date = datetime.date.today().isoformat() + parcels_n = 0 + buildings_n = 0 + errors_n = 0 + + for feat in features: + layer = feat.get("layer", "") + try: + if layer == "parcels": + if denorm_parcel_feature( + db, feature=feat, quarter_cad=quarter_cad, snapshot_date=snapshot_date + ): + parcels_n += 1 + else: + errors_n += 1 + elif layer == "buildings": + if denorm_building_feature( + db, feature=feat, quarter_cad=quarter_cad, snapshot_date=snapshot_date + ): + buildings_n += 1 + else: + errors_n += 1 + # Остальные layers (territorial_zones, zouit_*, risk_*, ...) — пропускаем + except Exception as e: + logger.warning("denorm feature (layer=%s quarter=%s) failed: %s", layer, quarter_cad, e) + errors_n += 1 + + db.commit() + logger.info( + "denorm_dump quarter=%s parcels=%d buildings=%d errors=%d", + quarter_cad, + parcels_n, + buildings_n, + errors_n, + ) + return {"parcels": parcels_n, "buildings": buildings_n, "errors": errors_n} diff --git a/backend/app/workers/tasks/nspd_denorm_backfill.py b/backend/app/workers/tasks/nspd_denorm_backfill.py new file mode 100644 index 00000000..c0788328 --- /dev/null +++ b/backend/app/workers/tasks/nspd_denorm_backfill.py @@ -0,0 +1,96 @@ +"""One-shot backfill: denormalize all existing nspd_quarter_dumps → nspd_parcels/buildings. + +Запускается вручную через admin endpoint POST /api/v1/admin/etl/nspd-denorm-backfill +или через Celery CLI: + celery -A app.workers.celery_app call nspd_denorm.backfill_all_dumps + +Идемпотентен: повторный запуск обновляет строки через ON CONFLICT DO UPDATE. +""" + +from __future__ import annotations + +import logging + +from celery import shared_task +from sqlalchemy import text + +from app.core.db import SessionLocal +from app.services.scrapers.nspd_denorm import denorm_dump + +logger = logging.getLogger(__name__) + + +@shared_task( + name="nspd_denorm.backfill_all_dumps", + bind=True, + soft_time_limit=3600, # 1 час — много кварталов × ~10ms per parcel + max_retries=0, # One-shot, не retry +) +def backfill_all_dumps(self: object, *, limit: int | None = None) -> dict[str, int]: # type: ignore[misc] + """Backfill all existing nspd_quarter_dumps → nspd_parcels / nspd_buildings. + + Читает все строки nspd_quarter_dumps (или limit строк), и для каждой вызывает + denorm_dump. Возвращает агрегированные счётчики по всем кварталам. + + Args: + limit: если задан — обработать только первые N кварталов (для тестирования). + + Returns: + dict { + "parcels": суммарно вставлено/обновлено parcel строк, + "buildings": суммарно вставлено/обновлено building строк, + "errors": суммарно пропусков (нет cad_num, DB ошибки), + "quarters_processed": количество обработанных кварталов, + } + """ + totals: dict[str, int] = { + "parcels": 0, + "buildings": 0, + "errors": 0, + "quarters_processed": 0, + } + + db = SessionLocal() + try: + sql = ( + "SELECT quarter_cad, features_json " + "FROM nspd_quarter_dumps " + "WHERE total_features > 0 " + "ORDER BY quarter_cad" + ) + if limit is not None: + # Используем параметр через text() чтобы избежать SQL-injection. + rows = db.execute( + text(sql + " LIMIT CAST(:lim AS integer)"), {"lim": int(limit)} + ).fetchall() + else: + rows = db.execute(text(sql)).fetchall() + + logger.info("backfill_all_dumps: starting, quarters=%d limit=%s", len(rows), limit) + + for row in rows: + quarter_cad: str = row[0] + features_json = row[1] + # features_json — уже list[dict] через psycopg v3 JSON decoding. + features: list[dict] = features_json if isinstance(features_json, list) else [] + + try: + counts = denorm_dump(db, quarter_cad=quarter_cad, features=features) + totals["parcels"] += counts["parcels"] + totals["buildings"] += counts["buildings"] + totals["errors"] += counts["errors"] + totals["quarters_processed"] += 1 + except Exception as e: + logger.warning("backfill_all_dumps: quarter=%s failed: %s", quarter_cad, e) + totals["errors"] += 1 + # Продолжаем с остальными кварталами + try: + db.rollback() + except Exception: + pass + + logger.info("backfill_all_dumps: done totals=%s", totals) + finally: + db.close() + + return totals diff --git a/backend/app/workers/tasks/nspd_sync.py b/backend/app/workers/tasks/nspd_sync.py index f87dcefc..2743a630 100644 --- a/backend/app/workers/tasks/nspd_sync.py +++ b/backend/app/workers/tasks/nspd_sync.py @@ -33,6 +33,7 @@ from app.services.scrapers.nspd_client import ( NspdLiteWafError, QuarterDump, ) +from app.services.scrapers.nspd_denorm import denorm_dump from app.workers.celery_app import celery_app logger = logging.getLogger(__name__) @@ -301,6 +302,32 @@ def harvest_quarter( _upsert_dump(quarter_cad, region_code, dump, features_json, duration_ms, None) + # Inline denorm: разложить parcels/buildings из dump в nspd_parcels/nspd_buildings. + # Ошибка denorm не должна фейлить весь harvest — только warning. + try: + denorm_db = SessionLocal() + try: + denorm_counts = denorm_dump( + denorm_db, + quarter_cad=quarter_cad, + features=features_json or [], + ) + logger.info( + "harvest_quarter denorm: cad=%s parcels=%d buildings=%d errors=%d", + quarter_cad, + denorm_counts["parcels"], + denorm_counts["buildings"], + denorm_counts["errors"], + ) + finally: + denorm_db.close() + except Exception as denorm_exc: + logger.warning( + "harvest_quarter denorm failed (non-fatal): cad=%s error=%s", + quarter_cad, + denorm_exc, + ) + logger.info( "harvest_quarter done: cad=%s region=%d duration=%dms total=%d", quarter_cad, diff --git a/backend/tests/services/test_nspd_denorm.py b/backend/tests/services/test_nspd_denorm.py new file mode 100644 index 00000000..3420a761 --- /dev/null +++ b/backend/tests/services/test_nspd_denorm.py @@ -0,0 +1,322 @@ +"""Тесты для nspd_denorm.py — coerce helpers + denorm_parcel/building/dump. + +Не требует реального PostgreSQL — мокает db.execute() и db.begin_nested(). +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from app.services.scrapers.nspd_denorm import ( + _coerce_int, + _coerce_numeric, + denorm_building_feature, + denorm_dump, + denorm_parcel_feature, +) + +# ── _coerce_int ──────────────────────────────────────────────────────────────── + + +def test_coerce_int_str_number() -> None: + assert _coerce_int("5") == 5 + + +def test_coerce_int_int_passthrough() -> None: + assert _coerce_int(17) == 17 + + +def test_coerce_int_none_returns_none() -> None: + assert _coerce_int(None) is None + + +def test_coerce_int_invalid_returns_none() -> None: + assert _coerce_int("не_число") is None + + +def test_coerce_int_float_truncates() -> None: + assert _coerce_int(3.9) == 3 + + +# ── _coerce_numeric ──────────────────────────────────────────────────────────── + + +def test_coerce_numeric_float_passthrough() -> None: + assert _coerce_numeric(12.5) == 12.5 + + +def test_coerce_numeric_str_dot() -> None: + assert _coerce_numeric("1234567.89") == pytest.approx(1234567.89) + + +def test_coerce_numeric_comma_decimal() -> None: + """Европейский формат с запятой → корректный float.""" + assert _coerce_numeric("1234567,89") == pytest.approx(1234567.89) + + +def test_coerce_numeric_none_returns_none() -> None: + assert _coerce_numeric(None) is None + + +def test_coerce_numeric_invalid_returns_none() -> None: + assert _coerce_numeric("N/A") is None + + +# ── Helpers для моков ────────────────────────────────────────────────────────── + + +def _make_mock_session() -> MagicMock: + """Мок Session: begin_nested() — context manager, execute() → MagicMock.""" + db = MagicMock() + # begin_nested() должен быть context manager + cm = MagicMock() + cm.__enter__ = MagicMock(return_value=None) + cm.__exit__ = MagicMock(return_value=False) + db.begin_nested.return_value = cm + return db + + +def _parcel_feature( + cad_num: str = "66:41:0204016:10", + area: Any = "500.0", + cost_value: Any = "1500000", + **extra_props: Any, +) -> dict[str, Any]: + props: dict[str, Any] = { + "cad_num": cad_num, + "area": area, + "cost_value": cost_value, + "permitted_use": "Для ИЖС", + "land_category": "Земли населённых пунктов", + "address": "г. Екатеринбург", + **extra_props, + } + return { + "layer": "parcels", + "feature_id": "123", + "geometry": {"type": "Polygon", "coordinates": [[[0, 0], [1, 0], [1, 1], [0, 0]]]}, + "properties": props, + } + + +def _building_feature( + cad_num: str = "66:41:0204016:10:1", + purpose: str = "Многоквартирный дом", + **extra_props: Any, +) -> dict[str, Any]: + props: dict[str, Any] = { + "cad_num": cad_num, + "purpose": purpose, + "floors_above_ground": 9, + "floors_underground": 1, + "year_built": 1985, + "cost_value": "50000000", + "build_record_area": "3200.0", + "address": "ул. Ленина 1", + **extra_props, + } + return { + "layer": "buildings", + "feature_id": "456", + "geometry": {"type": "Polygon", "coordinates": [[[0, 0], [1, 0], [1, 1], [0, 0]]]}, + "properties": props, + } + + +# ── denorm_parcel_feature ────────────────────────────────────────────────────── + + +def test_denorm_parcel_feature_inserts() -> None: + """Полный feature с cad_num → вызывает db.execute с правильными параметрами.""" + db = _make_mock_session() + + result = denorm_parcel_feature( + db, + feature=_parcel_feature(), + quarter_cad="66:41:0204016", + snapshot_date="2026-05-16", + ) + + assert result is True + db.begin_nested.assert_called_once() + db.execute.assert_called_once() + # Проверяем что параметры содержат нужные ключи + call_kwargs = db.execute.call_args[0][1] # positional dict + assert call_kwargs["cad_num"] == "66:41:0204016:10" + assert call_kwargs["quarter_cad"] == "66:41:0204016" + assert call_kwargs["permitted_use"] == "Для ИЖС" + assert call_kwargs["land_category"] == "Земли населённых пунктов" + assert call_kwargs["area_sqm"] == 500.0 + assert call_kwargs["cost_value"] == 1500000.0 + assert call_kwargs["cost_per_m2"] == pytest.approx(3000.0) + assert call_kwargs["snapshot_date"] == "2026-05-16" + + +def test_denorm_parcel_no_cad_num_skipped() -> None: + """Feature без cad_num → возвращает False, db не вызывается.""" + db = _make_mock_session() + feature: dict[str, Any] = { + "layer": "parcels", + "feature_id": None, + "geometry": None, + "properties": {"area": "100"}, + } + result = denorm_parcel_feature( + db, feature=feature, quarter_cad="66:41:0204016", snapshot_date="2026-05-16" + ) + assert result is False + db.execute.assert_not_called() + + +def test_denorm_parcel_zero_area_cost_per_m2_none() -> None: + """area=0 → cost_per_m2 = None (нет деления на ноль).""" + db = _make_mock_session() + result = denorm_parcel_feature( + db, + feature=_parcel_feature(area="0", cost_value="1000"), + quarter_cad="66:41:0204016", + snapshot_date="2026-05-16", + ) + assert result is True + call_kwargs = db.execute.call_args[0][1] + assert call_kwargs["cost_per_m2"] is None + + +def test_denorm_parcel_null_geometry() -> None: + """feature без geometry → geom_json=None, INSERT всё равно вызывается.""" + db = _make_mock_session() + feature = _parcel_feature() + feature["geometry"] = None + + result = denorm_parcel_feature( + db, feature=feature, quarter_cad="66:41:0204016", snapshot_date="2026-05-16" + ) + assert result is True + call_kwargs = db.execute.call_args[0][1] + assert call_kwargs["geom_json"] is None + + +def test_denorm_parcel_db_exception_returns_false() -> None: + """DB execute raises → возвращает False (не propagate).""" + db = _make_mock_session() + db.execute.side_effect = Exception("DB constraint violation") + + result = denorm_parcel_feature( + db, + feature=_parcel_feature(), + quarter_cad="66:41:0204016", + snapshot_date="2026-05-16", + ) + assert result is False + + +# ── denorm_building_feature ──────────────────────────────────────────────────── + + +def test_denorm_building_feature_inserts() -> None: + """Полный building feature → INSERT с правильными параметрами.""" + db = _make_mock_session() + + result = denorm_building_feature( + db, + feature=_building_feature(), + quarter_cad="66:41:0204016", + snapshot_date="2026-05-16", + ) + + assert result is True + db.execute.assert_called_once() + call_kwargs = db.execute.call_args[0][1] + assert call_kwargs["cad_num"] == "66:41:0204016:10:1" + assert call_kwargs["purpose"] == "Многоквартирный дом" + assert call_kwargs["floors"] == 9 + assert call_kwargs["floors_underground"] == 1 + assert call_kwargs["year_built"] == 1985 + assert call_kwargs["cost_value"] == 50000000.0 + assert call_kwargs["build_record_area"] == 3200.0 + + +def test_denorm_building_no_cad_num_skipped() -> None: + """Building feature без cad_num → False.""" + db = _make_mock_session() + feature: dict[str, Any] = { + "layer": "buildings", + "feature_id": None, + "geometry": None, + "properties": {"purpose": "Нежилое здание"}, + } + result = denorm_building_feature( + db, feature=feature, quarter_cad="66:41:0204016", snapshot_date="2026-05-16" + ) + assert result is False + db.execute.assert_not_called() + + +def test_denorm_building_str_floors_coerced() -> None: + """floors_above_ground строкой → корректно парсится в int.""" + db = _make_mock_session() + result = denorm_building_feature( + db, + feature=_building_feature(floors_above_ground="12"), + quarter_cad="66:41:0204016", + snapshot_date="2026-05-16", + ) + assert result is True + call_kwargs = db.execute.call_args[0][1] + assert call_kwargs["floors"] == 12 + + +# ── denorm_dump ──────────────────────────────────────────────────────────────── + + +def test_denorm_dump_aggregates() -> None: + """3 parcels + 2 buildings + 1 unknown layer → правильные счётчики.""" + db = _make_mock_session() + + features: list[dict[str, Any]] = [ + _parcel_feature("66:41:0101001:1"), + _parcel_feature("66:41:0101001:2"), + _parcel_feature("66:41:0101001:3"), + _building_feature("66:41:0101001:1:1"), + _building_feature("66:41:0101001:1:2"), + { + "layer": "territorial_zones", + "feature_id": "tz-1", + "geometry": None, + "properties": {"type_zone": "Ж-1"}, + }, + ] + + counts = denorm_dump(db, quarter_cad="66:41:0101001", features=features) + + assert counts["parcels"] == 3 + assert counts["buildings"] == 2 + # territorial_zones не считается как ошибка — просто пропускается + assert counts["errors"] == 0 + db.commit.assert_called_once() + + +def test_denorm_dump_empty_features() -> None: + """Пустой список features → нули, commit всё равно вызывается.""" + db = _make_mock_session() + counts = denorm_dump(db, quarter_cad="66:41:0101001", features=[]) + + assert counts == {"parcels": 0, "buildings": 0, "errors": 0} + db.commit.assert_called_once() + + +def test_denorm_dump_no_cad_num_counted_as_error() -> None: + """Parcel без cad_num → denorm_parcel_feature returns False → errors += 1.""" + db = _make_mock_session() + feature: dict[str, Any] = { + "layer": "parcels", + "feature_id": None, + "geometry": None, + "properties": {}, + } + counts = denorm_dump(db, quarter_cad="66:41:0101001", features=[feature]) + assert counts["parcels"] == 0 + assert counts["errors"] == 1 diff --git a/data/sql/99_nspd_entities_denorm.sql b/data/sql/99_nspd_entities_denorm.sql new file mode 100644 index 00000000..c65a6f2f --- /dev/null +++ b/data/sql/99_nspd_entities_denorm.sql @@ -0,0 +1,171 @@ +-- 99_nspd_entities_denorm.sql +-- Context : Denormalized entity tables для NSPD parcel и building features. +-- nspd_quarter_dumps.features_json хранит все feature'ы в JSONB; эти +-- таблицы дают быстрый lookup по cad_num без json-unpacking на query time. +-- Источник: layer="parcels" (NSPD layer 36048) и layer="buildings" (36049) +-- из features_json. Геометрия трансформируется 3857→4326 при UPSERT. +-- +-- Dependencies: 88_nspd_quarter_dumps.sql (logical; таблица должна существовать до backfill) +-- Deploy order: после 98_nspd_quarter_dumps_opportunity_flag.sql +-- Idempotent: yes — IF NOT EXISTS на всех объектах +-- Related: +-- - backend/app/services/scrapers/nspd_denorm.py :: denorm_parcel_feature / denorm_building_feature +-- - backend/app/workers/tasks/nspd_denorm_backfill.py :: backfill_all_dumps +-- - backend/app/workers/tasks/nspd_sync.py :: harvest_quarter (inline denorm) +-- Issue: #94 PR3 + +BEGIN; + +-- ── nspd_parcels ────────────────────────────────────────────────────────────── +-- Denormalized parcel features из NSPD quarter dumps (layer="parcels", id 36048). +-- Каждая строка = один земельный участок из любого квартала. +-- PK cad_num: один участок может пересекать несколько кварталов, берём +-- последний harvest (ON CONFLICT DO UPDATE). + +CREATE TABLE IF NOT EXISTS nspd_parcels ( + -- Кадастровый номер участка (4-сегментный, e.g. '66:41:0204016:10'). + -- Первичный ключ — natural, стабильный идентификатор в ЕГРН. + cad_num TEXT PRIMARY KEY, + + -- Квартал из которого была взята последняя версия этого участка. + quarter_cad TEXT NOT NULL, + + -- ВРИ (вид разрешённого использования) — properties.permitted_use из NSPD. + -- Nullable: НСПД не всегда возвращает ВРИ для старых записей. + permitted_use TEXT, + + -- Категория земель (properties.land_category). + -- Nullable: может отсутствовать в ответе. + land_category TEXT, + + -- Кадастровая стоимость ЗУ (properties.cost_value), руб. + cost_value NUMERIC, + + -- Кадастровая стоимость за м² = cost_value / area_sqm. + -- Вычисляется при UPSERT; NULL если area_sqm = 0 или cost_value = NULL. + cost_per_m2 NUMERIC, + + -- Площадь участка (properties.area), м². + area_sqm NUMERIC, + + -- Адрес участка (properties.address). + address TEXT, + + -- Геометрия участка в WGS84 (трансформируется из EPSG:3857 при UPSERT). + -- Geometry не Polygon т.к. НСПД иногда возвращает MultiPolygon или Point. + geom GEOMETRY(Geometry, 4326), + + -- Дата snapshot'а (день harvest'а). + snapshot_date DATE NOT NULL DEFAULT CURRENT_DATE, + + -- Timestamp последнего UPSERT. + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +COMMENT ON TABLE nspd_parcels IS + 'Denormalized parcel features из NSPD quarter dumps (layer=parcels, id 36048). ' + 'PK = cad_num (последний harvest wins). Заполняется Celery task backfill_all_dumps ' + 'или inline в harvest_quarter. Источник: features_json JSONB колонка nspd_quarter_dumps. ' + 'Геометрия трансформирована EPSG:3857→4326. (#94 PR3)'; + +COMMENT ON COLUMN nspd_parcels.cad_num IS + 'Кадастровый номер участка (4-сегментный). Natural PK из НСПД.'; + +COMMENT ON COLUMN nspd_parcels.quarter_cad IS + '3-сегментный квартал последнего harvest'а. Используется для debug/audit.'; + +COMMENT ON COLUMN nspd_parcels.cost_per_m2 IS + 'Кадастровая стоимость за м² (cost_value / area_sqm). ' + 'NULL если area_sqm = 0 или cost_value отсутствует. Вычисляется в Python при UPSERT.'; + +-- Индекс для lookup по кварталу (join с nspd_quarter_dumps). +CREATE INDEX IF NOT EXISTS idx_nspd_parcels_quarter_cad + ON nspd_parcels (quarter_cad); + +-- GIST индекс на геометрию для spatial queries (соседние участки, ST_DWithin). +CREATE INDEX IF NOT EXISTS idx_nspd_parcels_geom_gist + ON nspd_parcels USING GIST (geom); + +-- Partial B-tree на cost_per_m2 для ценовых аналитических запросов. +CREATE INDEX IF NOT EXISTS idx_nspd_parcels_cost_per_m2 + ON nspd_parcels (cost_per_m2) + WHERE cost_per_m2 IS NOT NULL; + + +-- ── nspd_buildings ──────────────────────────────────────────────────────────── +-- Denormalized building features из NSPD quarter dumps (layer="buildings", id 36049). +-- Каждая строка = одно здание/сооружение ЕГРН. + +CREATE TABLE IF NOT EXISTS nspd_buildings ( + -- Кадастровый номер здания (5-сегментный, e.g. '66:41:0204016:10:1'). + cad_num TEXT PRIMARY KEY, + + -- Квартал последнего harvest'а. + quarter_cad TEXT NOT NULL, + + -- Назначение здания (properties.purpose). + -- Ключевые значения: 'Многоквартирный дом', 'Нежилое здание', etc. + purpose TEXT, + + -- Количество надземных этажей (properties.floors_above_ground). + floors INTEGER, + + -- Количество подземных этажей (properties.floors_underground). + floors_underground INTEGER, + + -- Год постройки (properties.year_built). + year_built INTEGER, + + -- Кадастровая стоимость здания (properties.cost_value), руб. + cost_value NUMERIC, + + -- Площадь по данным ГКН (properties.build_record_area), м². + build_record_area NUMERIC, + + -- Адрес здания (properties.address). + address TEXT, + + -- Геометрия здания в WGS84 (трансформируется из EPSG:3857 при UPSERT). + geom GEOMETRY(Geometry, 4326), + + -- Дата snapshot'а. + snapshot_date DATE NOT NULL DEFAULT CURRENT_DATE, + + -- Timestamp последнего UPSERT. + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +COMMENT ON TABLE nspd_buildings IS + 'Denormalized building features из NSPD quarter dumps (layer=buildings, id 36049). ' + 'PK = cad_num (последний harvest wins). Заполняется Celery task backfill_all_dumps ' + 'или inline в harvest_quarter. Ключевой use-case: МКД-соседи для P2-scoring. ' + 'Геометрия трансформирована EPSG:3857→4326. (#94 PR3)'; + +COMMENT ON COLUMN nspd_buildings.purpose IS + 'Назначение здания (properties.purpose). Для МКД = "Многоквартирный дом". ' + 'Используется в partial index idx_nspd_buildings_mkd для быстрой фильтрации.'; + +COMMENT ON COLUMN nspd_buildings.floors IS + 'Количество надземных этажей (properties.floors_above_ground из НСПД). ' + 'Nullable: старые ОКС в ЕГРН не всегда имеют этажность.'; + +-- Индекс для lookup по кварталу. +CREATE INDEX IF NOT EXISTS idx_nspd_buildings_quarter_cad + ON nspd_buildings (quarter_cad); + +-- GIST индекс на геометрию (МКД-соседи через ST_DWithin). +CREATE INDEX IF NOT EXISTS idx_nspd_buildings_geom_gist + ON nspd_buildings USING GIST (geom); + +-- Partial index для МКД (основной use-case P2-scoring). +-- ILIKE не работает в partial index WHERE clause — используем точный строковый prefixmatch. +-- Здания с purpose IS NULL или другим purpose — в idx_nspd_buildings_geom_gist. +CREATE INDEX IF NOT EXISTS idx_nspd_buildings_mkd + ON nspd_buildings (quarter_cad) + WHERE purpose = 'Многоквартирный дом'; + +COMMENT ON INDEX idx_nspd_buildings_mkd IS + 'Partial index для быстрого lookup МКД по кварталу (P2-scoring neighbors). ' + 'WHERE purpose = exact строка из НСПД. (#94 PR3)'; + +COMMIT;