gendesign/backend/app/services/site_finder/quarter_dump_lookup.py

"""Quarter dump lookup helper for analyze_parcel.

Sprint 1.1 item #4 (feat/analyze-uses-quarter-dump):
Читает nspd_quarter_dumps кеш и извлекает:
- nspd_zoning   — территориальная зона ПЗЗ (G1) по centroid участка
- nspd_zouit_overlaps — список ЗОУИТ (G3) которые пересекаются с участком
- nspd_engineering_nearby — инженерные сооружения в 200м (I3)
- nspd_dump     — freshness metadata (доступность, возраст, trigger флаг)

Если дамп отсутствует или устарел (>180 дней) — fire-and-forget
harvest_quarter.apply_async() и продолжает без dump-derived полей.
"""

from __future__ import annotations

import json
import logging
import math
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import text
from sqlalchemy.exc import OperationalError, ProgrammingError
from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)

# Порог свежести дампа — 180 дней. Совпадает с beat-расписанием (раз в квартал
# на практике, но с запасом для редко запрашиваемых кварталов).
_DUMP_MAX_AGE_DAYS = 180

# Радиус поиска инженерных сооружений (метры) — договорённость #44 I3.
_ENGINEERING_RADIUS_M = 200

# Sentinel для isinstance-проверок и read-only fallback в parcels.py try/except.
# НИКОГДА не мутировать — использовать make_empty_result() для новых dict.
EMPTY_DUMP_RESULT: dict[str, Any] = {
    "nspd_zoning": None,
    "nspd_zouit_overlaps": [],
    "nspd_engineering_nearby": [],
    "nspd_risk_zones": [],
    "nspd_dump": {
        "available": False,
        "fetched_at_utc": None,
        "stale": False,
        "harvest_triggered": False,
        "total_features": None,
    },
}


def make_empty_result(
    *,
    fetched_at_utc: str | None = None,
    stale: bool = False,
    harvest_triggered: bool = False,
    total_features: int | None = None,
) -> dict[str, Any]:
    """Создаёт свежую копию empty-dump result с возможностью переопределить поля.

    Вызывать вместо EMPTY_DUMP_RESULT напрямую — чтобы не мутировать singleton.
    """
    return {
        "nspd_zoning": None,
        "nspd_zouit_overlaps": [],
        "nspd_engineering_nearby": [],
        "nspd_risk_zones": [],
        "nspd_dump": {
            "available": False,
            "fetched_at_utc": fetched_at_utc,
            "stale": stale,
            "harvest_triggered": harvest_triggered,
            "total_features": total_features,
        },
    }


def derive_quarter_cad(cad_num: str) -> str | None:
    """3-сегментный кадастровый квартал из любого кадастрового номера.

    - 3-сегмент (квартал)  '66:41:0204016'          → '66:41:0204016'
    - 4-сегмент (участок)  '66:41:0204016:10'        → '66:41:0204016'
    - 5-сегмент (здание)   '66:41:0204016:10:1'      → '66:41:0204016'
    - невалидный формат                               → None
    """
    parts = cad_num.strip().split(":")
    if len(parts) < 3:
        return None
    quarter = ":".join(parts[:3])
    # Каждый сегмент — только цифры
    if not all(p.isdigit() and len(p) >= 1 for p in parts[:3]):
        return None
    return quarter


