"""Оркестратор 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 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 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) 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 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 = 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 db.commit() update_progress( { "phase": f"grid_walk_{cat_label}_done", "quarter": quarter, f"{cat_label}_grid_discovered": discovered, } ) # ── 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}) result.phase_state = {"phase": "done", "quarter": quarter} logger.info( "harvest_quarter done: quarter=%s parcels=%d buildings=%d " "constructions=%d zouit=%d grid_requests=%d", quarter, result.parcels_upserted, result.buildings_upserted, result.constructions_upserted, result.zouit_upserted, result.grid_walk_requests, ) 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]: """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) """ 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 grid_points = generate_grid_click_points(bbox, grid_size=grid_size, tile_size=tile_size) discovered_cads: set[str] = set() upserted = 0 requests = 0 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 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). logger.debug( "_grid_walk_category: wms_feature_info error layer=%d quarter=%s: %s", layer_id, quarter, 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 += sum(stats.values()) except Exception as e: logger.warning( "_grid_walk_category: upsert failed cad=%s layer=%d: %s", cad, layer_id, e, ) return upserted, requests # ── 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 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 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": 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