"""Оркестратор bulk cadastre harvest — pure-async, без Celery декораторов. Вызывается из Celery-таска (scrape_cadastre.py) через asyncio.run(). Фазы harvest_quarter: Phase 1: search_by_quarter → upsert первые 20 per категории Phase 2: определить overflow-категории (meta > 20) Phase 3: grid_walk для overflow 36368 (ЗУ) и 36369 (Здания) Phase 4: upsert_quarter_stats из meta квартального snapshot (36381) Idempotent: все upsert'ы — ON CONFLICT DO UPDATE. Resumable: phase_state в cadastre_jobs показывает прогресс. Питфоллы (см. CLAUDE.md): - CAST(:x AS jsonb) не :x::jsonb (SQLAlchemy парсит :: как параметр) - psycopg v3 only (не psycopg2) - SAVEPOINT через db.begin_nested() вокруг каждого upsert """ from __future__ import annotations import json import logging from collections.abc import Callable from dataclasses import dataclass, field from typing import Any from sqlalchemy import text from sqlalchemy.orm import Session from app.schemas.nspd_bulk import NSPDBulkFeature, QuarterSnapshot from app.scrapers.nspd_bulk_client import ( NSPDBulkClient, NspdBulkRateLimitError, NspdBulkServerError, NspdBulkWafError, ) from app.services.cadastre.grid_geometry import generate_grid_click_points, quarter_bbox_3857 logger = logging.getLogger(__name__) # Категории NSPD для routing CAT_PARCEL = 36368 # ЗУ ЕГРН CAT_BUILDING = 36369 # Здания ЕГРН CAT_CONSTRUCTION = 36383 # Сооружения CAT_ONC = 36384 # ОНС CAT_ENK = 39663 # ЕНК CAT_QUARTER_STATS = 36381 # Квартальная статистика CAT_ZOUIT = frozenset({36940, 469039, 469040, 469042}) # ЗОУИТ # Категории для grid-walk overflow GRID_WALK_CATS = frozenset({CAT_PARCEL, CAT_BUILDING}) # Категории с выделенным thematicSearchId в NSPD API (curl-probe 2026-05-15). # Для них отдельный per-category вызов даёт ДРУГОЙ результат, чем базовый thematic=1: # - ЗОУИТ: thematicSearchId=5 → до 20 per subtype × 4 subtypes # - ЕНК: thematicSearchId=15 → до 20 ЕНК объектов # ЗУ/Здания/Сооружения/ОНС не имеют dedicated thematicSearchId — отдельный вызов # вернёт те же 20 что и базовый, поэтому per-category probe для них бессмыслен. PER_CATEGORY_PROBE_CATS = frozenset({*CAT_ZOUIT, CAT_ENK}) # ── Result type ────────────────────────────────────────────────────────────── @dataclass class HarvestResult: """Итог harvest одного квартала.""" quarter: str parcels_upserted: int = 0 buildings_upserted: int = 0 constructions_upserted: int = 0 oncs_upserted: int = 0 enks_upserted: int = 0 zouit_upserted: int = 0 snapshot_requests: int = 0 # search_by_quarter (Phase 1) + per-cat probes (Phase 1.5) grid_walk_requests: int = 0 # wms_feature_info (Phase 2-3) # Issue #252: layer_id'ы, чей grid-walk целиком упал на server-side 500/ # ServiceException (все cells сбойнули). Квартал НЕ валится — слой # пропускается, факт фиксируется в cadastre_jobs.phase_state.harvest_meta. failed_layers: list[int] = field(default_factory=list) phase_state: dict[str, Any] | None = None @property def total_requests(self) -> int: """Все NSPD HTTP requests для этого квартала (snapshot + grid-walk).""" return self.snapshot_requests + self.grid_walk_requests # ── Основной оркестратор ───────────────────────────────────────────────────── async def harvest_quarter( db: Session, client: NSPDBulkClient, quarter: str, job_id: int, update_progress: Callable[[dict[str, Any]], None], ) -> HarvestResult: """Собрать все NSPD данные для квартала в cad_* таблицы. Идемпотентен: повторный вызов не создаёт дубликаты (ON CONFLICT DO UPDATE). Возобновляем: проверяет job.phase_state, пропускает завершённые фазы. Args: db: SQLAlchemy session. Caller должен управлять commit/rollback. client: NSPDBulkClient (async context manager, уже открытый). quarter: кадастровый номер квартала. job_id: ID записи в cadastre_jobs для обновления прогресса. update_progress: callback для записи phase_state в cadastre_jobs. Returns: HarvestResult с счётчиками upserted объектов. Raises: NspdBulkWafError: при WAF 403 — caller не retry. NspdBulkRateLimitError: при исчерпании retries — caller может retry. Exception: прочие ошибки — propagate. """ result = HarvestResult(quarter=quarter) # NB: `cadastre_jobs.phase_state` шарится между ВСЕМИ parallel Celery workers одного job. # Раньше тут был early-exit `if phase_state.phase == 'done': return` — но он триггерился # как только ПЕРВЫЙ quarter записывал phase=done через update_progress callback, и все # остальные workers (читая ту же shared row) бейлились без работы. Эффект: 95% quarters # early-exit'или, coverage 1.6% parcels (pilot v7: 5 NSPD req для 50 quarters). # Idempotency обеспечивается ON CONFLICT DO UPDATE в каждом upsert_* — re-enqueued task # просто перепишет те же rows, не создавая дубликатов. # ── Phase 1: search_by_quarter snapshot ────────────────────────────────── update_progress({"phase": "snapshot_started", "quarter": quarter}) snapshot: QuarterSnapshot = await client.search_by_quarter(quarter) result.snapshot_requests += 1 snapshot_stats = upsert_features(db, snapshot.features, source="search") result.parcels_upserted += snapshot_stats["parcels"] result.buildings_upserted += snapshot_stats["buildings"] result.constructions_upserted += snapshot_stats["constructions"] result.oncs_upserted += snapshot_stats["oncs"] result.enks_upserted += snapshot_stats["enks"] result.zouit_upserted += snapshot_stats["zouit"] db.commit() parcels_meta = snapshot.meta_counts.get(CAT_PARCEL, 0) buildings_meta = snapshot.meta_counts.get(CAT_BUILDING, 0) update_progress( { "phase": "snapshot_done", "quarter": quarter, "parcels_total_meta": parcels_meta, "buildings_total_meta": buildings_meta, "overflow_cats": snapshot.overflow_categories, } ) # ── Phase 1.5: per-category snapshot для ЗОУИТ и ЕНК ──────────────────── # ЗОУИТ (36940/469039/469040/469042) и ЕНК (39663) имеют dedicated thematicSearchId # в NSPD API (5 и 15 соответственно) и не всегда попадают в базовый thematic=1 snapshot # если totalCount в их категории > 20. # Для ЗУ/Зданий/Сооружений/ОНС — dedicated thematicSearchId нет: отдельный вызов # вернёт те же 20 что и базовый snapshot, поэтому их пропускаем. # Если category ∈ GRID_WALK_CATS и bbox валиден — grid-walk даёт лучшее покрытие (225 точек). for cat_id in PER_CATEGORY_PROBE_CATS: meta_total = snapshot.meta_counts.get(cat_id, 0) if meta_total == 0: # NSPD не вернул meta для этой категории — объектов нет в квартале continue # Проверяем: нужен ли per-category probe? # Условия skip: category ∈ GRID_WALK_CATS И bbox валиден → grid-walk сделает это лучше. # PER_CATEGORY_PROBE_CATS ∩ GRID_WALK_CATS = ∅ (ZOUIT/ENK не в GRID_WALK_CATS), # но явная проверка оставлена как safety guard. if cat_id in GRID_WALK_CATS and quarter_bbox_3857(db, quarter) is not None: logger.debug( "harvest_quarter: skip per-cat probe cat=%d quarter=%s — grid-walk preferred", cat_id, quarter, ) continue update_progress({"phase": f"per_cat_probe_{cat_id}_started", "quarter": quarter}) try: cat_snapshot = await client.search_by_quarter(quarter, category_id=cat_id) result.snapshot_requests += 1 except (NspdBulkWafError, NspdBulkRateLimitError): # #2464-A: бан IP / исчерпанные ретраи — НЕ «этот cat не дошёл». # Контракт harvest_quarter (Raises:) обещает пробросить их наверх, # а голый except ниже их глотал: прогон доходил до status='done' # с частичными данными. Прод-замер 13.08: 23 job'а, 50 WAF-блоков, # 0 упавших — то есть бан ни разу не остановил сбор. raise except Exception as e: logger.warning( "harvest_quarter: per-cat probe failed cat=%d quarter=%s: %s", cat_id, quarter, e, ) continue if not cat_snapshot.features: continue cat_stats = upsert_features(db, cat_snapshot.features, source="search_per_cat") result.parcels_upserted += cat_stats["parcels"] result.buildings_upserted += cat_stats["buildings"] result.constructions_upserted += cat_stats["constructions"] result.oncs_upserted += cat_stats["oncs"] result.enks_upserted += cat_stats["enks"] result.zouit_upserted += cat_stats["zouit"] db.commit() new_total = sum(cat_stats.values()) logger.info( "harvest_quarter per-cat probe: quarter=%s cat=%d new_features=%d", quarter, cat_id, new_total, ) # ── Phase 2-3: grid-walk для overflow категорий ─────────────────────────── overflow_cats = [c for c in snapshot.overflow_categories if c in GRID_WALK_CATS] for cat_id in overflow_cats: cat_label = "parcels" if cat_id == CAT_PARCEL else "buildings" update_progress({"phase": f"grid_walk_{cat_label}_started", "quarter": quarter}) discovered, n_requests, layer_failed = await _grid_walk_category( db=db, client=client, quarter=quarter, layer_id=cat_id, update_progress=update_progress, ) result.grid_walk_requests += n_requests if cat_id == CAT_PARCEL: result.parcels_upserted += discovered else: result.buildings_upserted += discovered # Issue #252: per-layer skip — слой целиком сбойнул на server-side 500/ # ServiceException. Фиксируем в harvest_meta (cadastre_jobs.phase_state), # квартал продолжаем (остальные слои и фазы отрабатывают штатно). done_progress: dict[str, Any] = { "phase": f"grid_walk_{cat_label}_done", "quarter": quarter, f"{cat_label}_grid_discovered": discovered, } if layer_failed: result.failed_layers.append(cat_id) done_progress["harvest_meta"] = {f"layer_{cat_id}_failed": True} db.commit() update_progress(done_progress) # ── Phase 4: quarter stats + auto-heal geom из snapshot ───────────────── # Bug #1583: кварталы с broken/NULL geom дают quarter_bbox_3857() == None → # grid-walk следующего прогона молча пропускается. Чиним 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() # Issue #252: финальный phase_state несёт АГРЕГИРОВАННЫЙ harvest_meta по всем # сбойным слоям. progress_cb мержит phase_state через JSONB `||` (shallow) — # per-layer done-апдейты перетёрли бы harvest_meta друг друга, поэтому в # терминальном 'done' собираем полный список → итоговая строка корректна. done_state: dict[str, Any] = {"phase": "done", "quarter": quarter} if result.failed_layers: done_state["harvest_meta"] = {f"layer_{lid}_failed": True for lid in result.failed_layers} update_progress(done_state) result.phase_state = done_state logger.info( "harvest_quarter done: quarter=%s parcels=%d buildings=%d " "constructions=%d zouit=%d grid_requests=%d failed_layers=%s", quarter, result.parcels_upserted, result.buildings_upserted, result.constructions_upserted, result.zouit_upserted, result.grid_walk_requests, result.failed_layers or None, ) return result # ── Grid-walk для одной категории ──────────────────────────────────────────── async def _grid_walk_category( db: Session, client: NSPDBulkClient, quarter: str, layer_id: int, grid_size: int = 15, tile_size: int = 512, update_progress: Callable[[dict[str, Any]], None] | None = None, heartbeat_every: int = 50, ) -> tuple[int, int, bool]: """Grid-walk layer_id в bbox квартала. Генерирует grid_size×grid_size ячеек, для каждой делает wms_feature_info. Дедуплицирует по cad_num, upsert'ит новые объекты. Args: update_progress: Fix I — callback для heartbeat внутри grid-walk loop. 225 cells × ~70ms = ~15-75s без heartbeat, zombie-cleanup ложно помечал worker как мёртвого. Вызываем каждые heartbeat_every cells. Returns: (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: logger.warning( "_grid_walk_category: bbox для квартала %s не найден, пропускаем layer=%d", quarter, layer_id, ) return 0, 0, False grid_points = generate_grid_click_points(bbox, grid_size=grid_size, tile_size=tile_size) # Bug #1584: считаем discovered ТОЛЬКО для таблицы запрошенного layer_id, а не # sum(stats.values()). Иначе skipped/чужекатегорийные features завышают счётчик, # ошибочно приписываясь к parcels/buildings вызывающим (harvest_quarter:230-233). layer_count_key = "parcels" if layer_id == CAT_PARCEL else "buildings" 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 # чтобы cleanup_zombies (5min stale check) не cancel'нул живой worker. if update_progress is not None and idx > 0 and idx % heartbeat_every == 0: update_progress( { "phase": f"grid_walk_{layer_id}_cell_{idx}", "quarter": quarter, "grid_cells_done": idx, "grid_cells_total": len(grid_points), } ) try: features = await client.wms_feature_info( layer_id=layer_id, bbox=cell_bbox, click_xy=click_xy, width=tile_size, height=tile_size, ) requests += 1 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: server error layer=%d quarter=%s cell=%d: %s", layer_id, quarter, idx, e, ) requests += 1 server_errors += 1 continue except (NspdBulkWafError, NspdBulkRateLimitError): # #2464-A: 403 WAF — бан IP, а не «этот cell не дошёл». Продолжать # обход значит углублять бан и дописать в БД ложный нулевой слой. # Зеркало уже исправленных nspd_bulk_client.get_features_in_bbox_grid # и nspd_client.get_features_in_bbox_grid (#2464-G). logger.warning( "_grid_walk_category: WAF/rate-limit layer=%d quarter=%s cell=%d — прерываем", layer_id, quarter, idx, ) raise 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 continue for feature in features: cad = feature.cad_num if not cad or cad in discovered_cads: continue discovered_cads.add(cad) # Upsert в соответствующую таблицу try: with db.begin_nested(): stats = upsert_features(db, [feature], source="wms_grid_walk") upserted += stats[layer_count_key] except Exception as e: logger.warning( "_grid_walk_category: upsert failed cad=%s layer=%d: %s", cad, layer_id, e, ) # 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 (NspdBulkWafError, NspdBulkRateLimitError): # #2464: бан IP / исчерпанные ретраи — НЕ «сбойный квартал». Голый # except ниже их глотал, хотя его же комментарий обещал обратное: # «WAF 403 пробросится из client и прервёт прогон». Прервать он не мог — # ловил сам себя, и цикл шёл дальше по всем оставшимся кварталам, долбя # уже блокирующий WAF и углубляя бан. Замер прода 20.08: limit=500 # участков раскладывается на 174 квартала, каждый — grid-walk по 49 # запросов, то есть до ~8500 обращений вместо остановки на первом. # Тот же фикс, что в harvest_quarter выше (#2464-A) — там это место # уже чинили, а это пропустили. db.rollback() raise except Exception as e: # Один сбойный квартал не валит весь backfill — лог + продолжаем. 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 ─────────────────────────────────────────────────────── def upsert_features( db: Session, features: list[NSPDBulkFeature], source: str = "search", ) -> dict[str, int]: """Маршрутизировать features по category_id в специфичные upsert-функции. Returns: Словарь {table: count} с количеством upserted строк. """ counts: dict[str, int] = { "parcels": 0, "buildings": 0, "constructions": 0, "oncs": 0, "enks": 0, "zouit": 0, "skipped": 0, } for feature in features: cat = feature.category_id # Per-feature SAVEPOINT (backend.md SAVEPOINT rule): один битый feature (edge-case # GeoJSON / NOT NULL violation / type mismatch) НЕ должен ронять весь snapshot-tx # квартала. Зеркалит _grid_walk_category: # 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 # ── cad_parcels upsert ─────────────────────────────────────────────────────── def upsert_parcel(db: Session, feature: NSPDBulkFeature, source: str = "search") -> None: """Upsert ЗУ ЕГРН в cad_parcels. ON CONFLICT (cad_num) DO UPDATE.""" opts = feature.options cad_num = opts.effective_cad_num if not cad_num: logger.debug("upsert_parcel: пропуск — нет cad_num в feature id=%s", feature.id) return raw_opts = feature.properties.get("options") or {} # cad_parcels.geom — geometry(Polygon, 4326). NSPD изредка возвращает Point # для участков без detailed boundary — INSERT fail'нит на schema check. # Сохраняем raw_props (есть координаты), geom=NULL. Не блокирует Phase 1. geom = feature.geometry or {} geom_type = geom.get("type") if isinstance(geom, dict) else None if geom_type in ("Polygon", "MultiPolygon"): geom_json: str | None = json.dumps(feature.geometry) else: if geom_type: logger.info( "upsert_parcel: cad_num=%s имеет geometry type=%s (не Polygon) — geom=NULL", cad_num, geom_type, ) geom_json = None quarter_cad = opts.quarter_cad_number or _derive_quarter_cad(cad_num) db.execute( text(""" INSERT INTO cad_parcels ( cad_num, quarter_cad_number, category_id, land_record_area, specified_area, declared_area, land_record_category_type, land_record_subtype, land_record_type, permitted_use_established_by_document, cost_value, cost_index, ownership_type, right_type, readable_address, status, previously_posted, registration_date, subcategory, cadastral_districts_code, geom, raw_props, source, fetched_at, updated_at ) VALUES ( :cad_num, :quarter_cad_number, 36368, :land_record_area, :specified_area, :declared_area, :land_record_category_type, :land_record_subtype, :land_record_type, :permitted_use, :cost_value, :cost_index, :ownership_type, :right_type, :readable_address, :status, :previously_posted, CAST(:registration_date AS date), :subcategory, :cadastral_districts_code, -- ST_Multi: cad_parcels.geom = MULTIPOLYGON (migration 93). -- NSPD возвращает Polygon ИЛИ MultiPolygon (Многоконтурные участки) — -- оба нужно сохранить, поэтому ST_Multi() coerce'ит Polygon → MultiPolygon. CASE WHEN CAST(:geom AS text) IS NOT NULL THEN ST_Multi( ST_Transform( ST_SetSRID(ST_GeomFromGeoJSON(CAST(:geom AS text)), 3857), 4326 ) ) ELSE NULL END, CAST(:raw_props AS jsonb), :source, NOW(), NOW() ) ON CONFLICT (cad_num) DO UPDATE SET quarter_cad_number = EXCLUDED.quarter_cad_number, land_record_area = EXCLUDED.land_record_area, specified_area = EXCLUDED.specified_area, declared_area = EXCLUDED.declared_area, land_record_category_type = EXCLUDED.land_record_category_type, land_record_subtype = EXCLUDED.land_record_subtype, land_record_type = EXCLUDED.land_record_type, permitted_use_established_by_document = EXCLUDED.permitted_use_established_by_document, cost_value = EXCLUDED.cost_value, cost_index = EXCLUDED.cost_index, ownership_type = EXCLUDED.ownership_type, right_type = EXCLUDED.right_type, readable_address = EXCLUDED.readable_address, status = EXCLUDED.status, previously_posted = EXCLUDED.previously_posted, registration_date = EXCLUDED.registration_date, subcategory = EXCLUDED.subcategory, cadastral_districts_code = EXCLUDED.cadastral_districts_code, geom = COALESCE(EXCLUDED.geom, cad_parcels.geom), raw_props = EXCLUDED.raw_props, source = EXCLUDED.source, updated_at = NOW() """), { "cad_num": cad_num, "quarter_cad_number": quarter_cad, "land_record_area": _safe_numeric(raw_opts.get("land_record_area")), "specified_area": _safe_numeric(raw_opts.get("specified_area")), "declared_area": _safe_numeric(raw_opts.get("declared_area")), "land_record_category_type": raw_opts.get("land_record_category_type"), "land_record_subtype": raw_opts.get("land_record_subtype"), "land_record_type": raw_opts.get("land_record_type"), "permitted_use": raw_opts.get("permitted_use_established_by_document"), "cost_value": _safe_numeric(raw_opts.get("cost_value")), "cost_index": _safe_numeric(raw_opts.get("cost_index")), "ownership_type": raw_opts.get("ownership_type"), "right_type": raw_opts.get("right_type"), "readable_address": raw_opts.get("readable_address"), "status": raw_opts.get("status"), "previously_posted": raw_opts.get("previously_posted"), "registration_date": _parse_nspd_date(raw_opts.get("registration_date")), "subcategory": _safe_int(raw_opts.get("subcategory")), "cadastral_districts_code": _safe_int(raw_opts.get("cadastral_districts_code")), "geom": geom_json, "raw_props": json.dumps(raw_opts, ensure_ascii=False), "source": source, }, ) # ── cad_buildings upsert ───────────────────────────────────────────────────── def upsert_building(db: Session, feature: NSPDBulkFeature, source: str = "search") -> None: """Upsert здание ЕГРН в cad_buildings. ON CONFLICT (cad_num) DO UPDATE.""" opts = feature.options cad_num = opts.effective_cad_num if not cad_num: logger.debug("upsert_building: пропуск — нет cad_num в feature id=%s", feature.id) return raw_opts = feature.properties.get("options") or {} geom_json = json.dumps(feature.geometry) if feature.geometry else None quarter_cad = opts.quarter_cad_number or _derive_quarter_cad(cad_num) # floors из NSPD может быть str "5" или int 5 floors_val = _safe_int(raw_opts.get("floors")) ugnd_floors_val = _safe_int(raw_opts.get("underground_floors")) # cultural_heritage — строка → bool ch_str = raw_opts.get("cultural_heritage") cultural_heritage_bool = bool(ch_str) if ch_str is not None else None db.execute( text(""" INSERT INTO cad_buildings ( cad_num, quarter_cad_number, objdoc_id, registers_id, purpose, building_name, floors, underground_floors, area, build_record_area, cost_value, cost_index, year_built, year_commisioning, build_record_registration_date, registration_date, ownership_type, cultural_heritage_object, cultural_heritage_val, readable_address, status, common_data_status, type, build_record_type_value, geom, raw_props, source, fetched_at, updated_at ) VALUES ( :cad_num, :quarter_cad_number, :objdoc_id, :registers_id, :purpose, :building_name, :floors, :underground_floors, :area, :build_record_area, :cost_value, :cost_index, :year_built, :year_commisioning, CAST(:build_record_registration_date AS date), CAST(:registration_date AS date), :ownership_type, :cultural_heritage_object, :cultural_heritage_val, :readable_address, :status, :common_data_status, :type, :build_record_type_value, CASE WHEN CAST(:geom AS text) IS NOT NULL THEN ST_Transform( ST_SetSRID(ST_GeomFromGeoJSON(CAST(:geom AS text)), 3857), 4326 ) ELSE NULL END, CAST(:raw_props AS jsonb), :source, NOW(), NOW() ) ON CONFLICT (cad_num) DO UPDATE SET quarter_cad_number = EXCLUDED.quarter_cad_number, objdoc_id = COALESCE(EXCLUDED.objdoc_id, cad_buildings.objdoc_id), registers_id = COALESCE(EXCLUDED.registers_id, cad_buildings.registers_id), purpose = EXCLUDED.purpose, building_name = EXCLUDED.building_name, floors = EXCLUDED.floors, underground_floors = EXCLUDED.underground_floors, area = EXCLUDED.area, build_record_area = EXCLUDED.build_record_area, cost_value = EXCLUDED.cost_value, cost_index = EXCLUDED.cost_index, year_built = EXCLUDED.year_built, year_commisioning = EXCLUDED.year_commisioning, build_record_registration_date = EXCLUDED.build_record_registration_date, registration_date = EXCLUDED.registration_date, ownership_type = EXCLUDED.ownership_type, cultural_heritage_object = EXCLUDED.cultural_heritage_object, cultural_heritage_val = EXCLUDED.cultural_heritage_val, readable_address = EXCLUDED.readable_address, status = EXCLUDED.status, common_data_status = EXCLUDED.common_data_status, type = EXCLUDED.type, build_record_type_value = EXCLUDED.build_record_type_value, geom = COALESCE(EXCLUDED.geom, cad_buildings.geom), raw_props = EXCLUDED.raw_props, source = EXCLUDED.source, updated_at = NOW() """), { "cad_num": cad_num, "quarter_cad_number": quarter_cad, "objdoc_id": _safe_int(raw_opts.get("objdoc_id")), "registers_id": _safe_int(raw_opts.get("registers_id")), "purpose": raw_opts.get("purpose"), "building_name": raw_opts.get("building_name") or raw_opts.get("name"), "floors": floors_val, "underground_floors": ugnd_floors_val, "area": _safe_numeric(raw_opts.get("area")), "build_record_area": _safe_numeric(raw_opts.get("build_record_area")), "cost_value": _safe_numeric(raw_opts.get("cost_value")), "cost_index": _safe_numeric(raw_opts.get("cost_index")), "year_built": _safe_int(raw_opts.get("year_built")), "year_commisioning": _safe_int(raw_opts.get("year_commisioning")), "build_record_registration_date": _parse_nspd_date( raw_opts.get("build_record_registration_date") ), "registration_date": _parse_nspd_date(raw_opts.get("registration_date")), "ownership_type": raw_opts.get("ownership_type"), "cultural_heritage_object": cultural_heritage_bool, "cultural_heritage_val": ch_str, "readable_address": raw_opts.get("readable_address"), "status": raw_opts.get("status"), "common_data_status": raw_opts.get("common_data_status"), "type": raw_opts.get("obj_type") or raw_opts.get("type"), "build_record_type_value": ( raw_opts.get("build_record_type_value") or raw_opts.get("build_record_type") ), "geom": geom_json, "raw_props": json.dumps(raw_opts, ensure_ascii=False), "source": source, }, ) # ── cad_constructions upsert ───────────────────────────────────────────────── def upsert_construction(db: Session, feature: NSPDBulkFeature, source: str = "search") -> None: """Upsert сооружение в cad_constructions. ON CONFLICT (cad_num) DO UPDATE.""" raw_props = feature.properties raw_opts = raw_props.get("options") or {} cad_num = raw_opts.get("cad_num") or raw_opts.get("cad_number") if not cad_num: logger.debug("upsert_construction: пропуск — нет cad_num, feature id=%s", feature.id) return cat_id = feature.category_id # Определить тип: 36383 = сооружение, у NSPD есть 2 варианта is_egrn = bool(raw_opts.get("is_egrn") or raw_opts.get("registers_id")) category_name = raw_props.get("categoryName") or str(cat_id or "") quarter_cad = raw_opts.get("quarter_cad_number") or _derive_quarter_cad(cad_num) geom_json = json.dumps(feature.geometry) if feature.geometry else None # params_* поля из NSPD characteristics params = raw_opts.get("params") or {} db.execute( text(""" INSERT INTO cad_constructions ( cad_num, quarter_cad_number, category_name, is_egrn, params_purpose, params_name, params_area, params_built_up_area, params_year_built, params_year_commisioning, params_height, params_volume, params_floors, params_underground_floors, object_type_value, facility_cad_number, cost_value, cost_index, ownership_type, right_type, readable_address, address_readable_address, status, common_data_status, registration_date, registers_id, united_cad_number, geom, raw_props, source, fetched_at, updated_at ) VALUES ( :cad_num, :quarter_cad_number, :category_name, :is_egrn, :params_purpose, :params_name, :params_area, :params_built_up_area, :params_year_built, :params_year_commisioning, :params_height, :params_volume, :params_floors, :params_underground_floors, :object_type_value, :facility_cad_number, :cost_value, :cost_index, :ownership_type, :right_type, :readable_address, :address_readable_address, :status, :common_data_status, CAST(:registration_date AS date), :registers_id, :united_cad_number, CASE WHEN CAST(:geom AS text) IS NOT NULL THEN ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(CAST(:geom AS text)), 3857), 4326) ELSE NULL END, CAST(:raw_props AS jsonb), :source, NOW(), NOW() ) ON CONFLICT (cad_num) DO UPDATE SET category_name = EXCLUDED.category_name, is_egrn = EXCLUDED.is_egrn, params_purpose = EXCLUDED.params_purpose, params_name = EXCLUDED.params_name, params_area = EXCLUDED.params_area, params_built_up_area = EXCLUDED.params_built_up_area, params_year_built = EXCLUDED.params_year_built, params_year_commisioning = EXCLUDED.params_year_commisioning, params_height = EXCLUDED.params_height, params_volume = EXCLUDED.params_volume, params_floors = EXCLUDED.params_floors, params_underground_floors = EXCLUDED.params_underground_floors, object_type_value = EXCLUDED.object_type_value, facility_cad_number = EXCLUDED.facility_cad_number, cost_value = EXCLUDED.cost_value, cost_index = EXCLUDED.cost_index, ownership_type = EXCLUDED.ownership_type, right_type = EXCLUDED.right_type, readable_address = EXCLUDED.readable_address, address_readable_address = EXCLUDED.address_readable_address, status = EXCLUDED.status, common_data_status = EXCLUDED.common_data_status, registration_date = EXCLUDED.registration_date, registers_id = EXCLUDED.registers_id, united_cad_number = EXCLUDED.united_cad_number, geom = COALESCE(EXCLUDED.geom, cad_constructions.geom), raw_props = EXCLUDED.raw_props, source = EXCLUDED.source, updated_at = NOW() """), { "cad_num": cad_num, "quarter_cad_number": quarter_cad, "category_name": category_name, "is_egrn": is_egrn, "params_purpose": params.get("purpose") or raw_opts.get("purpose"), "params_name": params.get("name") or raw_opts.get("building_name"), "params_area": _safe_numeric(params.get("area") or raw_opts.get("area")), "params_built_up_area": _safe_numeric(raw_opts.get("built_up_area")), "params_year_built": _safe_int(params.get("year_built") or raw_opts.get("year_built")), "params_year_commisioning": _safe_int(raw_opts.get("year_commisioning")), "params_height": _safe_numeric(raw_opts.get("height")), "params_volume": _safe_numeric(raw_opts.get("volume")), "params_floors": _safe_int(raw_opts.get("floors")), "params_underground_floors": _safe_int(raw_opts.get("underground_floors")), "object_type_value": raw_opts.get("object_type_value"), "facility_cad_number": raw_opts.get("facility_cad_number"), "cost_value": _safe_numeric(raw_opts.get("cost_value")), "cost_index": _safe_numeric(raw_opts.get("cost_index")), "ownership_type": raw_opts.get("ownership_type"), "right_type": raw_opts.get("right_type"), "readable_address": raw_opts.get("readable_address"), "address_readable_address": raw_opts.get("address_readable_address"), "status": raw_opts.get("status"), "common_data_status": raw_opts.get("common_data_status"), "registration_date": _parse_nspd_date(raw_opts.get("registration_date")), "registers_id": _safe_int(raw_opts.get("registers_id")), "united_cad_number": raw_opts.get("united_cad_number"), "geom": geom_json, "raw_props": json.dumps(raw_opts, ensure_ascii=False), "source": source, }, ) # ── cad_oncs upsert ────────────────────────────────────────────────────────── def upsert_onc(db: Session, feature: NSPDBulkFeature, source: str = "search") -> None: """Upsert ОНС в cad_oncs. ON CONFLICT (cad_num) DO UPDATE.""" raw_opts = feature.properties.get("options") or {} cad_num = raw_opts.get("cad_num") or raw_opts.get("cad_number") if not cad_num: logger.debug("upsert_onc: пропуск — нет cad_num, feature id=%s", feature.id) return quarter_cad = raw_opts.get("quarter_cad_number") or _derive_quarter_cad(cad_num) geom_json = json.dumps(feature.geometry) if feature.geometry else None db.execute( text(""" INSERT INTO cad_oncs ( cad_num, quarter_cad_number, purpose, area, build_record_area, cost_value, status, readable_address, geom, raw_props, source, fetched_at, updated_at ) VALUES ( :cad_num, :quarter_cad_number, :purpose, :area, :build_record_area, :cost_value, :status, :readable_address, CASE WHEN CAST(:geom AS text) IS NOT NULL THEN ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(CAST(:geom AS text)), 3857), 4326) ELSE NULL END, CAST(:raw_props AS jsonb), :source, NOW(), NOW() ) ON CONFLICT (cad_num) DO UPDATE SET quarter_cad_number = EXCLUDED.quarter_cad_number, purpose = EXCLUDED.purpose, area = EXCLUDED.area, build_record_area = EXCLUDED.build_record_area, cost_value = EXCLUDED.cost_value, status = EXCLUDED.status, readable_address = EXCLUDED.readable_address, geom = COALESCE(EXCLUDED.geom, cad_oncs.geom), raw_props = EXCLUDED.raw_props, source = EXCLUDED.source, updated_at = NOW() """), { "cad_num": cad_num, "quarter_cad_number": quarter_cad, "purpose": raw_opts.get("purpose"), "area": _safe_numeric(raw_opts.get("area")), "build_record_area": _safe_numeric(raw_opts.get("build_record_area")), "cost_value": _safe_numeric(raw_opts.get("cost_value")), "status": raw_opts.get("status"), "readable_address": raw_opts.get("readable_address"), "geom": geom_json, "raw_props": json.dumps(raw_opts, ensure_ascii=False), "source": source, }, ) # ── cad_enk upsert ─────────────────────────────────────────────────────────── def upsert_enk(db: Session, feature: NSPDBulkFeature, source: str = "search") -> None: """Upsert ЕНК в cad_enk. ON CONFLICT (cad_num) DO UPDATE.""" raw_opts = feature.properties.get("options") or {} cad_num = raw_opts.get("cad_num") or raw_opts.get("cad_number") if not cad_num: logger.debug("upsert_enk: пропуск — нет cad_num, feature id=%s", feature.id) return quarter_cad = raw_opts.get("quarter_cad_number") or _derive_quarter_cad(cad_num) geom_json = json.dumps(feature.geometry) if feature.geometry else None db.execute( text(""" INSERT INTO cad_enk ( cad_num, quarter_cad_number, name, purpose, type, cost_value, facility_cad_number, readable_address, registration_date, registers_id, geom, raw_props, fetched_at, updated_at ) VALUES ( :cad_num, :quarter_cad_number, :name, :purpose, :type, :cost_value, :facility_cad_number, :readable_address, CAST(:registration_date AS date), :registers_id, CASE WHEN CAST(:geom AS text) IS NOT NULL THEN ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(CAST(:geom AS text)), 3857), 4326) ELSE NULL END, CAST(:raw_props AS jsonb), NOW(), NOW() ) ON CONFLICT (cad_num) DO UPDATE SET quarter_cad_number = EXCLUDED.quarter_cad_number, name = EXCLUDED.name, purpose = EXCLUDED.purpose, type = EXCLUDED.type, cost_value = EXCLUDED.cost_value, facility_cad_number = EXCLUDED.facility_cad_number, readable_address = EXCLUDED.readable_address, registration_date = EXCLUDED.registration_date, registers_id = EXCLUDED.registers_id, geom = COALESCE(EXCLUDED.geom, cad_enk.geom), raw_props = EXCLUDED.raw_props, updated_at = NOW() """), { "cad_num": cad_num, "quarter_cad_number": quarter_cad, "name": raw_opts.get("name"), "purpose": raw_opts.get("purpose"), "type": raw_opts.get("type") or raw_opts.get("object_type"), "cost_value": _safe_numeric(raw_opts.get("cost_value")), "facility_cad_number": raw_opts.get("facility_cad_number"), "readable_address": raw_opts.get("readable_address"), "registration_date": _parse_nspd_date(raw_opts.get("registration_date")), "registers_id": _safe_int(raw_opts.get("registers_id")), "geom": geom_json, "raw_props": json.dumps(raw_opts, ensure_ascii=False), }, ) # ── cad_zouit upsert ───────────────────────────────────────────────────────── def upsert_zouit(db: Session, feature: NSPDBulkFeature, source: str = "search") -> None: """Upsert ЗОУИТ в cad_zouit. Deduplicate по (reg_numb_border, category_id). Geom — MultiPolygon; если пришёл Polygon — NSPD иногда так делает. """ raw_opts = feature.properties.get("options") or {} cat_id = feature.category_id reg_numb_border = raw_opts.get("reg_numb_border") or raw_opts.get("registration_number") type_zone = raw_opts.get("type_zone") or raw_opts.get("typeZone") or "" if not reg_numb_border or cat_id is None: logger.debug( "upsert_zouit: пропуск — нет reg_numb_border или category_id, feature id=%s", feature.id, ) return # cad_zouit.geom — geometry(MultiPolygon, 4326). Polygon допустим (ST_Multi # обернёт в SQL), но Point/LineString → INSERT fail. Фильтруем как в upsert_parcel. geom = feature.geometry or {} geom_type = geom.get("type") if isinstance(geom, dict) else None if geom_type in ("Polygon", "MultiPolygon"): geom_json: str | None = json.dumps(feature.geometry) else: if geom_type: logger.info( "upsert_zouit: reg=%s имеет geometry type=%s (не Polygon) — geom=NULL", reg_numb_border, geom_type, ) geom_json = None db.execute( text(""" INSERT INTO cad_zouit ( reg_numb_border, category_id, category_name, type_zone, type_boundary_value, subcategory, cadastral_district, content_restrict_encumbrances, name_by_doc, legal_act_name, legal_act_date, legal_act_number, legal_act_issuer, registration_date, old_account_number, geom, raw_props, source, fetched_at, updated_at ) VALUES ( :reg_numb_border, :category_id, :category_name, :type_zone, :type_boundary_value, :subcategory, :cadastral_district, :content_restrict_encumbrances, :name_by_doc, :legal_act_name, CAST(:legal_act_date AS date), :legal_act_number, :legal_act_issuer, CAST(:registration_date AS date), :old_account_number, CASE WHEN CAST(:geom AS text) IS NOT NULL THEN ST_Multi( ST_Transform( ST_SetSRID(ST_GeomFromGeoJSON(CAST(:geom AS text)), 3857), 4326 ) ) ELSE NULL END, CAST(:raw_props AS jsonb), :source, NOW(), NOW() ) ON CONFLICT (reg_numb_border, category_id) DO UPDATE SET category_name = EXCLUDED.category_name, type_zone = EXCLUDED.type_zone, type_boundary_value = EXCLUDED.type_boundary_value, subcategory = EXCLUDED.subcategory, cadastral_district = EXCLUDED.cadastral_district, content_restrict_encumbrances = EXCLUDED.content_restrict_encumbrances, name_by_doc = EXCLUDED.name_by_doc, legal_act_name = EXCLUDED.legal_act_name, legal_act_date = EXCLUDED.legal_act_date, legal_act_number = EXCLUDED.legal_act_number, legal_act_issuer = EXCLUDED.legal_act_issuer, registration_date = EXCLUDED.registration_date, old_account_number = EXCLUDED.old_account_number, geom = COALESCE(EXCLUDED.geom, cad_zouit.geom), raw_props = EXCLUDED.raw_props, source = EXCLUDED.source, updated_at = NOW() """), { "reg_numb_border": reg_numb_border, "category_id": cat_id, "category_name": feature.properties.get("categoryName") or str(cat_id), "type_zone": type_zone, "type_boundary_value": raw_opts.get("type_boundary_value"), "subcategory": _safe_int(raw_opts.get("subcategory")), "cadastral_district": raw_opts.get("cadastral_district"), "content_restrict_encumbrances": raw_opts.get("content_restrict_encumbrances"), "name_by_doc": raw_opts.get("name_by_doc"), "legal_act_name": raw_opts.get("legal_act_name"), "legal_act_date": _parse_nspd_date(raw_opts.get("legal_act_date")), "legal_act_number": raw_opts.get("legal_act_number"), "legal_act_issuer": raw_opts.get("legal_act_issuer"), "registration_date": _parse_nspd_date(raw_opts.get("registration_date")), "old_account_number": raw_opts.get("old_account_number"), "geom": geom_json, "raw_props": json.dumps(raw_opts, ensure_ascii=False), "source": source, }, ) # ── 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 ───────────────────────────────────────────────── def upsert_quarter_stats( db: Session, quarter: str, feature: NSPDBulkFeature, ) -> None: """Upsert квартальную статистику из feature 36381 в cad_quarter_stats. cad_quarter_stats имеет FK на cad_quarters_geom(cad_number). Если квартала ещё нет в cad_quarters_geom — пропускаем тихо (INSERT не нарушит FK, но строка не появится; при следующем проходе после добавления квартала — сработает). """ raw_opts = feature.properties.get("options") or {} db.execute( text(""" INSERT INTO cad_quarter_stats ( cad_number, cnt_land, cnt_oks, cnt_land_geom, cnt_land_not_geom, cnt_oks_area, cnt_oks_geom, cnt_enk, cnt_pik, cost_value_total, cost_value_total_geom, sum_land_area, sum_land_geom_area, date_cr, real_srid, raw_props, fetched_at, updated_at ) SELECT :cad_number, :cnt_land, :cnt_oks, :cnt_land_geom, :cnt_land_not_geom, :cnt_oks_area, :cnt_oks_geom, :cnt_enk, :cnt_pik, :cost_value_total, :cost_value_total_geom, :sum_land_area, :sum_land_geom_area, CAST(:date_cr AS date), :real_srid, CAST(:raw_props AS jsonb), NOW(), NOW() WHERE EXISTS ( SELECT 1 FROM cad_quarters_geom WHERE cad_number = :cad_number ) ON CONFLICT (cad_number) DO UPDATE SET cnt_land = EXCLUDED.cnt_land, cnt_oks = EXCLUDED.cnt_oks, cnt_land_geom = EXCLUDED.cnt_land_geom, cnt_land_not_geom = EXCLUDED.cnt_land_not_geom, cnt_oks_area = EXCLUDED.cnt_oks_area, cnt_oks_geom = EXCLUDED.cnt_oks_geom, cnt_enk = EXCLUDED.cnt_enk, cnt_pik = EXCLUDED.cnt_pik, cost_value_total = EXCLUDED.cost_value_total, cost_value_total_geom = EXCLUDED.cost_value_total_geom, sum_land_area = EXCLUDED.sum_land_area, sum_land_geom_area = EXCLUDED.sum_land_geom_area, date_cr = EXCLUDED.date_cr, real_srid = EXCLUDED.real_srid, raw_props = EXCLUDED.raw_props, updated_at = NOW() """), { "cad_number": quarter, "cnt_land": _safe_int(raw_opts.get("cnt_land")), "cnt_oks": _safe_int(raw_opts.get("cnt_oks")), "cnt_land_geom": _safe_int(raw_opts.get("cnt_land_geom")), "cnt_land_not_geom": _safe_int(raw_opts.get("cnt_land_not_geom")), "cnt_oks_area": _safe_int(raw_opts.get("cnt_oks_area")), "cnt_oks_geom": _safe_int(raw_opts.get("cnt_oks_geom")), "cnt_enk": _safe_int(raw_opts.get("cnt_enk")), "cnt_pik": _safe_int(raw_opts.get("cnt_pik")), "cost_value_total": _safe_numeric(raw_opts.get("cost_value_total")), "cost_value_total_geom": _safe_numeric(raw_opts.get("cost_value_total_geom")), "sum_land_area": _safe_numeric(raw_opts.get("sum_land_area")), "sum_land_geom_area": _safe_numeric(raw_opts.get("sum_land_geom_area")), "date_cr": _parse_nspd_date(raw_opts.get("date_cr")), "real_srid": _safe_int(raw_opts.get("real_srid")), "raw_props": json.dumps(raw_opts, ensure_ascii=False), }, ) # ── Утилиты ────────────────────────────────────────────────────────────────── def _derive_quarter_cad(cad_num: str) -> str: """Вывести номер квартала из кад. номера: '66:41:0303161:1' → '66:41:0303161'.""" parts = cad_num.split(":") if len(parts) >= 3: return ":".join(parts[:3]) return cad_num def _safe_numeric(value: Any) -> float | None: """Конвертировать значение в float или None (безопасно).""" if value is None: return None try: return float(value) except (ValueError, TypeError): return None def _safe_int(value: Any) -> int | None: """Конвертировать значение в int или None (безопасно).""" if value is None: return None try: return int(value) except (ValueError, TypeError): return None def _parse_nspd_date(value: Any) -> str | None: """Конвертировать NSPD date в ISO YYYY-MM-DD string. NSPD возвращает даты в смешанных форматах: - '13.03.2023' (DD.MM.YYYY — для Сооружения/ОНС/ЕНК) - '2023-03-13' (ISO — для ЗУ/Здания) - '2023-03-13 00:00:00' (ISO with time) - None / empty string PostgreSQL DATE cast принимает ISO. Без conversion DD.MM.YYYY → DatetimeFieldOverflow crash в upsert_*. Также валидируем что результат — actually-parseable date чтобы invalid input не уехал в SQL. """ import datetime as _dt if value is None or value == "": return None if not isinstance(value, str): return None s = value.strip() candidate: str | None = None # ISO format already — pass through (PG accept YYYY-MM-DD) if len(s) >= 10 and s[4] == "-" and s[7] == "-": candidate = s[:10] # DD.MM.YYYY → YYYY-MM-DD elif len(s) == 10 and s[2] == "." and s[5] == ".": parts = s.split(".") if len(parts) == 3 and all(p.isdigit() for p in parts): candidate = f"{parts[2]}-{parts[1]}-{parts[0]}" if candidate is None: return None # Validate: actually parseable as date (catches '2023-13-45', '32.13.2023', etc) try: _dt.date.fromisoformat(candidate) return candidate except ValueError: return None