def get_quarter_dump_data(
    db: Session,
    cad_num: str,
    parcel_wkt: str | None,
) -> dict[str, Any]:
    """Читает quarter dump для квартала cad_num и возвращает НСПД-контекст.

    Returns dict с ключами:
    - nspd_zoning: dict | None  — зона ПЗЗ из territorial_zones
    - nspd_zouit_overlaps: list[dict]  — ЗОУИТ пересечения
    - nspd_engineering_nearby: list[dict]  — инженерные сооружения в 200м
    - nspd_dump: dict  — freshness metadata

    Если дамп отсутствует или устарел — вызывает harvest_quarter.apply_async()
    (non-blocking) и возвращает пустые spatial поля + nspd_dump.available=False.
    Если parcel_wkt=None — возвращает только freshness metadata (нет geom для
    spatial queries).
    """
    quarter = derive_quarter_cad(cad_num)
    if quarter is None:
        logger.warning("quarter_dump_lookup: cannot derive quarter from cad=%s", cad_num)
        return make_empty_result()

    # Читаем строку дампа из БД. Денормализованные счётчики слоёв используются
    # для early-exit в spatial helpers (M2 mitigation).
    row = db.execute(
        text(
            """
            SELECT quarter_cad,
                   fetched_at_utc,
                   total_features,
                   harvest_error,
                   territorial_zones_count,
                   zouit_count,
                   engineering_count,
                   risks_count
              FROM nspd_quarter_dumps
             WHERE quarter_cad = :q
            """
        ),
        {"q": quarter},
    ).first()

    now = datetime.now(UTC)
    max_age = timedelta(days=_DUMP_MAX_AGE_DAYS)

    if row is None:
        # Дампа нет — ставим harvest в очередь
        harvest_triggered = _trigger_harvest(quarter)
        return make_empty_result(harvest_triggered=harvest_triggered)

    fetched_at: datetime = row[1]
    # Убедимся что timezone-aware для корректного сравнения
    if fetched_at.tzinfo is None:
        fetched_at = fetched_at.replace(tzinfo=UTC)

    total_features: int | None = row[2]
    harvest_error: str | None = row[3]
    territorial_zones_count: int = row[4] or 0
    zouit_count: int = row[5] or 0
    engineering_count: int = row[6] or 0
    risks_count: int = row[7] or 0

    is_stale = (now - fetched_at) > max_age
    has_error = harvest_error is not None

    # Устаревший или с ошибкой — триггерим повторный harvest
    if is_stale or has_error:
        harvest_triggered = _trigger_harvest(quarter)
        return make_empty_result(
            fetched_at_utc=fetched_at.isoformat(),
            stale=is_stale,
            harvest_triggered=harvest_triggered,
            total_features=total_features,
        )

    # Свежий дамп без ошибок — извлекаем spatial данные
    dump_meta: dict[str, Any] = {
        "available": True,
        "fetched_at_utc": fetched_at.isoformat(),
        "stale": False,
        "harvest_triggered": False,
        "total_features": total_features,
    }

    if parcel_wkt is None:
        # Нет геометрии участка — возвращаем только метаданные
        return {
            "nspd_zoning": None,
            "nspd_zouit_overlaps": [],
            "nspd_engineering_nearby": [],
            "nspd_risk_zones": [],
            "nspd_dump": dump_meta,
        }

    layer_counts = {
        "territorial_zones_count": territorial_zones_count,
        "zouit_count": zouit_count,
        "engineering_count": engineering_count,
        "risks_count": risks_count,
    }
    nspd_zoning = _get_zoning(db, quarter, parcel_wkt, layer_counts)
    nspd_zouit = _get_zouit_overlaps(db, quarter, parcel_wkt, layer_counts)
    nspd_engineering = _get_engineering_nearby(db, quarter, parcel_wkt, layer_counts)
    nspd_risk_zones = _get_risk_zones(db, quarter, parcel_wkt, layer_counts)

    return {
        "nspd_zoning": nspd_zoning,
        "nspd_zouit_overlaps": nspd_zouit,
        "nspd_engineering_nearby": nspd_engineering,
        "nspd_risk_zones": nspd_risk_zones,
        "nspd_dump": dump_meta,
    }


# ── Spatial helpers ───────────────────────────────────────────────────────────


