gendesign/backend/app/workers/tasks/ird_harvest.py

"""Celery task: harvest ИРД-overlay геометрии из НСПД → ``ird_overlays`` (м.132).

#1067 (Site Finder v2 / слой ИРД для analyze, «GG-форсайт»), foundation B5. Для каждого
квартала ЕКБ (66:41:*) grid-walk'ит готовые НСПД-слои ИРД и UPSERT-ит фичи в ``ird_overlays``
по ключу ``uq_ird_overlays_feature`` (source_layer_id, geom_data_id). analyze читает overlap
участка через ``ST_Intersects``.

НСПД отдаёт по фиче ТОЛЬКО геометрию (EPSG:3857) + reg_numb_border + category/subcategory
(разведка 2026-06-06, vault research/NSPD_IRD_Layers_Automation_Recon_Jun06). content_* (тип/
ВРИ/параметры) дорезолвятся отдельно из ЕКБ-геопортала (C8/D9) и здесь НЕ трогаются.

Harvest-набор = готовые ИРД-слои (ID зашиты в nspd_client.LAYERS):
  ЗОУИТ 37577-37581 (ОКН/инж/природа/охрана/иные) · 875838 терзоны ПЗЗ (геометрия) ·
  875840 территории ОКН · 875845 ООПТ · 875866 лесничества · 875832/875835 береговые линии.

Mirror conventions (supply_layers_refresh.py / backend.md):
  • SessionLocal() + try/finally close; logger (не print).
  • SAVEPOINT per-row (``with db.begin_nested():``) — одна битая фича не валит батч; commit в конце.
  • CAST(:x AS type) — НИКОГДА :x::type (psycopg v3). Геометрия: ST_Transform(3857→4326).
  • ON CONFLICT обновляет геометрию/reg/category/subcategory/raw_props/fetched_at; content_*
    НЕ трогает (их пишет резолвер D9 — re-harvest не затирает резолвленный тип/регламент).
  • Сбой одного квартала/слоя не валит весь прогон.

Расписание — еженедельно (beat_schedule.py: ird-harvest-weekly). Геометрия зон стабильна,
чаще не нужно.
"""

from __future__ import annotations

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

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.db import SessionLocal
from app.services.scrapers.nspd_client import (
    LAYERS,
    NSPDClient,
    _geojson_bbox_3857,
)
from app.workers.celery_app import celery_app

logger = logging.getLogger(__name__)

# Готовые ИРД-слои для harvest — ключи nspd_client.LAYERS. layer_kind в ird_overlays = этот ключ.
IRD_HARVEST_LAYER_KEYS: tuple[str, ...] = (
    "zouit_okn",
    "zouit_engineering",
    "zouit_natural",
    "zouit_protected",
    "zouit_other",
    "territorial_zones",
    "okn_territory",
    "protected_areas",
    "forestries",
    "coast_lines_river",
    "coast_lines_sea",
)

# Источник кварталов ЕКБ: distinct quarter_cad_number из cad_parcels (66:41:*). quarter_cad_number
# деривится триггером (м.129). LIKE-паттерн — кадастровый район ЕКБ. CAST(:x AS type) psycopg v3.
_EKB_QUARTERS_SQL = text(
    """
    SELECT DISTINCT quarter_cad_number
    FROM cad_parcels
    WHERE quarter_cad_number LIKE CAST(:prefix AS text)
      AND quarter_cad_number IS NOT NULL
    ORDER BY quarter_cad_number
    """
)
_EKB_QUARTER_PREFIX = "66:41:%"

# UPSERT фичи. Геометрия НСПД (GeoJSON 3857) → ST_GeomFromGeoJSON → SetSRID 3857 → Transform 4326.
# ON CONFLICT (source_layer_id, geom_data_id): обновляем геометрию/реквизиты/fetched_at;
# content_* НЕ трогаем (их пишет резолвер D9).
_UPSERT_SQL = text(
    """
    INSERT INTO ird_overlays (
        source_layer_id, layer_kind, geom_data_id, reg_numb_border,
        category, subcategory, geom, raw_props, fetched_at
    ) VALUES (
        CAST(:source_layer_id AS integer),
        CAST(:layer_kind AS text),
        CAST(:geom_data_id AS bigint),
        CAST(:reg_numb_border AS text),
        CAST(:category AS integer),
        CAST(:subcategory AS integer),
        ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(CAST(:geojson AS text)), 3857), 4326),
        CAST(:raw_props AS jsonb),
        CAST(:fetched_at AS timestamptz)
    )
    ON CONFLICT (source_layer_id, geom_data_id) DO UPDATE SET
        layer_kind      = EXCLUDED.layer_kind,
        reg_numb_border = EXCLUDED.reg_numb_border,
        category        = EXCLUDED.category,
        subcategory     = EXCLUDED.subcategory,
        geom            = EXCLUDED.geom,
        raw_props       = EXCLUDED.raw_props,
        fetched_at      = EXCLUDED.fetched_at
    """
)


