diff --git a/backend/app/scrapers/nspd_bulk_client.py b/backend/app/scrapers/nspd_bulk_client.py index 5c337bd2..e2875e2c 100644 --- a/backend/app/scrapers/nspd_bulk_client.py +++ b/backend/app/scrapers/nspd_bulk_client.py @@ -104,6 +104,21 @@ class NspdBulkRateLimitError(NspdBulkError): """HTTP 429 Rate limit — исчерпаны retries после backoff.""" +class NspdBulkServerError(NspdBulkError): + """HTTP 5xx / WMS ServiceException — server-side ошибка NSPD. + + Issue #252: NSPD WMS отдаёт 500 + тело с на десятки cells + per quarter (повторяющиеся в GlitchTip BACKEND-8/10/11/12/14/16/17/24/36/37). + Это transient server-side noise, не наш баг и не клиентская ошибка (4xx). + Выделяем в отдельный подкласс, чтобы: + - caller (bulk_harvest_quarter_task) ретраил весь квартал — autoretry_for + ловит подкласс NspdBulkError, dont_autoretry_for=(NspdBulkWafError,) его + не исключает; + - grid-walk считал per-layer fail-rate и поднимал layer_X_failed-флаг, + если ВСЕ cells слоя упали с 500 (см. _grid_walk_category). + """ + + # ── NSPDBulkClient ───────────────────────────────────────────────────────────── @@ -169,7 +184,8 @@ class NSPDBulkClient: Raises: NspdBulkWafError при 403. NspdBulkRateLimitError при 429 после исчерпания retries. - NspdBulkError при прочих 4xx/5xx. + NspdBulkServerError при 5xx или WMS в теле. + NspdBulkError при прочих 4xx. """ if self._client is None or self._sem is None: raise RuntimeError("NSPDBulkClient не инициализирован — используй async with") @@ -207,10 +223,29 @@ class NSPDBulkClient: await asyncio.sleep(backoff) continue + # Issue #252: 5xx — server-side ошибка NSPD (transient). Отдельный + # подкласс, чтобы caller ретраил квартал и считал per-layer fail-rate. + if resp.status_code >= 500: + body_preview = resp.text[:300] + raise NspdBulkServerError( + f"HTTP {resp.status_code} server error: {url} — {body_preview}" + ) + if resp.status_code >= 400: body_preview = resp.text[:300] raise NspdBulkError(f"HTTP {resp.status_code}: {url} — {body_preview}") + # Issue #252: NSPD WMS (GeoServer) иногда отдаёт HTTP 200 с XML-телом + # вместо запрошенного application/json (например при + # внутренней ошибке рендера слоя). resp.json() в этом случае кинул бы + # JSONDecodeError, маскируя server-side природу сбоя. Детектим маркер + # в теле и поднимаем NspdBulkServerError — единый retry/skip-путь с 5xx. + body = resp.text + if "ServiceException" in body or " tuple[int, int]: +) -> tuple[int, int, bool]: """Grid-walk layer_id в bbox квартала. Генерирует grid_size×grid_size ячеек, для каждой делает wms_feature_info. @@ -306,7 +324,14 @@ async def _grid_walk_category( помечал worker как мёртвого. Вызываем каждые heartbeat_every cells. Returns: - (upserted_count, requests_count) + (upserted_count, requests_count, layer_failed). + + layer_failed=True — issue #252 — когда ВСЕ выполненные запросы слоя + упали на server-side 500/ServiceException и ни один не прошёл. Это + отличает «слой реально пуст в квартале» (0 discovered, 0 ошибок) от + «слой временно недоступен на стороне NSPD» (0 discovered потому что + каждый cell отдал 500). Вызывающий фиксирует флаг в harvest_meta — + квартал НЕ валится, один сбойный слой просто пропускается. """ bbox = quarter_bbox_3857(db, quarter) if bbox is None: @@ -315,13 +340,15 @@ async def _grid_walk_category( quarter, layer_id, ) - return 0, 0 + return 0, 0, False grid_points = generate_grid_click_points(bbox, grid_size=grid_size, tile_size=tile_size) discovered_cads: set[str] = set() upserted = 0 requests = 0 + server_errors = 0 # cells, упавшие на 5xx/ServiceException (issue #252) + ok_cells = 0 # cells, вернувшие ответ (пусть и пустой) — слой жив for idx, (cell_bbox, click_xy) in enumerate(grid_points): # Fix I: heartbeat каждые N cells — обновляет cadastre_jobs.heartbeat_at @@ -345,14 +372,31 @@ async def _grid_walk_category( height=tile_size, ) requests += 1 - except Exception as e: - # Fix J: NSPD WMS возвращает HTTP 500 на десятки cells per quarter — - # это server-side noise, не наш bug. logger.debug чтобы не засорять - # prod логи (раньше было warning → 200+ warnings per quarter). + ok_cells += 1 + except NspdBulkServerError as e: + # Issue #252: 5xx / WMS ServiceException — transient server-side noise + # NSPD (десятки cells per quarter). logger.debug чтобы не засорять prod + # логи. Считаем отдельно: если ВСЕ cells слоя упали так и ни один не + # прошёл — поднимем layer_failed (см. return) → harvest_meta-флаг. logger.debug( - "_grid_walk_category: wms_feature_info error layer=%d quarter=%s: %s", + "_grid_walk_category: server error layer=%d quarter=%s cell=%d: %s", layer_id, quarter, + idx, + e, + ) + requests += 1 + server_errors += 1 + continue + except Exception as e: + # Прочие (сетевые / parse) ошибки одного cell — тоже не валим квартал, + # но это НЕ server-side 500 → не учитываем в server_errors (иначе сеть + # ложно triggers layer_failed). Слой жив, просто этот cell не дошёл. + logger.debug( + "_grid_walk_category: wms_feature_info error layer=%d quarter=%s cell=%d: %s", + layer_id, + quarter, + idx, e, ) requests += 1 @@ -376,7 +420,190 @@ async def _grid_walk_category( e, ) - return upserted, requests + # Issue #252: слой считаем «сбойным» только если БЫЛИ server-side ошибки И + # ни один cell не прошёл успешно. ok_cells>0 → слой жив (0 discovered значит + # реально пусто). server_errors==0 → штатный пустой/полный обход. + layer_failed = server_errors > 0 and ok_cells == 0 + if layer_failed: + logger.warning( + "_grid_walk_category: layer=%d quarter=%s ПОЛНОСТЬЮ сбойный — " + "%d/%d cells вернули 5xx/ServiceException, 0 успешных. Слой пропущен.", + layer_id, + quarter, + server_errors, + len(grid_points), + ) + + return upserted, requests, layer_failed + + +# ── Geom backfill для участков с geom IS NULL (issue #200) ──────────────────── + + +@dataclass +class GeomBackfillResult: + """Итог докачки geom для участков с geom IS NULL (issue #200).""" + + quarters_scanned: int = 0 + parcels_targeted: int = 0 # участки без geom на входе + parcels_healed: int = 0 # получили полигон после grid-walk + parcels_marked_unavailable: int = 0 # geom так и не нашёлся → флаг + grid_walk_requests: int = 0 + + +async def backfill_parcel_geom( + db: Session, + client: NSPDBulkClient, + *, + limit: int = 500, + update_heartbeat: Callable[[dict[str, Any]], None] | None = None, +) -> GeomBackfillResult: + """Докачать geom для участков cad_parcels с geom IS NULL (issue #200). + + Алгоритм: + 1. Выбрать до `limit` участков `geom IS NULL AND NOT geom_unavailable`. + 2. Сгруппировать по кварталу (3-сегментный cad). NSPD search-эндпоинт + (Phase 1) отдаёт эти участки без полигона; реальную геометрию + возвращает grid-walk слоя 36368 (WMS GetFeatureInfo по bbox квартала). + 3. Для каждого квартала вызвать _grid_walk_category(36368) — он upsert'ит + полигоны через ON CONFLICT ... geom = COALESCE(EXCLUDED.geom, ...). + 4. Участки, которые ПОСЛЕ grid-walk всё ещё geom IS NULL, пометить + geom_unavailable=TRUE — geom систематически недоступен (кросс-региональные + артефакты поиска без полигона в слое), чтобы не ретраить вечно. + + Идемпотентен: повторный вызов берёт следующую порцию (healed уходят из + выборки — у них geom уже не NULL; unavailable исключены флагом). + + Args: + limit: максимум участков за один прогон (батч против WAF-burst). + update_heartbeat: callback для heartbeat (как в harvest_quarter) — задача + может идти минуты при многих кварталах. + + Returns: + GeomBackfillResult со счётчиками. + """ + result = GeomBackfillResult() + + rows = ( + db.execute( + text( + "SELECT cad_num FROM cad_parcels " + "WHERE geom IS NULL AND geom_unavailable = FALSE " + "ORDER BY cad_num LIMIT :lim" + ), + {"lim": limit}, + ) + .scalars() + .all() + ) + if not rows: + logger.info("backfill_parcel_geom: нет участков с geom IS NULL — нечего докачивать") + return result + + result.parcels_targeted = len(rows) + + # Группируем по кварталу (первые 3 colon-сегмента). Участки без валидного + # 3-сегментного префикса (мусорные cad) — сразу в unavailable, grid-walk + # для них невозможен (нет квартала → нет bbox). + by_quarter: dict[str, list[str]] = {} + orphan_cads: list[str] = [] + for cad in rows: + parts = cad.split(":") + if len(parts) >= 3 and all(parts[:3]): + by_quarter.setdefault(":".join(parts[:3]), []).append(cad) + else: + orphan_cads.append(cad) + + if orphan_cads: + result.parcels_marked_unavailable += _mark_geom_unavailable(db, orphan_cads) + db.commit() + + for quarter, cads in by_quarter.items(): + result.quarters_scanned += 1 + if update_heartbeat is not None: + update_heartbeat( + { + "phase": "geom_backfill", + "quarter": quarter, + "quarters_scanned": result.quarters_scanned, + "quarters_total": len(by_quarter), + } + ) + + try: + _discovered, n_requests, _failed = await _grid_walk_category( + db=db, + client=client, + quarter=quarter, + layer_id=CAT_PARCEL, + update_progress=update_heartbeat, + ) + result.grid_walk_requests += n_requests + db.commit() + except Exception as e: + # Один сбойный квартал не валит весь backfill — лог + продолжаем. + # (WAF 403 пробросится из client и прервёт прогон — это ожидаемо, + # caller-task ловит и не ретраит, как в bulk_harvest.) + logger.warning( + "backfill_parcel_geom: grid-walk failed quarter=%s: %s", quarter, e + ) + db.rollback() + continue + + # Какие из целевых участков квартала всё ещё без geom? + still_null = ( + db.execute( + text( + "SELECT cad_num FROM cad_parcels " + "WHERE cad_num = ANY(CAST(:cads AS text[])) AND geom IS NULL" + ), + {"cads": cads}, + ) + .scalars() + .all() + ) + healed_here = len(cads) - len(still_null) + result.parcels_healed += healed_here + + if still_null: + result.parcels_marked_unavailable += _mark_geom_unavailable(db, list(still_null)) + db.commit() + + logger.info( + "backfill_parcel_geom: quarter=%s targeted=%d healed=%d unavailable=%d", + quarter, + len(cads), + healed_here, + len(still_null), + ) + + logger.info( + "backfill_parcel_geom done: quarters=%d targeted=%d healed=%d unavailable=%d requests=%d", + result.quarters_scanned, + result.parcels_targeted, + result.parcels_healed, + result.parcels_marked_unavailable, + result.grid_walk_requests, + ) + return result + + +def _mark_geom_unavailable(db: Session, cad_nums: list[str]) -> int: + """Пометить участки geom_unavailable=TRUE (issue #200). Возвращает rowcount. + + Параметризованный UPDATE по списку cad_num (ANY(text[])) — не f-string SQL. + """ + if not cad_nums: + return 0 + res = db.execute( + text( + "UPDATE cad_parcels SET geom_unavailable = TRUE, updated_at = NOW() " + "WHERE cad_num = ANY(CAST(:cads AS text[])) " + "AND geom IS NULL AND geom_unavailable = FALSE" + ), + {"cads": cad_nums}, + ) + return res.rowcount or 0 # ── Routing dispatcher ─────────────────────────────────────────────────────── diff --git a/backend/app/workers/beat_schedule.py b/backend/app/workers/beat_schedule.py index bd3cda19..defdc47c 100644 --- a/backend/app/workers/beat_schedule.py +++ b/backend/app/workers/beat_schedule.py @@ -277,17 +277,23 @@ def build_beat_schedule() -> dict: "options": {"queue": "celery"}, } - # ПЗЗ территориальных зон ЕКБ из PKK6 ArcGIS — ежемесячно 1-го числа в 03:00 МСК. - # Celery conf.timezone=Europe/Moscow → crontab трактуется в МСК (#1233). - # PKK6 нестабилен под нагрузкой, но данные ПЗЗ меняются редко — раз в месяц достаточно. - # Task: tasks/pzz_sync.py → sync_pzz_zones_ekb. - # Admin trigger: POST /api/v1/admin/scrape/pzz-sync. - # Ref: issue #233 (pzz_zones_ekb = 0 rows, задача никогда не запускалась автоматически). - schedule["pzz-sync-monthly"] = { - "task": "tasks.pzz_sync.sync_pzz_zones_ekb", - "schedule": _parse_cron("0 3 1 * *"), - "options": {"queue": "celery"}, - } + # ПЗЗ территориальных зон ЕКБ из PKK6 ArcGIS — DISABLED 2026-06-13 (issue #259). + # + # ПОЧЕМУ ОТКЛЮЧЕНО: источник pkk.rosreestr.ru/.../PKK6/ZONES deprecated (Росреестр + # вывел PKK6-эндпоинт), таблица pzz_zones_ekb пуста (0 rows на prod) — задача ни + # разу не наполнила её успешно. ПЗЗ-данные пришли в систему ИНЫМ путём: + # zone_regulation_cache (#1059, beat zone-regulation-refresh-monthly) + NSPD + # territorial_zones dumps (Phase 2.5 bulk_harvest, layer 875838). Активный beat + # каждый месяц дёргал deprecated PKK6 → broad except в pzz_sync.py логировал + # ошибку как error → рекуррентил GlitchTip BACKEND-1B. + # + # Ручной перезапуск (если появится рабочий источник) остаётся через admin-эндпоинт + # POST /api/v1/admin/scrape/pzz-sync. Ref: #233 (исходно «0 rows»), #259 (disable). + # schedule["pzz-sync-monthly"] = { + # "task": "tasks.pzz_sync.sync_pzz_zones_ekb", + # "schedule": _parse_cron("0 3 1 * *"), + # "options": {"queue": "celery"}, + # } # Catalog-object scrape — наполняет ~25 NULL колонок domrf_kn_objects из SSR-страниц. # kn-API не отдаёт wall_type, energy_eff, ceiling_height_m, parking_* и т.д. diff --git a/backend/app/workers/tasks/pzz_sync.py b/backend/app/workers/tasks/pzz_sync.py index 83f01aa7..d29e2dc6 100644 --- a/backend/app/workers/tasks/pzz_sync.py +++ b/backend/app/workers/tasks/pzz_sync.py @@ -34,6 +34,12 @@ def _log_breadcrumb(level: str, message: str) -> None: def sync_pzz_zones_ekb() -> dict: """Импорт ПЗЗ территориальных зон ЕКБ из Росреестр PKK6. + DEPRECATED (issue #259): источник PKK6 выведен Росреестром, beat-расписание + pzz-sync-monthly отключено (beat_schedule.py), таблица pzz_zones_ekb пуста. + ПЗЗ-данные теперь поступают через zone_regulation_cache + NSPD territorial_zones + dumps. Задача оставлена только для ручного запуска через admin (на случай + появления рабочего источника). + Запускается вручную через POST /api/v1/admin/scrape/pzz-sync. Persistent breadcrumbs в nspd_geo_log (stage='pzz_sync') для диагностики. """ @@ -46,6 +52,10 @@ def sync_pzz_zones_ekb() -> dict: _log_breadcrumb("info", f"done: {result}") return result except Exception as e: - logger.exception("pzz_sync failed: %s", e) - _log_breadcrumb("error", f"{type(e).__name__}: {e}") + # Issue #259: PKK6 deprecated → этот путь почти всегда падает. logger.warning + # (не .exception) — Sentry/GlitchTip НЕ захватывает warning как error-event, + # поэтому ручной запуск по дохлому источнику не рекуррентит BACKEND-1B. + # re-raise сохранён: admin-эндпоинт обязан увидеть фактический провал импорта. + logger.warning("pzz_sync failed (PKK6 deprecated, см. #259): %s: %s", type(e).__name__, e) + _log_breadcrumb("warning", f"{type(e).__name__}: {e}") raise diff --git a/backend/app/workers/tasks/scrape_cadastre.py b/backend/app/workers/tasks/scrape_cadastre.py index 9057f776..70b6fcea 100644 --- a/backend/app/workers/tasks/scrape_cadastre.py +++ b/backend/app/workers/tasks/scrape_cadastre.py @@ -24,7 +24,7 @@ from typing import Any from sqlalchemy import text from app.core.db import SessionLocal -from app.scrapers.nspd_bulk_client import NspdBulkWafError +from app.scrapers.nspd_bulk_client import NspdBulkRateLimitError, NspdBulkWafError from app.workers.celery_app import celery_app logger = logging.getLogger(__name__) @@ -132,6 +132,12 @@ def _is_job_cancelled(db: Any, job_id: int) -> bool: name="tasks.cadastre.harvest_quarter", acks_late=True, max_retries=3, + # autoretry_for=(Exception,) ловит ВСЁ кроме явных исключений ниже. Issue #252: + # квартал-уровневый NspdBulkServerError (HTTP 500 / WMS ServiceException на + # Phase 1 search_by_quarter) — подкласс Exception и НЕ в dont_autoretry_for, + # значит ретраится с backoff (GlitchTip BACKEND-8/10/11/.../37). Per-cell 500 + # внутри grid-walk квартал НЕ валят — слой пропускается с harvest_meta-флагом + # (см. bulk_harvest._grid_walk_category). autoretry_for=(Exception,), retry_backoff=True, retry_backoff_max=60, @@ -211,6 +217,9 @@ def bulk_harvest_quarter_task(self: Any, quarter: str, job_id: int) -> dict[str, "snapshot_requests": result.snapshot_requests, "grid_walk_requests": result.grid_walk_requests, "total_requests": result.total_requests, + # Issue #252: слои, чей grid-walk целиком сбойнул на 500 — для + # observability в результате task (квартал при этом успешен). + "failed_layers": result.failed_layers, } result_dict = asyncio.run(_run()) @@ -472,3 +481,74 @@ def cleanup_cadastre_zombies() -> dict[str, Any]: finally: db.close() return {"stale_paused": stale_ids} + + +# ── Geom backfill (issue #200) ─────────────────────────────────────────────── + + +@celery_app.task( + bind=True, + name="tasks.cadastre.backfill_parcel_geom", + acks_late=True, + max_retries=2, + # WAF/429 — не наш баг, не ретраим (как bulk_harvest_quarter_task). Прочие + # ошибки докачки тоже не ретраим автоматически: backfill идемпотентен и + # запускается вручную/по расписанию, повторный прогон возьмёт остаток. + autoretry_for=(), + soft_time_limit=900, + time_limit=1200, +) +def backfill_parcel_geom_task(self: Any, limit: int = 500) -> dict[str, Any]: + """Докачать geom для участков cad_parcels с geom IS NULL (issue #200). + + Sync→async bridge через asyncio.run() (как bulk_harvest_quarter_task). + Один прогон обрабатывает до `limit` участков; idempotent — повторный вызов + берёт следующую порцию (вылеченные уходят из выборки, систематически + недоступные исключаются флагом geom_unavailable). + + Запуск вручную (CLI/admin) или через beat, если нужно периодически дочищать + хвост NULL-geom после bulk-harvest. Возвращает счётчики GeomBackfillResult. + """ + from app.services.cadastre.bulk_harvest import backfill_parcel_geom + + db = SessionLocal() + try: + + async def _run() -> dict[str, Any]: + from app.scrapers.nspd_bulk_client import NSPDBulkClient + + async with NSPDBulkClient() as client: + res = await backfill_parcel_geom(db=db, client=client, limit=limit) + return { + "quarters_scanned": res.quarters_scanned, + "parcels_targeted": res.parcels_targeted, + "parcels_healed": res.parcels_healed, + "parcels_marked_unavailable": res.parcels_marked_unavailable, + "grid_walk_requests": res.grid_walk_requests, + } + + result_dict = asyncio.run(_run()) + logger.info("backfill_parcel_geom_task done: %s", result_dict) + return result_dict + + except (NspdBulkWafError, NspdBulkRateLimitError) as e: + # WAF 403 / исчерпанный rate-limit — прерываем прогон без retry. Уже + # вылеченные участки закоммичены поквартально внутри backfill_parcel_geom, + # следующий ручной/плановый запуск добьёт остаток. + logger.warning("backfill_parcel_geom_task aborted (WAF/rate-limit): %s", e) + try: + db.rollback() + except Exception: + pass + return {"aborted": "waf_or_rate_limit", "error": str(e)[:300]} + + except Exception as e: + logger.exception("backfill_parcel_geom_task failed: %s", e) + try: + db.rollback() + except Exception: + pass + raise + + finally: + db.close() diff --git a/backend/tests/scrapers/test_nspd_bulk_client.py b/backend/tests/scrapers/test_nspd_bulk_client.py index 7add40f5..b98bb7e9 100644 --- a/backend/tests/scrapers/test_nspd_bulk_client.py +++ b/backend/tests/scrapers/test_nspd_bulk_client.py @@ -410,6 +410,75 @@ async def test_http_429_backoff_exhausted() -> None: assert call_count == 4 +# ── Issue #252: 5xx / WMS ServiceException → NspdBulkServerError ────────────── + + +@pytest.mark.asyncio +async def test_http_500_raises_server_error() -> None: + """500 → NspdBulkServerError (подкласс NspdBulkError) — для retry + skip.""" + import httpx + + from app.scrapers.nspd_bulk_client import NspdBulkError, NspdBulkServerError + + async def fake_get(*args: Any, **kwargs: Any) -> httpx.Response: + return httpx.Response(500, text="Internal Server Error", request=MagicMock()) + + async with NSPDBulkClient() as client: + assert client._client is not None + with patch.object(client._client, "get", new=AsyncMock(side_effect=fake_get)): + with pytest.raises(NspdBulkServerError) as exc_info: + await client._get_json("https://nspd.gov.ru/test") + + # Подкласс NspdBulkError → autoretry_for=(Exception,) его поймает, а + # except NspdBulkError-ветки (404/400) тоже сработают если понадобится. + assert isinstance(exc_info.value, NspdBulkError) + + +@pytest.mark.asyncio +async def test_wms_service_exception_in_200_body_raises_server_error() -> None: + """HTTP 200 с XML в теле → NspdBulkServerError (не JSONDecodeError).""" + import httpx + + from app.scrapers.nspd_bulk_client import NspdBulkServerError + + xml_body = ( + '' + "Rendering error" + ) + + async def fake_get(*args: Any, **kwargs: Any) -> httpx.Response: + return httpx.Response( + 200, text=xml_body, headers={"content-type": "application/xml"}, request=MagicMock() + ) + + async with NSPDBulkClient() as client: + assert client._client is not None + with patch.object(client._client, "get", new=AsyncMock(side_effect=fake_get)): + with pytest.raises(NspdBulkServerError): + await client._get_json("https://nspd.gov.ru/wms") + + +@pytest.mark.asyncio +async def test_wms_feature_info_propagates_server_error( + sample_wms_response: dict[str, Any], +) -> None: + """wms_feature_info пробрасывает NspdBulkServerError (caller считает fail-rate).""" + from app.scrapers.nspd_bulk_client import NspdBulkServerError + + async with NSPDBulkClient() as client: + with patch.object( + client, + "_get_json", + new=AsyncMock(side_effect=NspdBulkServerError("HTTP 500 server error")), + ): + with pytest.raises(NspdBulkServerError): + await client.wms_feature_info( + layer_id=36368, + bbox=(6700000.0, 7700000.0, 6701000.0, 7701000.0), + click_xy=(256, 256), + ) + + # ── E2E tests (network, skipped by default) ─────────────────────────────────── pytestmark_slow = pytest.mark.slow diff --git a/backend/tests/services/test_cadastre_bulk.py b/backend/tests/services/test_cadastre_bulk.py index 598f5a7c..bea84372 100644 --- a/backend/tests/services/test_cadastre_bulk.py +++ b/backend/tests/services/test_cadastre_bulk.py @@ -673,7 +673,7 @@ async def test_grid_walk_dedup_same_cad_num() -> None: } # Используем grid_size=2 для скорости теста (4 ячейки) - upserted, requests = await _grid_walk_category( + upserted, requests, layer_failed = await _grid_walk_category( db=db, client=client, quarter="66:41:0303161", @@ -685,6 +685,7 @@ async def test_grid_walk_dedup_same_cad_num() -> None: assert mock_upsert.call_count == 1 assert upserted == 1 assert requests == 4 # 2×2 grid + assert layer_failed is False # все cells успешны # ── ЗОУИТ dedup по (reg_numb_border, category_id) ─────────────────────────── @@ -1306,7 +1307,7 @@ async def test_grid_walk_emits_heartbeat_callbacks() -> None: progress_states: list[dict[str, Any]] = [] - _upserted, requests = await _grid_walk_category( + _upserted, requests, _failed = await _grid_walk_category( db=db, client=client, quarter="66:41:0303161", @@ -1341,10 +1342,297 @@ async def test_grid_walk_no_heartbeat_when_callback_none() -> None: client.wms_feature_info = AsyncMock(return_value=[]) # Не передаём update_progress — должно не падать - _upserted, requests = await _grid_walk_category( + _upserted, requests, _failed = await _grid_walk_category( db=db, client=client, quarter="66:41:0303161", layer_id=36368, ) assert requests == 225 + + +# ── Issue #252: per-layer skip при 5xx/ServiceException ────────────────────── + + +def _mock_db_grid_bbox() -> MagicMock: + """Mock db с валидным bbox для quarter_bbox_3857 + begin_nested CM.""" + db = MagicMock() + db.execute.return_value.mappings.return_value.first.return_value = { + "xmin": 6735845.0, + "ymin": 8329000.0, + "xmax": 6736595.0, + "ymax": 8329750.0, + } + db.begin_nested = MagicMock( + return_value=MagicMock(__enter__=MagicMock(return_value=None), __exit__=MagicMock()) + ) + db.commit = MagicMock() + return db + + +@pytest.mark.asyncio +async def test_grid_walk_marks_layer_failed_when_all_cells_500() -> None: + """Issue #252: ВСЕ cells слоя упали на NspdBulkServerError → layer_failed=True.""" + from app.scrapers.nspd_bulk_client import NspdBulkServerError + from app.services.cadastre.bulk_harvest import _grid_walk_category + + db = _mock_db_grid_bbox() + client = AsyncMock() + client.wms_feature_info = AsyncMock(side_effect=NspdBulkServerError("HTTP 500")) + + upserted, requests, layer_failed = await _grid_walk_category( + db=db, + client=client, + quarter="66:41:0303161", + layer_id=36368, + grid_size=3, # 9 cells, все упадут + ) + + assert upserted == 0 + assert requests == 9 + assert layer_failed is True + + +@pytest.mark.asyncio +async def test_grid_walk_layer_not_failed_when_some_cells_ok() -> None: + """Issue #252: если хоть один cell прошёл — layer_failed=False (слой жив, просто пуст).""" + from app.scrapers.nspd_bulk_client import NspdBulkServerError + from app.services.cadastre.bulk_harvest import _grid_walk_category + + db = _mock_db_grid_bbox() + client = AsyncMock() + # Первый cell ок (пустой ответ), остальные 500 + client.wms_feature_info = AsyncMock( + side_effect=[[], NspdBulkServerError("500"), NspdBulkServerError("500")] + ) + + _upserted, requests, layer_failed = await _grid_walk_category( + db=db, + client=client, + quarter="66:41:0303161", + layer_id=36368, + grid_size=1, # уйдёт по side_effect — но grid_size=1 → 1 cell + ) + + # grid_size=1 → 1 cell (ок) → layer жив + assert layer_failed is False + assert requests == 1 + + +@pytest.mark.asyncio +async def test_grid_walk_network_error_does_not_trigger_layer_failed() -> None: + """Issue #252: сетевая (не server-side) ошибка cell НЕ учитывается в fail-rate.""" + from app.scrapers.nspd_bulk_client import NspdBulkError + from app.services.cadastre.bulk_harvest import _grid_walk_category + + db = _mock_db_grid_bbox() + client = AsyncMock() + # NspdBulkError (не Server) — например network. layer_failed требует server_errors>0. + client.wms_feature_info = AsyncMock(side_effect=NspdBulkError("Network error")) + + _upserted, requests, layer_failed = await _grid_walk_category( + db=db, + client=client, + quarter="66:41:0303161", + layer_id=36368, + grid_size=3, + ) + + # server_errors==0 (это не 5xx) → layer_failed=False, квартал не получит skip-флаг + assert layer_failed is False + assert requests == 9 + + +@pytest.mark.asyncio +async def test_harvest_quarter_records_failed_layer_in_phase_state() -> None: + """Issue #252: сбойный слой grid-walk → harvest_meta.layer_X_failed в done-state.""" + from app.services.cadastre.bulk_harvest import CAT_PARCEL, harvest_quarter + + parcel = _make_parcel_feature() + snapshot = QuarterSnapshot( + quarter="66:41:0303161", + fetched_at="2026-05-15T10:00:00+00:00", + features=[parcel], + meta_counts={CAT_PARCEL: 170}, # overflow → grid-walk слоя 36368 + ) + + db = _mock_db_grid_bbox() + client = AsyncMock() + client.search_by_quarter = AsyncMock(return_value=snapshot) + client.get_territorial_zones_in_bbox = AsyncMock(return_value=[]) + + progress_states: list[dict[str, Any]] = [] + + with ( + patch("app.services.cadastre.bulk_harvest.upsert_features") as mock_upsert, + patch("app.services.cadastre.bulk_harvest._grid_walk_category") as mock_walk, + patch("app.services.cadastre.bulk_harvest.quarter_bbox_3857") as mock_bbox, + ): + mock_bbox.return_value = (6735845.0, 8329000.0, 6736595.0, 8329750.0) + mock_upsert.return_value = { + "parcels": 1, + "buildings": 0, + "constructions": 0, + "oncs": 0, + "enks": 0, + "zouit": 0, + "skipped": 0, + } + # grid-walk слоя CAT_PARCEL вернул layer_failed=True + mock_walk.return_value = (0, 9, True) + + result = await harvest_quarter( + db=db, + client=client, + quarter="66:41:0303161", + job_id=1, + update_progress=lambda s: progress_states.append(s), + ) + + # failed_layers содержит CAT_PARCEL + assert result.failed_layers == [CAT_PARCEL] + # Финальный done-state несёт harvest_meta-флаг + assert result.phase_state is not None + assert result.phase_state["harvest_meta"] == {f"layer_{CAT_PARCEL}_failed": True} + # Квартал НЕ свалился — дошёл до done + assert progress_states[-1]["phase"] == "done" + + +# ── Issue #200: geom backfill для участков с geom IS NULL ───────────────────── + + +def test_mark_geom_unavailable_parametrized_no_fstring() -> None: + """_mark_geom_unavailable использует ANY(CAST(:cads AS text[])) — не f-string SQL.""" + from app.services.cadastre.bulk_harvest import _mark_geom_unavailable + + executed: list[tuple[str, Any]] = [] + + db = MagicMock() + + def capture(stmt: Any, params: Any = None) -> MagicMock: + executed.append((str(stmt), params)) + return MagicMock(rowcount=2) + + db.execute = capture + + n = _mark_geom_unavailable(db, ["66:41:0303161:1", "66:41:0303161:2"]) + + assert n == 2 + assert len(executed) == 1 + sql, params = executed[0] + assert "ANY(CAST(:cads AS text[]))" in sql + assert "geom_unavailable = TRUE" in sql + assert params["cads"] == ["66:41:0303161:1", "66:41:0303161:2"] + + +def test_mark_geom_unavailable_empty_noop() -> None: + """Пустой список → 0, без db.execute.""" + from app.services.cadastre.bulk_harvest import _mark_geom_unavailable + + db = MagicMock() + assert _mark_geom_unavailable(db, []) == 0 + db.execute.assert_not_called() + + +@pytest.mark.asyncio +async def test_backfill_parcel_geom_heals_and_marks_unavailable() -> None: + """Issue #200: участок с восстановленным geom → healed; оставшийся NULL → unavailable.""" + from app.services.cadastre.bulk_harvest import backfill_parcel_geom + + # Очередь: 2 участка одного квартала. + null_rows = ["66:41:0303161:1", "66:41:0303161:2"] + # После grid-walk участок :2 всё ещё NULL → должен быть помечен unavailable. + still_null = ["66:41:0303161:2"] + + db = MagicMock() + captured: list[dict[str, Any]] = [] + + def execute_router(stmt: Any, params: Any = None) -> MagicMock: + sql = str(stmt) + result = MagicMock() + if "geom IS NULL AND geom_unavailable = FALSE" in sql and "LIMIT" in sql: + result.scalars.return_value.all.return_value = null_rows + elif "AND geom IS NULL" in sql and "ANY(CAST(:cads AS text[]))" in sql and "SELECT" in sql: + result.scalars.return_value.all.return_value = still_null + elif "UPDATE cad_parcels SET geom_unavailable = TRUE" in sql: + captured.append({"sql": sql, "params": params or {}}) + result.rowcount = len(still_null) + else: + result.scalars.return_value.all.return_value = [] + result.rowcount = 0 + return result + + db.execute = execute_router + db.commit = MagicMock() + + client = AsyncMock() + + with patch("app.services.cadastre.bulk_harvest._grid_walk_category") as mock_walk: + # grid-walk слоя 36368 «нашёл» 1 полигон, без layer_failed + mock_walk.return_value = (1, 9, False) + result = await backfill_parcel_geom(db=db, client=client, limit=500) + + assert result.parcels_targeted == 2 + assert result.quarters_scanned == 1 + assert result.parcels_healed == 1 # :1 получил geom + assert result.parcels_marked_unavailable == 1 # :2 помечен + # grid-walk вызван для квартала с layer_id=36368 + mock_walk.assert_called_once() + assert mock_walk.call_args.kwargs["quarter"] == "66:41:0303161" + assert mock_walk.call_args.kwargs["layer_id"] == 36368 + # UPDATE geom_unavailable выполнен ровно для оставшегося NULL + assert len(captured) == 1 + assert captured[0]["params"]["cads"] == still_null + + +@pytest.mark.asyncio +async def test_backfill_parcel_geom_empty_queue_noop() -> None: + """Issue #200: нет участков с geom IS NULL → пустой результат, grid-walk не вызывается.""" + from app.services.cadastre.bulk_harvest import backfill_parcel_geom + + db = MagicMock() + db.execute.return_value.scalars.return_value.all.return_value = [] + client = AsyncMock() + + with patch("app.services.cadastre.bulk_harvest._grid_walk_category") as mock_walk: + result = await backfill_parcel_geom(db=db, client=client, limit=500) + + assert result.parcels_targeted == 0 + assert result.quarters_scanned == 0 + mock_walk.assert_not_called() + + +@pytest.mark.asyncio +async def test_backfill_parcel_geom_marks_orphan_cads_unavailable() -> None: + """Issue #200: cad без валидного 3-сегментного квартала → сразу unavailable (нет bbox).""" + from app.services.cadastre.bulk_harvest import backfill_parcel_geom + + orphan_rows = ["66:41", "garbage"] # <3 сегментов → grid-walk невозможен + + db = MagicMock() + marked: list[Any] = [] + + def execute_router(stmt: Any, params: Any = None) -> MagicMock: + sql = str(stmt) + result = MagicMock() + if "geom IS NULL AND geom_unavailable = FALSE" in sql and "LIMIT" in sql: + result.scalars.return_value.all.return_value = orphan_rows + elif "UPDATE cad_parcels SET geom_unavailable = TRUE" in sql: + marked.append(params) + result.rowcount = len(params["cads"]) + else: + result.scalars.return_value.all.return_value = [] + result.rowcount = 0 + return result + + db.execute = execute_router + db.commit = MagicMock() + client = AsyncMock() + + with patch("app.services.cadastre.bulk_harvest._grid_walk_category") as mock_walk: + result = await backfill_parcel_geom(db=db, client=client, limit=500) + + # Оба orphan помечены unavailable, grid-walk не вызывался + mock_walk.assert_not_called() + assert result.parcels_marked_unavailable == 2 + assert marked and sorted(marked[0]["cads"]) == ["66:41", "garbage"] diff --git a/data/sql/151_cad_parcels_geom_unavailable.sql b/data/sql/151_cad_parcels_geom_unavailable.sql new file mode 100644 index 00000000..4d6f180e --- /dev/null +++ b/data/sql/151_cad_parcels_geom_unavailable.sql @@ -0,0 +1,51 @@ +-- 151_cad_parcels_geom_unavailable.sql +-- --------------------------------------------------------------------------- +-- Контекст (issue #200): +-- На prod ~964 строки cad_parcels с geom IS NULL. Ground-truth (измерено +-- 2026-06-13): 930/964 — реальные участки ЕКБ 66:41:NNNNNN:NN, все +-- source='search'. NSPD search-эндпоинт для квартала возвращает участок без +-- полигона (только центроид/Point ИЛИ вовсе без geometry → upsert_parcel +-- пишет geom=NULL, см. bulk_harvest.py). Остаток (34) — кросс-региональные +-- артефакты поиска (последний сегмент :41 / :66 — это номер участка из +-- ДРУГОГО квартала, попавший в выдачу), у которых полигона в WMS-слое +-- 36368 нет вовсе. +-- +-- Полигон для большинства восстановим grid-walk'ом слоя 36368 (WMS +-- GetFeatureInfo по bbox квартала — Phase 2-3 bulk_harvest). Но для +-- систематически недоступных (кросс-региональные артефакты) бесконечный +-- ретрай вреден: каждый прогон бэкфилла снова бы их дёргал. +-- +-- Действие: +-- Добавить флаг geom_unavailable BOOLEAN. Backfill-task +-- (tasks.cadastre.backfill_parcel_geom) выставляет его в TRUE, когда после +-- grid-walk слоя участок ВСЁ ЕЩЁ без полигона — значит geom систематически +-- недоступен, перестаём ретраить. Дефолт FALSE — обычные участки и новые +-- inserts не затрагиваются; пересчёт делает только бэкфилл. +-- +-- Partial-индекс под выборку «кого ещё докачивать» +-- (geom IS NULL AND NOT geom_unavailable) — узкий, индексирует только +-- рабочую очередь, не всю таблицу (40k строк). +-- +-- Идемпотентность: ADD COLUMN IF NOT EXISTS + CREATE INDEX IF NOT EXISTS. +-- Apply order: чистая схема, без зависимости от кода. Backend-код с +-- backfill-task деплоится после (правило sql.md: схема → код). +-- --------------------------------------------------------------------------- + +BEGIN; + +ALTER TABLE cad_parcels + ADD COLUMN IF NOT EXISTS geom_unavailable BOOLEAN NOT NULL DEFAULT FALSE; + +COMMENT ON COLUMN cad_parcels.geom_unavailable IS + 'issue #200: TRUE — geom систематически недоступен в НСПД (grid-walk слоя ' + '36368 не вернул полигон). Выставляется backfill-task' + ' (tasks.cadastre.backfill_parcel_geom), чтобы не ретраить докачку вечно. ' + 'FALSE (дефолт) — geom либо есть, либо ещё не пробовали докачать.'; + +-- Рабочая очередь бэкфилла: участки без геометрии, которые ещё имеет смысл +-- докачивать. Partial WHERE держит индекс компактным. +CREATE INDEX IF NOT EXISTS cad_parcels_geom_backfill_queue + ON cad_parcels (cad_num) + WHERE geom IS NULL AND geom_unavailable = FALSE; + +COMMIT;