def _get_zoning(
    db: Session,
    quarter: str,
    parcel_wkt: str,
    layer_counts: dict[str, int] | None = None,
) -> dict[str, Any] | None:
    """G1: ПЗЗ территориальная зона по centroid участка из dump.

    Geometry в features_json в EPSG:3857 — ST_Transform на чтении → 4326.
    layer_counts — денормализованные счётчики из строки дампа. Если
    territorial_zones_count == 0 — пропускаем heavy jsonb_array_elements scan.
    """
    if layer_counts is not None and layer_counts.get("territorial_zones_count", 1) == 0:
        return None
    try:
        row = db.execute(
            text(
                """
                SELECT feat.value->'properties' AS zone_props
                  FROM nspd_quarter_dumps d,
                       jsonb_array_elements(d.features_json) AS feat(value)
                 WHERE d.quarter_cad = :q
                   AND feat.value->>'layer' = 'territorial_zones'
                   AND (feat.value->'geometry') IS NOT NULL
                   AND feat.value->>'geometry' != 'null'
                   AND ST_Intersects(
                       ST_Transform(
                           ST_SetSRID(
                               ST_GeomFromGeoJSON(feat.value->>'geometry'),
                               3857
                           ),
                           4326
                       ),
                       ST_Centroid(ST_GeomFromText(:wkt, 4326))
                   )
                 LIMIT 1
                """
            ),
            {"q": quarter, "wkt": parcel_wkt},
        ).first()

        if row is None:
            return None

        props: dict[str, Any] = row[0] if isinstance(row[0], dict) else {}
        zone_code = props.get("reg_numb_border") or props.get("zone_code") or props.get("name")
        zone_name = props.get("type_zone") or props.get("zone_name") or props.get("name")
        return {
            "zone_code": zone_code,
            "zone_name": zone_name,
            "source": "nspd-quarter-dump",
            "raw_props": props,
        }
    except Exception as e:
        logger.warning("nspd zoning query failed for quarter=%s: %s", quarter, e)
        return None


def _get_zouit_overlaps(
    db: Session,
    quarter: str,
    parcel_wkt: str,
    layer_counts: dict[str, int] | None = None,
) -> list[dict[str, Any]]:
    """G3: ЗОУИТ которые пересекают участок.

    Проверяем 5 групп: zouit_okn, zouit_engineering, zouit_natural,
    zouit_protected, zouit_other.
    layer_counts — денормализованные счётчики. Если zouit_count == 0 —
    пропускаем heavy jsonb_array_elements scan.
    """
    if layer_counts is not None and layer_counts.get("zouit_count", 1) == 0:
        return []
    try:
        rows = db.execute(
            text(
                """
                SELECT feat.value->>'layer'         AS layer,
                       feat.value->'properties'     AS props
                  FROM nspd_quarter_dumps d,
                       jsonb_array_elements(d.features_json) AS feat(value)
                 WHERE d.quarter_cad = :q
                   AND feat.value->>'layer' LIKE 'zouit_%'
                   AND (feat.value->'geometry') IS NOT NULL
                   AND feat.value->>'geometry' != 'null'
                   AND ST_Intersects(
                       ST_Transform(
                           ST_SetSRID(
                               ST_GeomFromGeoJSON(feat.value->>'geometry'),
                               3857
                           ),
                           4326
                       ),
                       ST_GeomFromText(:wkt, 4326)
                   )
                """
            ),
            {"q": quarter, "wkt": parcel_wkt},
        ).fetchall()

        result: list[dict[str, Any]] = []
        for r in rows:
            layer: str = r[0] or ""
            props: dict[str, Any] = r[1] if isinstance(r[1], dict) else {}
            group_key = layer.removeprefix("zouit_")
            result.append(
                {
                    "group_key": group_key,
                    "layer": layer,
                    "subcategory": props.get("subcategory") or props.get("type_zone"),
                    "name": props.get("name") or props.get("object_name"),
                    "raw_props": props,
                }
            )
        return result
    except Exception as e:
        logger.warning("nspd zouit query failed for quarter=%s: %s", quarter, e)
        return []


def _get_engineering_nearby(
    db: Session,
    quarter: str,
    parcel_wkt: str,
    layer_counts: dict[str, int] | None = None,
) -> list[dict[str, Any]]:
    """I3: Инженерные сооружения в радиусе _ENGINEERING_RADIUS_M от centroid участка.

    layer_counts — денормализованные счётчики. Если engineering_count == 0 —
    пропускаем heavy jsonb_array_elements + ST_DWithin scan.
    """
    if layer_counts is not None and layer_counts.get("engineering_count", 1) == 0:
        return []
    try:
        rows = db.execute(
            text(
                """
                SELECT feat.value->'properties'  AS props,
                       ST_Distance(
                           ST_Transform(
                               ST_SetSRID(
                                   ST_GeomFromGeoJSON(feat.value->>'geometry'),
                                   3857
                               ),
                               4326
                           )::geography,
                           ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography
                       ) AS distance_m
                  FROM nspd_quarter_dumps d,
                       jsonb_array_elements(d.features_json) AS feat(value)
                 WHERE d.quarter_cad = :q
                   AND feat.value->>'layer' = 'engineering_structures'
                   AND (feat.value->'geometry') IS NOT NULL
                   AND feat.value->>'geometry' != 'null'
                   AND ST_DWithin(
                       ST_Transform(
                           ST_SetSRID(
                               ST_GeomFromGeoJSON(feat.value->>'geometry'),
                               3857
                           ),
                           4326
                       )::geography,
                       ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography,
                       :radius_m
                   )
                 ORDER BY distance_m ASC
                 LIMIT 20
                """
            ),
            {"q": quarter, "wkt": parcel_wkt, "radius_m": _ENGINEERING_RADIUS_M},
        ).fetchall()

        result: list[dict[str, Any]] = []
        for r in rows:
            props: dict[str, Any] = r[0] if isinstance(r[0], dict) else {}
            distance_m = float(r[1]) if r[1] is not None else None
            name = props.get("name") or props.get("object_name")
            obj_type = props.get("object_type") or props.get("type_zone")
            result.append(
                {
                    "name": name,
                    "type": obj_type,
                    "distance_m": round(distance_m) if distance_m is not None else None,
                    "raw_props": props,
                }
            )
        return result
    except Exception as e:
        logger.warning("nspd engineering query failed for quarter=%s: %s", quarter, e)
        return []


# ── Risk zones (issue #94 TIER 3) ────────────────────────────────────────────

# Human-readable subtype labels for risk layer keys.
# Keys match NSPDClient.QUARTER_RISK_LAYERS short_names.
_RISK_SUBTYPE_LABELS: dict[str, str] = {
    "flooding_underground": "Подтопление",
    "flooding": "Затопление",
    "swampification": "Заболачивание",
    "landslide": "Обвально-осыпные процессы",
    "abrasion": "Абразия",
    "erosion_water": "Водная эрозия",
    "erosion_linear": "Линейная эрозия",
    "erosion_wind": "Ветровая эрозия",
    "desertification": "Опустынивание",
    "clutter": "Захламление",
    "burns": "Гари",
}


def _extract_features_by_layer(
    features: list[dict[str, Any]],
    layer_prefix: str,
) -> list[dict[str, Any]]:
    """Generic filter: вернуть features у которых layer начинается с layer_prefix.

    Предназначен для re-use в PR 2+ (opportunity layers, etc.).

    Args:
        features: list of feature dicts из features_json (уже загруженных в Python).
        layer_prefix: префикс для фильтрации, e.g. 'risk_', 'zouit_'.

    Returns:
        Подсписок features где feat['layer'].startswith(layer_prefix).
    """
    return [
        f
        for f in features
        if isinstance(f.get("layer"), str) and f["layer"].startswith(layer_prefix)
    ]


def _get_risk_zones(
    db: Session,
    quarter: str,
    parcel_wkt: str,
    layer_counts: dict[str, int] | None = None,
) -> list[dict[str, Any]]:
    """TIER 3: риск-зоны НСПД (11 слоёв), пересекающие участок.

    Возвращает список риск-зон с полями:
        layer, subtype, geom_wkt, intersection_area_sqm.

    layer_counts — денормализованные счётчики. Если risks_count == 0 —
    пропускаем heavy jsonb_array_elements scan.
    """
    if layer_counts is not None and layer_counts.get("risks_count", 1) == 0:
        return []
    try:
        rows = db.execute(
            text(
                """
                SELECT feat.value->>'layer'         AS layer,
                       feat.value->'properties'     AS props,
                       ST_AsText(
                           ST_Transform(
                               ST_SetSRID(
                                   ST_GeomFromGeoJSON(feat.value->>'geometry'),
                                   3857
                               ),
                               4326
                           )
                       )                            AS geom_wkt,
                       ST_Area(
                           ST_Intersection(
                               ST_Transform(
                                   ST_SetSRID(
                                       ST_GeomFromGeoJSON(feat.value->>'geometry'),
                                       3857
                                   ),
                                   4326
                               )::geography,
                               ST_GeomFromText(:wkt, 4326)::geography
                           )
                       )                            AS intersection_area_sqm
                  FROM nspd_quarter_dumps d,
                       jsonb_array_elements(d.features_json) AS feat(value)
                 WHERE d.quarter_cad = :q
                   AND feat.value->>'layer' LIKE 'risk_%'
                   AND (feat.value->'geometry') IS NOT NULL
                   AND feat.value->>'geometry' != 'null'
                   AND ST_Intersects(
                       ST_Transform(
                           ST_SetSRID(
                               ST_GeomFromGeoJSON(feat.value->>'geometry'),
                               3857
                           ),
                           4326
                       ),
                       ST_GeomFromText(:wkt, 4326)
                   )
                 ORDER BY layer, intersection_area_sqm DESC NULLS LAST
                """
            ),
            {"q": quarter, "wkt": parcel_wkt},
        ).fetchall()

        result: list[dict[str, Any]] = []
        for r in rows:
            layer: str = r[0] or ""
            props: dict[str, Any] = r[1] if isinstance(r[1], dict) else {}
            geom_wkt_val: str | None = r[2]
            raw_area: Any = r[3]
            intersection_area_sqm: float | None = None
            if raw_area is not None:
                try:
                    val = float(raw_area)
                    intersection_area_sqm = round(val, 1) if not math.isnan(val) else None
                except (TypeError, ValueError):
                    pass

            # Извлечь short_name из "risk_<short_name>" для человекочитаемого subtype
            short_name = layer.removeprefix("risk_")
            # Попытка взять subtype из properties, затем из mapping
            subtype = (
                props.get("type_zone")
                or props.get("subcategory")
                or props.get("name")
                or _RISK_SUBTYPE_LABELS.get(short_name)
            )
            if subtype is not None:
                subtype = str(subtype)

            result.append(
                {
                    "layer": layer,
                    "subtype": subtype,
                    "geom_wkt": geom_wkt_val,
                    "intersection_area_sqm": intersection_area_sqm,
                }
            )
        return result
    except Exception as e:
        logger.warning("nspd risk zones query failed for quarter=%s: %s", quarter, e)
        return []


# ── Connection-points lookup (issue #115) ────────────────────────────────────


def get_connection_points(db: Session, cad_num: str, radius_m: int = 500) -> dict[str, Any]:
    """Получить точки инженерных подключений в radius_m от boundary участка.

    Источник: `nspd_quarter_dumps.features_json` для квартала cad_num —
    layers `engineering_structures` (NSPD cat 36328, ТП/ЦТП/насосные/опоры ЛЭП)
    и `zouit_engineering` (NSPD cat 37578 — охранные зоны инжен.коммуникаций).

    Args:
        db: SQLAlchemy session.
        cad_num: кадастровый номер участка (e.g. '66:41:0204016:10').
        radius_m: радиус поиска в метрах от boundary участка (default 500).

    Returns:
        {
          "engineering_structures": [...],
          "zouit_engineering_overlaps": [...],
          "summary": {...},
          "dump_available": bool,
          "dump_fetched_at": str | None,
        }

    Raises:
        ValueError: если parcel не найден в cad_parcels_geom / cad_quarters_geom.
    """
    quarter = derive_quarter_cad(cad_num)
    if quarter is None:
        raise ValueError(f"Невалидный формат кадастрового номера: {cad_num!r}")

    # Получаем WKT геометрию участка (boundary, не centroid)
    parcel_wkt = _get_parcel_wkt(db, cad_num)
    if parcel_wkt is None:
        raise ValueError(f"Участок {cad_num!r} не найден в БД")

    # Проверяем наличие дампа
    dump_row = db.execute(
        text(
            """
            SELECT fetched_at_utc, total_features
              FROM nspd_quarter_dumps
             WHERE quarter_cad = :q
             ORDER BY fetched_at_utc DESC
             LIMIT 1
            """
        ),
        {"q": quarter},
    ).first()

    if dump_row is None:
        _trigger_harvest(quarter)
        return {
            "engineering_structures": [],
            "zouit_engineering_overlaps": [],
            "summary": {
                "nearest_structure_distance_m": None,
                "in_protection_zone": False,
                "protection_zones_intersecting": 0,
                "total_structures_in_radius": 0,
            },
            "dump_available": False,
            "dump_fetched_at": None,
        }

    fetched_at = dump_row[0]
    if fetched_at is not None and getattr(fetched_at, "isoformat", None):
        dump_fetched_at: str | None = fetched_at.isoformat()
    else:
        dump_fetched_at = str(fetched_at) if fetched_at is not None else None

    structures = _get_engineering_structures_by_boundary(db, quarter, parcel_wkt, radius_m)
    zouit_overlaps = _get_zouit_engineering_overlaps(db, quarter, parcel_wkt)

    nearest_dist: float | None = structures[0]["distance_to_boundary_m"] if structures else None
    protection_count = len(zouit_overlaps)

    return {
        "engineering_structures": structures,
        "zouit_engineering_overlaps": zouit_overlaps,
        "summary": {
            "nearest_structure_distance_m": nearest_dist,
            "in_protection_zone": protection_count > 0,
            "protection_zones_intersecting": protection_count,
            "total_structures_in_radius": len(structures),
        },
        "dump_available": True,
        "dump_fetched_at": dump_fetched_at,
    }


