"""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 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_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_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 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 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_dump": dump_meta, } layer_counts = { "territorial_zones_count": territorial_zones_count, "zouit_count": zouit_count, "engineering_count": engineering_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) return { "nspd_zoning": nspd_zoning, "nspd_zouit_overlaps": nspd_zouit, "nspd_engineering_nearby": nspd_engineering, "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 [] # ── 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