fix(cadastre): auto-heal cad_quarters_geom from CAT_QUARTER_STATS #181
2 changed files with 328 additions and 1 deletions
|
|
@ -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 ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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"}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue