gendesign/backend/app/services/cadastre/bulk_harvest.py

"""Оркестратор 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,
        )
        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,
) -> tuple[int, int]:
    """Grid-walk layer_id в bbox квартала.

    Генерирует grid_size×grid_size ячеек, для каждой делает wms_feature_info.
    Дедуплицирует по cad_num, upsert'ит новые объекты.

    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 cell_bbox, click_xy in 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:
            logger.warning(
                "_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": 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": raw_opts.get("build_record_registration_date"),
            "registration_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": 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": 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": 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": 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