def _get_parcel_wkt(db: Session, cad_num: str) -> str | None:
    """Получить WKT геометрию участка из cad_parcels_geom или fallback источников.

    Игнорирует строки с geom IS NULL (data quality) — иначе ST_AsText(NULL)
    вернёт SQL NULL → str(None) = "None" → ST_GeomFromText упадёт, ошибка
    тихо проглотится в caller'е, клиент получит пустые массивы без причины.
    """
    row = db.execute(
        text(
            """
            SELECT ST_AsText(geom) AS wkt
              FROM cad_parcels_geom
             WHERE cad_num = :c AND geom IS NOT NULL
             LIMIT 1
            """
        ),
        {"c": cad_num},
    ).first()
    if row is not None and row[0] is not None:
        return str(row[0])

    # Fallback: кварталы (более крупный объект — менее точно, но лучше чем ничего).
    # cad_quarters_geom хранит 3-сегментные ключи (66:41:0204016), не 4-сегментные.
    quarter_key = derive_quarter_cad(cad_num)
    row = db.execute(
        text(
            """
            SELECT ST_AsText(geom) AS wkt
              FROM cad_quarters_geom
             WHERE cad_number = :c AND geom IS NOT NULL
             LIMIT 1
            """
        ),
        {"c": quarter_key},
    ).first()
    if row is not None and row[0] is not None:
        return str(row[0])

    return None


def _get_engineering_structures_by_boundary(
    db: Session,
    quarter: str,
    parcel_wkt: str,
    radius_m: int,
) -> list[dict[str, Any]]:
    """Engineering structures из dump в radius_m от boundary участка.

    Использует ST_Distance к boundary (не centroid) для корректного расстояния.
    Geometry трансформируется из EPSG:3857 (хранение dump) → 4326.
    """
    try:
        rows = db.execute(
            text(
                """
                SELECT feat.value->'properties'         AS props,
                       feat.value->>'geometry'           AS geom_json,
                       ST_Distance(
                           ST_Transform(
                               ST_SetSRID(
                                   ST_GeomFromGeoJSON(feat.value->>'geometry'),
                                   3857
                               ),
                               4326
                           )::geography,
                           ST_GeomFromText(:wkt, 4326)::geography
                       ) AS distance_m
                  FROM nspd_quarter_dumps d,
                       jsonb_array_elements(d.features_json) AS feat(value)
                 WHERE d.quarter_cad = :q
                   AND feat.value->>'layer' = 'engineering_structures'
                   AND (feat.value->'geometry') IS NOT NULL
                   AND feat.value->>'geometry' != 'null'
                   AND ST_DWithin(
                       ST_Transform(
                           ST_SetSRID(
                               ST_GeomFromGeoJSON(feat.value->>'geometry'),
                               3857
                           ),
                           4326
                       )::geography,
                       ST_GeomFromText(:wkt, 4326)::geography,
                       :radius_m
                   )
                 ORDER BY distance_m ASC
                 LIMIT 50
                """
            ),
            {"q": quarter, "wkt": parcel_wkt, "radius_m": radius_m},
        ).fetchall()
    except (OperationalError, ProgrammingError) as e:
        logger.warning(
            "engineering_structures query failed for quarter=%s: %s",
            quarter,
            e,
        )
        return []

    result: list[dict[str, Any]] = []
    for r in rows:
        props: dict[str, Any] = r[0] if isinstance(r[0], dict) else {}
        geom_raw: str | None = r[1]
        distance_m = float(r[2]) if r[2] is not None else 0.0

        # Попытка распарсить geometry как dict для GeoJSON поля
        geom_dict: dict[str, Any] = {}
        if geom_raw:
            try:
                geom_dict = json.loads(geom_raw)
            except (ValueError, json.JSONDecodeError):
                geom_dict = {}

        result.append(
            {
                "name": props.get("name") or props.get("object_name"),
                "type": props.get("purpose") or props.get("object_type") or props.get("type_zone"),
                "cad_num": props.get("cad_num") or props.get("cadastral_number"),
                "distance_to_boundary_m": round(distance_m, 1),
                "geometry_geojson": geom_dict,
                "readable_address": props.get("readable_address") or props.get("address"),
                "raw_props": props,
                "source": "nspd_36328",
            }
        )
    return result