def _ekb_quarters(db: Session) -> list[str]:
    """Distinct кварталы ЕКБ (66:41:*) из cad_parcels."""
    rows = db.execute(_EKB_QUARTERS_SQL, {"prefix": _EKB_QUARTER_PREFIX}).fetchall()
    return [r[0] for r in rows if r[0]]


def _upsert_feature(
    db: Session,
    *,
    layer_id: int,
    layer_kind: str,
    feature: Any,
    fetched_at: str,
) -> bool:
    """UPSERT одной НСПД-фичи в ird_overlays под SAVEPOINT. True если записана.

    Пропускает фичи без geometry или без geom_data_id (дедуп-ключ) — defensive,
    одна битая не валит батч.
    """
    geom = feature.geometry
    props = feature.properties or {}
    geom_data_id = props.get("geom_data_id")
    if not geom or geom_data_id is None:
        return False
    opts = props.get("options") or {}
    reg = props.get("reg_numb_border") or opts.get("reg_numb_border") or props.get("descr")
    try:
        with db.begin_nested():
            db.execute(
                _UPSERT_SQL,
                {
                    "source_layer_id": layer_id,
                    "layer_kind": layer_kind,
                    "geom_data_id": int(geom_data_id),
                    "reg_numb_border": str(reg) if reg is not None else None,
                    "category": props.get("category"),
                    "subcategory": props.get("subcategory"),
                    "geojson": json.dumps(geom),
                    "raw_props": json.dumps(props, ensure_ascii=False),
                    "fetched_at": fetched_at,
                },
            )
        return True
    except Exception as exc:
        logger.warning(
            "ird_harvest: upsert failed layer=%s geom_data_id=%s: %s",
            layer_kind,
            geom_data_id,
            exc,
        )
        return False


@celery_app.task(name="tasks.ird_harvest.harvest_ird_overlays")
def harvest_ird_overlays(quarters: list[str] | None = None) -> dict[str, int]:
    """Harvest ИРД-слоёв НСПД по кварталам ЕКБ → upsert ird_overlays.

    Args:
        quarters: явный список cad-номеров кварталов (3-сегментные '66:41:NNNNNNN').
            None → берём distinct кварталы ЕКБ из cad_parcels.

    Returns:
        {'quarters': N, 'features': M} — обработано кварталов / записано фич.
    """
    fetched_at = datetime.now(UTC).isoformat()
    client = NSPDClient()
    db = SessionLocal()
    n_quarters = 0
    n_features = 0
    try:
        quarter_list = quarters if quarters is not None else _ekb_quarters(db)
        logger.info("ird_harvest: старт, кварталов=%d", len(quarter_list))
        for quarter in quarter_list:
            try:
                qres = client.search_by_cad(quarter, thematic_id=2)
                qfeat = qres.first
                if not qfeat or not qfeat.geometry:
                    logger.info("ird_harvest: квартал %s пуст в НСПД, скип", quarter)
                    continue
                bbox = _geojson_bbox_3857(qfeat.geometry)
                if bbox is None:
                    continue
                n_quarters += 1
                for layer_key in IRD_HARVEST_LAYER_KEYS:
                    layer_id = LAYERS.get(layer_key)
                    if layer_id is None:
                        logger.warning("ird_harvest: нет layer_id для %s", layer_key)
                        continue
                    try:
                        feats = client.get_features_in_bbox_grid(layer_id, bbox)
                    except Exception as exc:
                        logger.warning(
                            "ird_harvest: grid-walk failed quarter=%s layer=%s: %s",
                            quarter,
                            layer_key,
                            exc,
                        )
                        continue
                    for feat in feats:
                        if _upsert_feature(
                            db,
                            layer_id=layer_id,
                            layer_kind=layer_key,
                            feature=feat,
                            fetched_at=fetched_at,
                        ):
                            n_features += 1
            except Exception as exc:
                logger.warning("ird_harvest: квартал %s failed: %s", quarter, exc)
                continue
        db.commit()
        logger.info("ird_harvest: готово, кварталов=%d фич=%d", n_quarters, n_features)
        return {"quarters": n_quarters, "features": n_features}
    finally:
        db.close()