backend/app/services/cadastre/bulk_harvest.py

@@ -404,30 +404,42 @@ def upsert_features(
 
     for feature in features:
         cat = feature.category_id
-        if cat == CAT_PARCEL:
-            upsert_parcel(db, feature, source=source)
-            counts["parcels"] += 1
-        elif cat == CAT_BUILDING:
-            upsert_building(db, feature, source=source)
-            counts["buildings"] += 1
-        elif cat == CAT_CONSTRUCTION:
-            upsert_construction(db, feature, source=source)
-            counts["constructions"] += 1
-        elif cat == CAT_ONC:
-            upsert_onc(db, feature, source=source)
-            counts["oncs"] += 1
-        elif cat == CAT_ENK:
-            upsert_enk(db, feature, source=source)
-            counts["enks"] += 1
-        elif cat in CAT_ZOUIT:
-            upsert_zouit(db, feature, source=source)
-            counts["zouit"] += 1
-        elif cat == CAT_QUARTER_STATS:
-            pass  # обрабатывается отдельно в upsert_quarter_stats
-        else:
-            if cat is not None:
-                logger.debug("upsert_features: неизвестный category_id=%s, skip", cat)
-            counts["skipped"] += 1
+        # Per-feature SAVEPOINT (backend.md SAVEPOINT rule): один битый feature (edge-case
+        # GeoJSON / NOT NULL violation / type mismatch) НЕ должен ронять весь snapshot-tx
+        # квартала. Зеркалит _grid_walk_category (~368) и _save_territorial_zones (~1234):
+        # begin_nested() rollback'ит только этот feature, остальные сохраняются. Счётчик
+        # инкрементим ТОЛЬКО при успехе (внутри блока) → counts остаются точными.
+        try:
+            with db.begin_nested():
+                if cat == CAT_PARCEL:
+                    upsert_parcel(db, feature, source=source)
+                    counts["parcels"] += 1
+                elif cat == CAT_BUILDING:
+                    upsert_building(db, feature, source=source)
+                    counts["buildings"] += 1
+                elif cat == CAT_CONSTRUCTION:
+                    upsert_construction(db, feature, source=source)
+                    counts["constructions"] += 1
+                elif cat == CAT_ONC:
+                    upsert_onc(db, feature, source=source)
+                    counts["oncs"] += 1
+                elif cat == CAT_ENK:
+                    upsert_enk(db, feature, source=source)
+                    counts["enks"] += 1
+                elif cat in CAT_ZOUIT:
+                    upsert_zouit(db, feature, source=source)
+                    counts["zouit"] += 1
+                elif cat == CAT_QUARTER_STATS:
+                    pass  # обрабатывается отдельно в upsert_quarter_stats
+                else:
+                    if cat is not None:
+                        logger.debug("upsert_features: неизвестный category_id=%s, skip", cat)
+                    counts["skipped"] += 1
+        except Exception as e:
+            # Логируем + skip (не re-raise) — один битый feature не отравляет квартал.
+            logger.warning(
+                "upsert_features: feature failed (cat=%s, id=%s): %s", cat, feature.id, e
+            )
 
     return counts
 

backend/app/workers/tasks/nspd_geo.py

@@ -27,6 +27,7 @@ from sqlalchemy import text
 from sqlalchemy.orm import Session
 
 from app.core.db import SessionLocal
+from app.services.cadastre.bulk_harvest import _derive_quarter_cad, _safe_int, _safe_numeric
 from app.services.scrapers.nspd_lite import (
     NspdLiteError,
     NspdLiteWafError,
@@ -136,7 +137,12 @@ def _save_quarter(db: Session, payload: dict, cad_num: str) -> int:
         text(
             """
             INSERT INTO cad_quarters_geom (cad_number, geom, raw_props, fetched_at)
-            VALUES (:cad, ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:g), 3857), 4326),
+            VALUES (:cad,
+                    -- cad_quarters_geom.geom = geometry(MultiPolygon,4326) NOT NULL (migration 58).
+                    -- NSPD может вернуть одноконтурный Polygon → ST_Multi coerce'ит в MultiPolygon,
+                    -- иначе "Geometry type (Polygon) does not match column type (MultiPolygon)".
+                    -- 3857→4326 transform как в upsert_quarter_geom_from_feature (Mercator).
+                    ST_Multi(ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:g), 3857), 4326)),
                     CAST(:props AS jsonb), NOW())
             ON CONFLICT (cad_number) DO UPDATE
                SET geom = EXCLUDED.geom,
@@ -154,7 +160,14 @@ def _save_quarter(db: Session, payload: dict, cad_num: str) -> int:
 
 
 def _save_building(db: Session, payload: dict, cad_num: str) -> int:
-    """Парсит NSPD-response для здания и UPSERT'ит в cad_buildings."""
+    """Парсит NSPD-response для здания и UPSERT'ит в cad_buildings (wide schema, migration 92).
+
+    cad_buildings (migration 92) — широкая схема:
+      - quarter_cad_number (НЕ quarter_cad_num),
+      - floors INT (NSPD может отдать "5-7" → _safe_int → NULL, иначе InvalidTextRepresentation),
+      - source TEXT NOT NULL (нет дефолта) → обязателен,
+      - geom GEOMETRY(Geometry, 4326) → полиморфный, ST_Multi НЕ нужен (mirror upsert_building).
+    """
     feats = (payload.get("data") or {}).get("features") or []
     if not feats:
         return 0
@@ -167,40 +180,42 @@ def _save_building(db: Session, payload: dict, cad_num: str) -> int:
         # Только этого здания (NSPD может вернуть здания всего квартала)
         if opts.get("cad_num") != cad_num:
             continue
-        import json as _json
 
+        quarter_cad = opts.get("quarter_cad_number") or _derive_quarter_cad(cad_num)
         db.execute(
             text(
                 """
                 INSERT INTO cad_buildings (
-                    cad_num, quarter_cad_num, geom, purpose, building_name,
-                    readable_address, area, floors, raw_props
+                    cad_num, quarter_cad_number, geom, purpose, building_name,
+                    readable_address, area, floors, raw_props, source
                 ) VALUES (
                     :cad, :qcad,
                     ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:g), 3857), 4326),
-                    :purpose, :name, :addr, :area, :floors, CAST(:props AS jsonb)
+                    :purpose, :name, :addr, :area, :floors, CAST(:props AS jsonb), :source
                 )
                 ON CONFLICT (cad_num) DO UPDATE
-                   SET quarter_cad_num = EXCLUDED.quarter_cad_num,
+                   SET quarter_cad_number = EXCLUDED.quarter_cad_number,
                        geom = EXCLUDED.geom,
                        purpose = EXCLUDED.purpose,
                        building_name = EXCLUDED.building_name,
                        readable_address = EXCLUDED.readable_address,
                        area = EXCLUDED.area,
                        floors = EXCLUDED.floors,
-                       raw_props = EXCLUDED.raw_props
+                       raw_props = EXCLUDED.raw_props,
+                       source = EXCLUDED.source
                 """
             ),
             {
                 "cad": cad_num,
-                "qcad": opts.get("quarter_cad_number"),
-                "g": _json.dumps(geom_geojson, ensure_ascii=False),
+                "qcad": quarter_cad,
+                "g": json.dumps(geom_geojson, ensure_ascii=False),
                 "purpose": opts.get("purpose"),
                 "name": opts.get("building_name") or opts.get("name"),
                 "addr": opts.get("readable_address"),
-                "area": opts.get("area"),
-                "floors": opts.get("floors"),
-                "props": _json.dumps(opts, ensure_ascii=False),
+                "area": _safe_numeric(opts.get("area")),
+                "floors": _safe_int(opts.get("floors")),
+                "props": json.dumps(opts, ensure_ascii=False),
+                "source": "nspd",
             },
         )
         n += 1
@@ -226,7 +241,14 @@ def _save_parcel(db: Session, payload: dict, cad_num: str) -> int:
         text(
             """
             INSERT INTO cad_parcels_geom (cad_num, geom, raw_props, fetched_at)
-            VALUES (:cad, ST_SetSRID(ST_GeomFromGeoJSON(:g), 4326),
+            VALUES (:cad,
+                    -- cad_parcels.geom = geometry(MultiPolygon,4326) (migration 93;
+                    -- cad_parcels_geom теперь back-compat VIEW над cad_parcels).
+                    -- NSPD/rosreestr2coord отдаёт одноконтурный Polygon → ST_Multi coerce'ит
+                    -- в MultiPolygon, иначе "Geometry type (Polygon) does not match column
+                    -- type (MultiPolygon)". БЕЗ ST_Transform: parcel GeoJSON уже WGS84
+                    -- (см. docstring) — только ST_SetSRID.
+                    ST_Multi(ST_SetSRID(ST_GeomFromGeoJSON(:g), 4326)),
                     CAST(:props AS jsonb), NOW())
             ON CONFLICT (cad_num) DO UPDATE
                SET geom = EXCLUDED.geom,

backend/tests/services/test_cadastre_bulk.py

@@ -353,6 +353,70 @@ def test_upsert_features_quarter_stats_not_counted() -> None:
     assert counts["skipped"] == 0
 
 
+def _db_with_real_savepoints() -> tuple[MagicMock, list[Any]]:
+    """Mock Session где begin_nested() — настоящий context manager (для SAVEPOINT-теста).
+
+    Возвращает (db, savepoint_calls) — savepoint_calls фиксирует каждый __enter__,
+    т.е. сколько раз был открыт per-feature SAVEPOINT.
+    """
+    savepoint_calls: list[Any] = []
+    db = MagicMock()
+
+    def _begin_nested() -> Any:
+        cm = MagicMock()
+        cm.__enter__ = MagicMock(side_effect=lambda: savepoint_calls.append(1))
+        cm.__exit__ = MagicMock(return_value=False)
+        return cm
+
+    db.begin_nested = MagicMock(side_effect=_begin_nested)
+    return db, savepoint_calls
+
+
+def test_upsert_features_wraps_each_row_in_savepoint() -> None:
+    """backend.md SAVEPOINT rule: upsert_features открывает begin_nested() per feature."""
+    parcel = _make_parcel_feature("66:41:0303161:1")
+    building = _make_building_feature("66:41:0303161:1:1")
+    db, savepoint_calls = _db_with_real_savepoints()
+
+    with (
+        patch("app.services.cadastre.bulk_harvest.upsert_parcel"),
+        patch("app.services.cadastre.bulk_harvest.upsert_building"),
+    ):
+        upsert_features(db, [parcel, building], source="search")
+
+    # По одному SAVEPOINT на каждый из 2 features
+    assert db.begin_nested.call_count == 2
+    assert len(savepoint_calls) == 2
+
+
+def test_upsert_features_one_bad_feature_does_not_abort_others() -> None:
+    """Один битый feature (raise в upsert_*) НЕ роняет весь snapshot — остальные сохраняются,
+    counts остаются точными (инкремент только при успехе).
+
+    Раньше: без per-row begin_nested один битый feature отравлял всю tx квартала →
+    все features rollback'ились (а счётчики уже инкрементированы = inconsistent).
+    """
+    parcel_ok = _make_parcel_feature("66:41:0303161:1")
+    building_bad = _make_building_feature("66:41:0303161:1:1")
+    parcel_ok2 = _make_parcel_feature("66:41:0303161:2")
+    db, _savepoint_calls = _db_with_real_savepoints()
+
+    with (
+        patch("app.services.cadastre.bulk_harvest.upsert_parcel") as mock_parcel,
+        patch("app.services.cadastre.bulk_harvest.upsert_building") as mock_building,
+    ):
+        # Здание падает (например type mismatch / NOT NULL), парцелы — ок
+        mock_building.side_effect = Exception("geom type mismatch / NOT NULL violation")
+        counts = upsert_features(db, [parcel_ok, building_bad, parcel_ok2], source="search")
+
+    # Оба парцела обработаны несмотря на падение здания между ними
+    assert mock_parcel.call_count == 2
+    mock_building.assert_called_once()
+    # Счётчики точные: 2 успешных парцела, 0 зданий (битое не считается)
+    assert counts["parcels"] == 2
+    assert counts["buildings"] == 0
+
+
 # ── upsert_parcel SQL params ─────────────────────────────────────────────────
 
 

backend/tests/workers/test_nspd_geo.py

@@ -0,0 +1,294 @@
+"""Тесты для backend/app/workers/tasks/nspd_geo.py — on-demand НСПД-ингест.
+
+Покрывает 3 prod-confirmed бага «пользователь вводит новый/произвольный реальный участок»:
+  - Finding 1: _save_parcel / _save_quarter должны ST_Multi() geom (колонки MultiPolygon,
+    migration 93 / 58) — иначе "Geometry type (Polygon) does not match column type (MultiPolygon)".
+  - Finding 2: _save_building пишет в широкую схему cad_buildings (migration 92):
+    quarter_cad_number (не quarter_cad_num), source NOT NULL, floors через _safe_int.
+
+Mock DB через unittest.mock — без реального PostgreSQL / NSPD сети.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock
+
+from app.workers.tasks.nspd_geo import _save_building, _save_parcel, _save_quarter
+
+# ── Фикстуры payload (форма ответа _persist_target: {"data": {"features": [...]}}) ──
+
+
+def _parcel_payload(cad_num: str = "66:41:0610029:83") -> dict[str, Any]:
+    """NSPD-ответ для ЗУ (одноконтурный Polygon — триггер бага type-mismatch)."""
+    return {
+        "data": {
+            "features": [
+                {
+                    "type": "Feature",
+                    "geometry": {
+                        "type": "Polygon",
+                        "coordinates": [
+                            [
+                                [60.6, 56.8],
+                                [60.61, 56.8],
+                                [60.61, 56.81],
+                                [60.6, 56.81],
+                                [60.6, 56.8],
+                            ]
+                        ],
+                    },
+                    "properties": {"options": {"cad_num": cad_num}},
+                }
+            ]
+        }
+    }
+
+
+def _quarter_payload(cad_num: str = "66:41:0610029") -> dict[str, Any]:
+    """NSPD-ответ для квартала (одноконтурный Polygon в EPSG:3857)."""
+    return {
+        "data": {
+            "features": [
+                {
+                    "type": "Feature",
+                    "geometry": {
+                        "type": "Polygon",
+                        "coordinates": [
+                            [
+                                [6090000.0, 7590000.0],
+                                [6090200.0, 7590000.0],
+                                [6090200.0, 7590200.0],
+                                [6090000.0, 7590200.0],
+                                [6090000.0, 7590000.0],
+                            ]
+                        ],
+                    },
+                    "properties": {"options": {}},
+                }
+            ]
+        }
+    }
+
+
+def _building_payload(
+    cad_num: str = "66:41:0610029:83",
+    floors: Any = "5-7",
+    quarter: str | None = "66:41:0610029",
+) -> dict[str, Any]:
+    """NSPD-ответ для здания. floors по умолчанию range-строка "5-7" (триггер бага floors INT)."""
+    opts: dict[str, Any] = {
+        "cad_num": cad_num,
+        "purpose": "Жилое",
+        "building_name": "Тестовое здание",
+        "readable_address": "г Екатеринбург, ул Тестовая, д 1",
+        "area": "1234.56",
+        "floors": floors,
+    }
+    if quarter is not None:
+        opts["quarter_cad_number"] = quarter
+    return {
+        "data": {
+            "features": [
+                {
+                    "type": "Feature",
+                    "geometry": {
+                        "type": "Polygon",
+                        "coordinates": [
+                            [
+                                [6090000.0, 7590000.0],
+                                [6090050.0, 7590000.0],
+                                [6090050.0, 7590050.0],
+                                [6090000.0, 7590050.0],
+                                [6090000.0, 7590000.0],
+                            ]
+                        ],
+                    },
+                    "properties": {"options": opts},
+                }
+            ]
+        }
+    }
+
+
+def _capturing_db() -> tuple[MagicMock, list[tuple[str, Any]]]:
+    """Mock Session, фиксирующий (sql_text, params) каждого db.execute."""
+    captured: list[tuple[str, Any]] = []
+    db = MagicMock()
+    db.execute = lambda stmt, params=None: captured.append((str(stmt), params)) or MagicMock()
+    return db, captured
+
+
+def _strip_sql_comments(sql: str) -> str:
+    """Убрать `-- ...` комментарии — чтобы assert'ить ИСПОЛНЯЕМЫЙ SQL, не текст комментов.
+
+    Комментарии легитимно упоминают ST_Transform/ST_Multi, поэтому проверяем выражения
+    только в реальном SQL.
+    """
+    lines = []
+    for line in sql.splitlines():
+        idx = line.find("--")
+        lines.append(line if idx == -1 else line[:idx])
+    return "\n".join(lines)
+
+
+# ── Finding 1: ST_Multi для MultiPolygon-колонок ─────────────────────────────
+
+
+def test_save_parcel_sql_wraps_geom_in_st_multi() -> None:
+    """cad_parcels.geom = MultiPolygon (migration 93) → SQL должен ST_Multi() обернуть geom.
+
+    Иначе одноконтурный Polygon из rosreestr2coord → "Geometry type (Polygon) does not
+    match column type (MultiPolygon)" → on-demand ингест участка крашится.
+    """
+    db, captured = _capturing_db()
+    n = _save_parcel(db, _parcel_payload(), "66:41:0610029:83")
+
+    assert n == 1
+    assert len(captured) == 1
+    sql, _params = captured[0]
+    assert "ST_Multi(" in sql, "SQL _save_parcel должен содержать ST_Multi для MultiPolygon schema"
+
+
+def test_save_parcel_sql_no_transform_parcel_is_4326() -> None:
+    """parcel GeoJSON уже WGS84 (rosreestr2coord) — только ST_SetSRID(...,4326), без ST_Transform.
+
+    ST_Transform 3857→4326 на 4326-геометрии схлопнул бы участок в точку у (0,0).
+    """
+    db, captured = _capturing_db()
+    _save_parcel(db, _parcel_payload(), "66:41:0610029:83")
+
+    sql, _params = captured[0]
+    executable = _strip_sql_comments(sql)
+    assert "ST_Transform" not in executable, "parcel уже 4326 — ST_Transform схлопнет в точку"
+    assert "ST_SetSRID(ST_GeomFromGeoJSON(:g), 4326)" in executable
+
+
+def test_save_quarter_sql_wraps_geom_in_st_multi() -> None:
+    """cad_quarters_geom.geom = MultiPolygon NOT NULL (migration 58) → ST_Multi() обязателен."""
+    db, captured = _capturing_db()
+    n = _save_quarter(db, _quarter_payload(), "66:41:0610029")
+
+    assert n == 1
+    assert len(captured) == 1
+    sql, _params = captured[0]
+    assert "ST_Multi(" in sql, "SQL _save_quarter должен содержать ST_Multi для MultiPolygon schema"
+
+
+def test_save_quarter_sql_keeps_3857_to_4326_transform() -> None:
+    """quarter GeoJSON в Mercator (3857) → ST_Transform к 4326 сохраняется (mirror bulk path)."""
+    db, captured = _capturing_db()
+    _save_quarter(db, _quarter_payload(), "66:41:0610029")
+
+    sql, _params = captured[0]
+    assert "ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:g), 3857), 4326)" in sql
+
+
+def test_save_parcel_empty_features_returns_zero() -> None:
+    """Пустой features → 0, без db.execute."""
+    db, captured = _capturing_db()
+    assert _save_parcel(db, {"data": {"features": []}}, "66:41:0610029:83") == 0
+    assert captured == []
+
+
+# ── Finding 2: _save_building → широкая схема cad_buildings (migration 92) ─────
+
+
+def test_save_building_uses_quarter_cad_number_column() -> None:
+    """migration 92: колонка quarter_cad_number (НЕ quarter_cad_num) → иначе UndefinedColumn."""
+    db, captured = _capturing_db()
+    n = _save_building(db, _building_payload(), "66:41:0610029:83")
+
+    assert n == 1
+    sql, params = captured[0]
+    assert "quarter_cad_number" in sql
+    # старое имя колонки не должно встречаться (как отдельное слово quarter_cad_num)
+    assert "quarter_cad_num " not in sql
+    assert "quarter_cad_num," not in sql
+    assert "quarter_cad_num=" not in sql
+    assert params["qcad"] == "66:41:0610029"
+
+
+def test_save_building_sets_source_not_null() -> None:
+    """cad_buildings.source NOT NULL без дефолта (migration 92) → INSERT должен биндить source."""
+    db, captured = _capturing_db()
+    _save_building(db, _building_payload(), "66:41:0610029:83")
+
+    sql, params = captured[0]
+    assert ":source" in sql
+    assert params["source"] == "nspd"
+
+
+def test_save_building_safe_int_floors_range_string() -> None:
+    """floors INT: NSPD range "5-7" → _safe_int → None (иначе InvalidTextRepresentation).
+
+    Сырой "5-7" в floors INT крашил бы INSERT.
+    """
+    db, captured = _capturing_db()
+    _save_building(db, _building_payload(floors="5-7"), "66:41:0610029:83")
+
+    _sql, params = captured[0]
+    assert params["floors"] is None  # "5-7" не парсится → None, не сырая строка
+
+
+def test_save_building_safe_int_floors_numeric() -> None:
+    """floors "9" → int 9 (валидное число парсится)."""
+    db, captured = _capturing_db()
+    _save_building(db, _building_payload(floors="9"), "66:41:0610029:83")
+
+    _sql, params = captured[0]
+    assert params["floors"] == 9
+
+
+def test_save_building_derives_quarter_when_missing() -> None:
+    """Если NSPD не дал quarter_cad_number → выводим из cad_num (первые 3 сегмента)."""
+    db, captured = _capturing_db()
+    _save_building(db, _building_payload(quarter=None), "66:41:0610029:83")
+
+    _sql, params = captured[0]
+    assert params["qcad"] == "66:41:0610029"
+
+
+def test_save_building_geom_transform_no_st_multi() -> None:
+    """cad_buildings.geom = GEOMETRY(Geometry, 4326) полиморфный → ST_Multi НЕ нужен,
+    transform 3857→4326 сохраняется (mirror upsert_building)."""
+    db, captured = _capturing_db()
+    _save_building(db, _building_payload(), "66:41:0610029:83")
+
+    sql, _params = captured[0]
+    assert "ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:g), 3857), 4326)" in sql
+    assert "ST_Multi(" not in sql  # полиморфная колонка — coerce не нужен
+
+
+def test_save_building_skips_feature_with_other_cad_num() -> None:
+    """NSPD может вернуть здания всего квартала — берём только запрошенный cad_num."""
+    db, captured = _capturing_db()
+    payload = _building_payload(cad_num="66:41:0610029:99")  # чужой cad_num
+    n = _save_building(db, payload, "66:41:0610029:83")
+
+    assert n == 0
+    assert captured == []
+
+
+def test_save_building_area_safe_numeric() -> None:
+    """area NUMERIC: строковое "1234.56" → float через _safe_numeric."""
+    db, captured = _capturing_db()
+    _save_building(db, _building_payload(), "66:41:0610029:83")
+
+    _sql, params = captured[0]
+    assert params["area"] == 1234.56
+
+
+# ── psycopg v3 CAST-trap guard (backend.md) ─────────────────────────────────
+
+
+def test_save_sql_uses_cast_not_double_colon() -> None:
+    """SQL _save_* использует CAST(:x AS jsonb), не :x::jsonb (SQLAlchemy psycopg v3 trap)."""
+    db, captured = _capturing_db()
+    _save_parcel(db, _parcel_payload(), "66:41:0610029:83")
+    _save_quarter(db, _quarter_payload(), "66:41:0610029")
+    _save_building(db, _building_payload(), "66:41:0610029:83")
+
+    for sql, _params in captured:
+        assert "::jsonb" not in sql, f"SQL содержит ::jsonb — нужен CAST(:x AS jsonb): {sql[:80]}"
+        assert "CAST(:props AS jsonb)" in sql