def _get_zouit_engineering_overlaps(
    db: Session,
    quarter: str,
    parcel_wkt: str,
) -> list[dict[str, Any]]:
    """ZOUIT engineering (cat 37578) — охранные зоны, пересекающие участок.

    Использует слой 'zouit_engineering' из dump.
    """
    try:
        rows = db.execute(
            text(
                """
                SELECT feat.value->'properties'    AS props,
                       feat.value->>'geometry'      AS geom_json
                  FROM nspd_quarter_dumps d,
                       jsonb_array_elements(d.features_json) AS feat(value)
                 WHERE d.quarter_cad = :q
                   AND feat.value->>'layer' = 'zouit_engineering'
                   AND (feat.value->'geometry') IS NOT NULL
                   AND feat.value->>'geometry' != 'null'
                   AND ST_Intersects(
                       ST_Transform(
                           ST_SetSRID(
                               ST_GeomFromGeoJSON(feat.value->>'geometry'),
                               3857
                           ),
                           4326
                       ),
                       ST_GeomFromText(:wkt, 4326)
                   )
                 LIMIT 100
                """
            ),
            {"q": quarter, "wkt": parcel_wkt},
        ).fetchall()
    except (OperationalError, ProgrammingError) as e:
        logger.warning(
            "zouit_engineering query failed for quarter=%s: %s",
            quarter,
            e,
        )
        return []

    result: list[dict[str, Any]] = []
    for r in rows:
        props: dict[str, Any] = r[0] if isinstance(r[0], dict) else {}
        geom_raw: str | None = r[1]

        geom_dict: dict[str, Any] = {}
        if geom_raw:
            try:
                geom_dict = json.loads(geom_raw)
            except (ValueError, json.JSONDecodeError):
                geom_dict = {}

        subcategory_raw = props.get("subcategory")
        subcategory: int | None = None
        if subcategory_raw is not None:
            try:
                subcategory = int(subcategory_raw)
            except (ValueError, TypeError):
                subcategory = None

        result.append(
            {
                "reg_numb_border": props.get("reg_numb_border"),
                "type_zone": props.get("type_zone") or props.get("zone_name"),
                "subcategory": subcategory,
                "intersects_parcel": True,
                "geometry_geojson": geom_dict,
                "raw_props": props,
                "source": "nspd_37578",
            }
        )
    return result


# ── Harvest trigger ───────────────────────────────────────────────────────────


def _trigger_harvest(quarter: str) -> bool:
    """Fire-and-forget harvest_quarter.apply_async(). Возвращает True если enqueue OK.

    Ленивый импорт чтобы избежать circular import (tasks → services → tasks).

    Known limitation: burst из N concurrent analyze_parcel запросов на один
    и тот же ещё не закешированный квартал может поставить N одинаковых задач
    в очередь (нет дедупликации). UPSERT в harvest_quarter идемпотентен, поэтому
    данные не портятся, но WAF traffic тратится впустую.
    TODO: добавить Redis SETNX lock с TTL перед apply_async — отдельная issue/PR.
    """
    try:
        from app.workers.tasks.nspd_sync import harvest_quarter

        harvest_quarter.apply_async(args=[quarter], kwargs={"region_code": 66})
        logger.info("quarter dump harvest triggered for quarter=%s", quarter)
        return True
    except Exception as e:
        logger.warning("failed to trigger harvest for quarter=%s: %s", quarter, e)
        return False