diff --git a/backend/app/services/cadastre/bulk_harvest.py b/backend/app/services/cadastre/bulk_harvest.py index bcc262a3..e782f61f 100644 --- a/backend/app/services/cadastre/bulk_harvest.py +++ b/backend/app/services/cadastre/bulk_harvest.py @@ -237,10 +237,19 @@ async def harvest_quarter( } ) - # ── Phase 4: quarter stats из snapshot ─────────────────────────────────── + # ── Phase 4: quarter stats + auto-heal geom из snapshot ───────────────── stats_features = [f for f in snapshot.features if f.category_id == CAT_QUARTER_STATS] if stats_features: upsert_quarter_stats(db, quarter, stats_features[0]) + # NSPD изредка возвращает edge-case GeoJSON (self-intersecting, unclosed ring) — + # ST_GeomFromGeoJSON упадёт в PostgreSQL exception. Savepoint изолирует failure: + # rollback'нёт только geom UPDATE, успешный upsert_quarter_stats сохранится. + try: + with db.begin_nested(): + if upsert_quarter_geom_from_feature(db, quarter, stats_features[0]): + logger.info("harvest_quarter: auto-healed cad_quarters_geom for %s", quarter) + except Exception as e: + logger.warning("harvest_quarter: geom auto-heal failed for %s: %s", quarter, e) db.commit() update_progress({"phase": "done", "quarter": quarter}) @@ -946,6 +955,90 @@ def upsert_zouit(db: Session, feature: NSPDBulkFeature, source: str = "search") ) +# ── cad_quarters_geom auto-heal ───────────────────────────────────────────── + + +def upsert_quarter_geom_from_feature( + db: Session, + quarter: str, + feature: NSPDBulkFeature, + min_width_m: float = 100.0, + max_width_m: float = 10000.0, +) -> bool: + """Auto-heal cad_quarters_geom используя CAT_QUARTER_STATS feature. + + NSPD возвращает геометрию квартала в CAT_QUARTER_STATS feature в EPSG:3857. + Трансформируем в EPSG:4326 на стороне DB (ST_Transform), сохраняем как ST_Multi. + + Sanity-check выполняется внутри SQL: UPDATE происходит только если: + - Новая geom является валидным полигоном шириной [min_width_m, max_width_m] в EPSG:3857 + - Существующая geom отсутствует (NULL) ИЛИ сломана (ширина вне допустимого диапазона) + + Это предотвращает перезапись здоровой geom сломанной и не upserт сломанной обратно. + + Args: + db: SQLAlchemy session (без commit — caller коммитит вместе с quarter_stats). + quarter: кадастровый номер квартала (cad_number). + feature: NSPDBulkFeature с category_id=CAT_QUARTER_STATS. + min_width_m: минимальная ширина bbox в метрах (EPSG:3857) — по умолчанию 100м. + max_width_m: максимальная ширина bbox в метрах (EPSG:3857) — по умолчанию 10000м. + + Returns: + True если geom была обновлена, False если пропущено (нет геометрии / broken / уже healthy). + """ + geom = feature.geometry + if not geom or geom.get("type") not in ("Polygon", "MultiPolygon"): + return False + + geom_json = json.dumps(geom) + + result = db.execute( + text(""" + WITH new_geom AS ( + SELECT ST_Multi( + ST_Transform( + ST_SetSRID(ST_GeomFromGeoJSON(CAST(:geom AS text)), 3857), + 4326 + ) + ) AS g + ), + new_geom_3857 AS ( + SELECT ST_Transform(g, 3857) AS g3 FROM new_geom + ), + width_check AS ( + SELECT + (ST_XMax(g3) - ST_XMin(g3)) AS w + FROM new_geom_3857 + ) + UPDATE cad_quarters_geom cqg + SET + geom = (SELECT g FROM new_geom), + source = 'quarter_stats_autoheal', + fetched_at = NOW() + FROM width_check + WHERE cqg.cad_number = :quarter + AND width_check.w BETWEEN :min_w AND :max_w + AND ( + cqg.geom IS NULL + OR ( + ST_XMax(ST_Transform(cqg.geom, 3857)) + - ST_XMin(ST_Transform(cqg.geom, 3857)) + NOT BETWEEN :min_w AND :max_w + ) + ) + RETURNING cqg.cad_number + """), + { + "geom": geom_json, + "quarter": quarter, + "min_w": min_width_m, + "max_w": max_width_m, + }, + ) + updated = result.scalar() + return updated is not None + + # ── cad_quarter_stats upsert ───────────────────────────────────────────────── diff --git a/backend/tests/services/test_cadastre_bulk.py b/backend/tests/services/test_cadastre_bulk.py index c5e9b722..9251b821 100644 --- a/backend/tests/services/test_cadastre_bulk.py +++ b/backend/tests/services/test_cadastre_bulk.py @@ -14,11 +14,13 @@ import pytest from app.schemas.nspd_bulk import NSPDBulkFeature, QuarterSnapshot from app.services.cadastre.bulk_harvest import ( CAT_ENK, + CAT_QUARTER_STATS, CAT_ZOUIT, _derive_quarter_cad, _safe_int, _safe_numeric, upsert_features, + upsert_quarter_geom_from_feature, ) from app.services.cadastre.grid_geometry import ( generate_grid_click_points, @@ -723,3 +725,235 @@ async def test_harvest_quarter_per_cat_probe_enk_called_when_meta_nonzero() -> N if c.kwargs.get("category_id") == CAT_ENK ] assert len(per_cat_enk_calls) >= 1, "Ожидался per-cat probe с category_id=39663 (ENK)" + + +# ── auto-heal cad_quarters_geom ────────────────────────────────────────────── + + +def _make_quarter_stats_feature( + geom_type: str = "Polygon", + geom_coords: list[Any] | None = None, +) -> NSPDBulkFeature: + """Создать NSPDBulkFeature с category_id=36381 (квартальная статистика).""" + if geom_coords is None: + # 200m × 200m полигон в EPSG:3857 — здоровый квартал + geom_coords = [ + [ + [6090000.0, 7590000.0], + [6090200.0, 7590000.0], + [6090200.0, 7590200.0], + [6090000.0, 7590200.0], + [6090000.0, 7590000.0], + ] + ] + return NSPDBulkFeature.model_validate( + { + "id": 99, + "type": "Feature", + "geometry": {"type": geom_type, "coordinates": geom_coords}, + "properties": { + "category": CAT_QUARTER_STATS, + "options": { + "cnt_land": 5, + "cnt_oks": 3, + }, + }, + } + ) + + +def _mock_db_execute_scalar(scalar_value: Any) -> MagicMock: + """Mock db.execute().scalar() возвращает заданное значение.""" + mock_db = MagicMock() + mock_db.execute.return_value.scalar.return_value = scalar_value + return mock_db + + +def test_upsert_quarter_geom_skips_invalid_geometry_type() -> None: + """Point / LineString / None geometry → False (skip без DB call).""" + db = MagicMock() + + # Нет geometry вообще + feature_no_geom = NSPDBulkFeature.model_validate( + {"type": "Feature", "geometry": None, "properties": {"category": CAT_QUARTER_STATS}} + ) + assert upsert_quarter_geom_from_feature(db, "66:41:0303161", feature_no_geom) is False + db.execute.assert_not_called() + + # Point geometry + feature_point = NSPDBulkFeature.model_validate( + { + "type": "Feature", + "geometry": {"type": "Point", "coordinates": [6090000.0, 7590000.0]}, + "properties": {"category": CAT_QUARTER_STATS}, + } + ) + assert upsert_quarter_geom_from_feature(db, "66:41:0303161", feature_point) is False + db.execute.assert_not_called() + + # LineString geometry + feature_line = NSPDBulkFeature.model_validate( + { + "type": "Feature", + "geometry": { + "type": "LineString", + "coordinates": [[6090000.0, 7590000.0], [6090200.0, 7590200.0]], + }, + "properties": {"category": CAT_QUARTER_STATS}, + } + ) + assert upsert_quarter_geom_from_feature(db, "66:41:0303161", feature_line) is False + db.execute.assert_not_called() + + +def test_upsert_quarter_geom_returns_false_when_sql_updates_nothing() -> None: + """Если UPDATE не затронул строк (скипнул SQL sanity check) → scalar()=None → False. + + Симулирует случай: новая geom тоже broken (<100m) ИЛИ existing geom уже здорова. + В обоих случаях RETURNING не вернёт строку, scalar() = None. + """ + db = _mock_db_execute_scalar(None) # RETURNING ничего не вернул + feature = _make_quarter_stats_feature() # здоровая feature + + result = upsert_quarter_geom_from_feature(db, "66:41:0303161", feature) + + assert result is False + db.execute.assert_called_once() + # Убедиться что был передан корректный quarter параметр + call_params = db.execute.call_args[0][1] + assert call_params["quarter"] == "66:41:0303161" + assert "geom" in call_params + assert call_params["min_w"] == 100.0 + assert call_params["max_w"] == 10000.0 + + +def test_upsert_quarter_geom_returns_true_when_sql_updates_row() -> None: + """Если UPDATE обновил строку (existing broken + new healthy) → scalar()=cad_number → True.""" + db = _mock_db_execute_scalar("66:41:0303161") # RETURNING вернул cad_number + feature = _make_quarter_stats_feature() # здоровая feature 200m×200m + + result = upsert_quarter_geom_from_feature(db, "66:41:0303161", feature) + + assert result is True + db.execute.assert_called_once() + + +def test_upsert_quarter_geom_multipolygon_accepted() -> None: + """MultiPolygon тип тоже принимается (не только Polygon).""" + db = _mock_db_execute_scalar("66:41:0303161") + # MultiPolygon coords: список из одного полигона + mp_coords = [ + [ + [ + [6090000.0, 7590000.0], + [6090200.0, 7590000.0], + [6090200.0, 7590200.0], + [6090000.0, 7590200.0], + [6090000.0, 7590000.0], + ] + ] + ] + feature = _make_quarter_stats_feature(geom_type="MultiPolygon", geom_coords=mp_coords) + + result = upsert_quarter_geom_from_feature(db, "66:41:0303161", feature) + + assert result is True + db.execute.assert_called_once() + + +def test_upsert_quarter_geom_sql_uses_cast_not_double_colon() -> None: + """SQL использует CAST(:x AS text) не :x::text (SQLAlchemy psycopg v3 trap).""" + executed_sqls: list[str] = [] + + db = MagicMock() + db.execute.side_effect = lambda stmt, params=None: ( + executed_sqls.append(str(stmt)) or MagicMock(scalar=MagicMock(return_value=None)) + ) + + feature = _make_quarter_stats_feature() + upsert_quarter_geom_from_feature(db, "66:41:0303161", feature) + + assert len(executed_sqls) == 1 + sql = executed_sqls[0] + assert "::jsonb" not in sql + assert "::text" not in sql + assert "CAST(:geom AS text)" in sql + + +def test_upsert_quarter_geom_custom_width_bounds() -> None: + """min_width_m и max_width_m пробрасываются как параметры SQL (не hardcoded).""" + db = _mock_db_execute_scalar(None) + feature = _make_quarter_stats_feature() + + upsert_quarter_geom_from_feature( + db, "66:41:0303161", feature, min_width_m=50.0, max_width_m=5000.0 + ) + + call_params = db.execute.call_args[0][1] + assert call_params["min_w"] == 50.0 + assert call_params["max_w"] == 5000.0 + + +@pytest.mark.asyncio +async def test_harvest_quarter_geom_heal_failure_does_not_propagate() -> None: + """Phase 4: если upsert_quarter_geom_from_feature бросает (malformed GeoJSON + из NSPD → ST_GeomFromGeoJSON exception), harvest_quarter должен: + 1. Залогировать warning + 2. Rollback'нуть SAVEPOINT через begin_nested + 3. ПРОДОЛЖИТЬ — НЕ propagate exception + 4. Сохранить upsert_quarter_stats (успешный до geom failure) + """ + from app.services.cadastre.bulk_harvest import harvest_quarter + + stats_feature = _make_quarter_stats_feature() + snapshot = QuarterSnapshot( + quarter="66:41:0303161", + fetched_at="2026-05-15T10:00:00+00:00", + features=[stats_feature], + meta_counts={CAT_QUARTER_STATS: 1}, + ) + + db = MagicMock() + db.execute = MagicMock( + return_value=MagicMock( + mappings=MagicMock( + return_value=MagicMock(first=MagicMock(return_value={"phase_state": None})) + ) + ) + ) + db.commit = MagicMock() + db.begin_nested = MagicMock() # context manager mock — exit без error + + client = AsyncMock() + client.search_by_quarter = AsyncMock(return_value=snapshot) + + with ( + patch("app.services.cadastre.bulk_harvest.upsert_features") as mock_upsert, + patch("app.services.cadastre.bulk_harvest.upsert_quarter_stats") as mock_stats, + patch("app.services.cadastre.bulk_harvest.upsert_quarter_geom_from_feature") as mock_heal, + ): + mock_upsert.return_value = { + "parcels": 0, + "buildings": 0, + "constructions": 0, + "oncs": 0, + "enks": 0, + "zouit": 0, + "skipped": 0, + } + mock_heal.side_effect = Exception("ST_GeomFromGeoJSON: invalid polygon") + + # Это не должно raise + result = await harvest_quarter( + db=db, + client=client, + quarter="66:41:0303161", + job_id=1, + update_progress=lambda s: None, + ) + + # Stats был вызван (Phase 4 продолжил после geom failure) + mock_stats.assert_called_once() + mock_heal.assert_called_once() + # Phase=done — harvest завершился штатно + assert result.phase_state == {"phase": "done", "quarter": "66:41:0303161"}