"""Celery task: harvest ППТ/ПМТ ЕКБ из геопортал-WFS → ``planning_projects`` (#1085). Future-supply: утверждённые проекты планировки (ППТ) и межевания (ПМТ) территории ЕКБ. Источник — ЕКБ-геопортал GeoServer WFS (``ekb_geoportal_client``, #1083): слои ``ppt`` (dmd_prj_plan) и ``pmt`` (dmd_prj_mej). Каждая фича — документ планировки с геометрией охвата (EPSG:4326) + doc_status / dmd_actual_year (live-verified 2026-06-06, vault EKB_Geoportal_WFS_Construction_Intel). Отличие от ИРД/opportunity-harvest (НСПД WMS grid-walk по кварталам): WFS BBOX отдаёт все фичи слоя в прямоугольнике ОДНИМ запросом (нет point-probe grid-walk), геометрия уже 4326 (нет ST_Transform). Поэтому проходим ОДИН bbox агломерации на слой. Mirror conventions (ird_harvest.py / backend.md): SessionLocal()+try/finally; SAVEPOINT per-row (``with db.begin_nested():``); CAST(:x AS type) (psycopg v3); сбой фичи/слоя не валит прогон. ТЭП/очередность из PDF пояснительных записок (pdfplumber) — отдельный проход (#1085 follow-up): тут только WFS-слой (геометрия + статус документа). Расписание — ежемесячно (документы стабильны). """ from __future__ import annotations import json import logging from datetime import UTC, datetime from sqlalchemy import text from sqlalchemy.orm import Session from app.core.db import SessionLocal from app.services.scrapers.ekb_geoportal_client import EKBFeature, EKBGeoportalClient from app.workers.celery_app import celery_app logger = logging.getLogger(__name__) # Слои геопортала для harvest — ключи WFS_LAYERS → project_type в planning_projects. PLANNING_LAYERS: dict[str, str] = {"ppt": "ppt", "pmt": "pmt"} # bbox агломерации ЕКБ (minlon, minlat, maxlon, maxlat) EPSG:4326 — охватывает город + пригороды. EKB_AGGLOMERATION_BBOX: tuple[float, float, float, float] = (60.0, 56.6, 61.1, 57.1) # UPSERT фичи. Геометрия WFS (GeoJSON 4326) → ST_GeomFromGeoJSON → SetSRID 4326 (уже 4326, не # Transform). ON CONFLICT (project_type, source_key): обновляем реквизиты/геометрию/fetched_at. _UPSERT_SQL = text( """ INSERT INTO planning_projects ( project_type, source_key, full_name, doc_full_name, doc_status, doc_status_name, project_name, dmd_actual_year, geom, raw_props, fetched_at ) VALUES ( CAST(:project_type AS text), CAST(:source_key AS bigint), CAST(:full_name AS text), CAST(:doc_full_name AS text), CAST(:doc_status AS text), CAST(:doc_status_name AS text), CAST(:project_name AS text), CAST(:dmd_actual_year AS integer), ST_SetSRID(ST_GeomFromGeoJSON(CAST(:geojson AS text)), 4326), CAST(:raw_props AS jsonb), CAST(:fetched_at AS timestamptz) ) ON CONFLICT (project_type, source_key) DO UPDATE SET full_name = EXCLUDED.full_name, doc_full_name = EXCLUDED.doc_full_name, doc_status = EXCLUDED.doc_status, doc_status_name = EXCLUDED.doc_status_name, project_name = EXCLUDED.project_name, dmd_actual_year = EXCLUDED.dmd_actual_year, geom = EXCLUDED.geom, raw_props = EXCLUDED.raw_props, fetched_at = EXCLUDED.fetched_at """ ) def _to_int(val: object) -> int | None: """Год актуальности может прийти строкой/числом/None — нормализуем в int|None.""" if val is None: return None try: return int(str(val).strip()) except (ValueError, TypeError): return None def _upsert_feature( db: Session, *, project_type: str, feature: EKBFeature, fetched_at: str ) -> bool: """UPSERT одной WFS-фичи в planning_projects под SAVEPOINT. True если записана. Пропускает фичи без geometry или без `key` (дедуп-ключ) — defensive, одна битая не валит батч. """ geom = feature.geometry props = feature.properties or {} source_key = props.get("key") if not geom or source_key is None: return False doc_status = props.get("doc_status") try: with db.begin_nested(): db.execute( _UPSERT_SQL, { "project_type": project_type, "source_key": int(source_key), "full_name": props.get("full_name"), "doc_full_name": props.get("doc_full_name"), "doc_status": str(doc_status) if doc_status is not None else None, "doc_status_name": props.get("doc_status_name"), "project_name": props.get("project_name"), "dmd_actual_year": _to_int(props.get("dmd_actual_year")), "geojson": json.dumps(geom), "raw_props": json.dumps(props, ensure_ascii=False), "fetched_at": fetched_at, }, ) return True except Exception as exc: logger.warning( "planning_harvest: upsert failed type=%s key=%s: %s", project_type, source_key, exc ) return False @celery_app.task(name="tasks.planning_harvest.harvest_planning_projects") def harvest_planning_projects( bbox: tuple[float, float, float, float] | None = None, ) -> dict[str, int]: """Harvest ППТ/ПМТ ЕКБ из геопортал-WFS по bbox → upsert planning_projects. Args: bbox: (minlon, minlat, maxlon, maxlat) EPSG:4326. None → bbox агломерации ЕКБ. Returns: {'features': M} — записано/обновлено документов планировки. """ fetched_at = datetime.now(UTC).isoformat() target_bbox = bbox if bbox is not None else EKB_AGGLOMERATION_BBOX client = EKBGeoportalClient() db = SessionLocal() n_features = 0 try: logger.info("planning_harvest: старт, bbox=%s", target_bbox) for layer_key, project_type in PLANNING_LAYERS.items(): try: feats = client.features_in_bbox(layer_key, target_bbox) except Exception as exc: logger.warning("planning_harvest: WFS failed layer=%s: %s", layer_key, exc) continue for feat in feats: if _upsert_feature( db, project_type=project_type, feature=feat, fetched_at=fetched_at ): n_features += 1 db.commit() logger.info("planning_harvest: готово, документов=%d", n_features) return {"features": n_features} finally: db.close()