feat(sf): ППТ/ПМТ WFS ingest → planning_projects (future-supply, #1085)

backend/app/workers/beat_schedule.py

@@ -341,4 +341,13 @@ def build_beat_schedule() -> dict:
         "options": {"queue": "celery"},
     }
 
+    # ППТ/ПМТ harvest (#1085): WFS-фетч геопортала ЕКБ по bbox агломерации → planning_projects.
+    # 1-е число месяца 06:00 МСК — после opportunity (05:30). Документы планировки стабильны,
+    # ежемесячно достаточно. WFS BBOX (не grid-walk) — один запрос на слой.
+    schedule["planning-harvest-monthly"] = {
+        "task": "tasks.planning_harvest.harvest_planning_projects",
+        "schedule": _parse_cron("0 6 1 * *"),
+        "options": {"queue": "celery"},
+    }
+
     return schedule

backend/app/workers/celery_app.py

@@ -65,6 +65,7 @@ celery_app = Celery(
         "app.workers.tasks.forecast",
         "app.workers.tasks.ird_harvest",
         "app.workers.tasks.opportunity_harvest",
+        "app.workers.tasks.planning_harvest",
     ],
 )
 celery_app.conf.timezone = "Europe/Moscow"

backend/app/workers/tasks/planning_harvest.py

@@ -0,0 +1,157 @@
+"""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()

backend/tests/workers/test_planning_harvest.py

@@ -0,0 +1,111 @@
+"""Тесты harvest_planning_projects (#1085) — геопортал-WFS ППТ/ПМТ → planning_projects.
+
+Сеть/БД не дёргаются: мокаем EKBGeoportalClient + SessionLocal, проверяем фетч обоих слоёв,
+project_type-маппинг, skip битых фич, нормализацию года и upsert-параметры.
+"""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+from typing import Any
+
+from app.workers.tasks import planning_harvest
+
+
+class _Feat:
+    def __init__(self, geometry: dict[str, Any] | None, properties: dict[str, Any]) -> None:
+        self.geometry = geometry
+        self.properties = properties
+
+
+class _FakeClient:
+    """EKBGeoportalClient-двойник: layer → 1 фича на слой."""
+
+    def __init__(self, *a: Any, **kw: Any) -> None:
+        self.bbox_calls: list[str] = []
+
+    def features_in_bbox(self, layer: str, bbox: Any) -> list[_Feat]:
+        self.bbox_calls.append(layer)
+        return [
+            _Feat(
+                {"type": "Polygon", "coordinates": [[[60, 56], [60.1, 56], [60, 56.1], [60, 56]]]},
+                {
+                    "key": 1000010173321902 if layer == "ppt" else 1000010209297292,
+                    "full_name": f"ПАГЕ № 1 ({layer})",
+                    "doc_status": 1,
+                    "doc_status_name": "действующий",
+                    "project_name": "Тестовый проект",
+                    "dmd_actual_year": "2021",
+                },
+            )
+        ]
+
+
+class _FakeDB:
+    def __init__(self) -> None:
+        self.executed: list[tuple[Any, dict[str, Any] | None]] = []
+        self.committed = False
+        self.closed = False
+
+    @contextmanager
+    def begin_nested(self) -> Any:
+        yield
+
+    def execute(self, sql: Any, params: dict[str, Any] | None = None) -> Any:
+        self.executed.append((sql, params))
+        return None
+
+    def commit(self) -> None:
+        self.committed = True
+
+    def close(self) -> None:
+        self.closed = True
+
+
+def _patch(monkeypatch: Any, db: _FakeDB) -> _FakeClient:
+    client = _FakeClient()
+    monkeypatch.setattr(planning_harvest, "EKBGeoportalClient", lambda *a, **kw: client)
+    monkeypatch.setattr(planning_harvest, "SessionLocal", lambda: db)
+    return client
+
+
+def test_harvest_both_layers(monkeypatch: Any) -> None:
+    """ППТ + ПМТ → upsert по фиче на слой; project_type-маппинг; year нормализован в int."""
+    db = _FakeDB()
+    client = _patch(monkeypatch, db)
+
+    result = planning_harvest.harvest_planning_projects()
+
+    assert result == {"features": 2}
+    assert sorted(client.bbox_calls) == ["pmt", "ppt"]
+    assert len(db.executed) == 2
+    types = {params["project_type"] for _, params in db.executed if params}
+    assert types == {"ppt", "pmt"}
+    _, p0 = db.executed[0]
+    assert p0 is not None
+    assert p0["dmd_actual_year"] == 2021  # строка "2021" → int
+    assert p0["doc_status"] == "1"  # приведён к строке
+    assert isinstance(p0["geojson"], str)
+
+
+def test_skips_feature_without_key_or_geom() -> None:
+    """Фича без key / geometry → не пишется (defensive)."""
+    db = _FakeDB()
+    no_key = _Feat({"type": "Polygon", "coordinates": []}, {"full_name": "x"})
+    no_geom = _Feat(None, {"key": 5})
+    assert (
+        planning_harvest._upsert_feature(db, project_type="ppt", feature=no_key, fetched_at="t")
+        is False
+    )
+    assert (
+        planning_harvest._upsert_feature(db, project_type="pmt", feature=no_geom, fetched_at="t")
+        is False
+    )
+    assert db.executed == []
+
+
+def test_year_normalization() -> None:
+    assert planning_harvest._to_int("2021") == 2021
+    assert planning_harvest._to_int(2021) == 2021
+    assert planning_harvest._to_int(None) is None
+    assert planning_harvest._to_int("н/д") is None

data/sql/134_planning_projects.sql

@@ -0,0 +1,78 @@
+-- 134_planning_projects.sql
+-- #1085 (Site Finder v2 / future-supply «GG-форсайт») — store ППТ/ПМТ ЕКБ (геометрия + реквизиты).
+--
+-- Future-supply: что/где утверждено к застройке. Источник — ЕКБ-геопортал GeoServer WFS
+-- (ekb_geoportal_client.py, #1083): слои portal_geo_dmd_prj_plan (ППТ — проекты планировки) и
+-- portal_geo_dmd_prj_mej (ПМТ — проекты межевания). Каждая фича — утверждённый документ
+-- планировки с геометрией охвата + doc_status/год (live-verified 2026-06-06, vault research/
+-- EKB_Geoportal_WFS_Construction_Intel_Jun06).
+--
+-- ОТДЕЛЬНАЯ таблица (не ird_overlays м.132): ird_overlays — НСПД-ограничения участка (per-feature
+-- ЗОУИТ/ПЗЗ/ОКН + reg_numb_border + category/subcategory), а тут документ-центричная модель
+-- геопортала (doc_status / dmd_actual_year / project_name / full_name) из другого источника (WFS,
+-- уже EPSG:4326). Смешивать схемы — натяжка; держим раздельно, analyze читает overlap отдельно.
+--
+-- Что отдаёт WFS-фича (DescribeFeatureType + live): key (стабильный id портала), full_name
+-- (документ-описание), doc_full_name/doc_status/doc_status_name (статус утверждения),
+-- dmd_actual_year (год актуальности), project_name, geoloc (геометрия 4326).
+--
+-- Зерно (grain): одна строка = одна WFS-фича (один документ планировки) в слое. Дедуп-ключ —
+--   (project_type, source_key): source_key = портальный `key`; project_type 'ppt'|'pmt'.
+--   re-harvest UPSERT-ит (latest wins): документы стабильны, обновление = свежий fetched_at.
+--
+-- ТЭП/очередность из PDF пояснительных записок (pdfplumber) — отдельный проход (#1085 follow-up);
+--   колонки tep/staging добавятся ALTER-ом позже (тут только WFS-слой: геометрия + статус).
+--
+-- Deploy: auto-applied deploy.yml через _schema_migrations (ровно один раз по NN). Idempotent:
+--   CREATE TABLE/INDEX IF NOT EXISTS + guarded DO для UNIQUE (прецедент м.125/132/133).
+
+BEGIN;
+
+-- ── 1. Таблица planning_projects ───────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS planning_projects (
+    id               BIGSERIAL PRIMARY KEY,
+    project_type     TEXT NOT NULL,            -- 'ppt' (проект планировки) | 'pmt' (проект межевания)
+    source_key       BIGINT NOT NULL,          -- стабильный `key` фичи геопортала (дедуп внутри типа)
+    full_name        TEXT,                     -- описание документа (ПАГЕ № / «Об утверждении ...»)
+    doc_full_name    TEXT,                     -- doc_full_name WFS
+    doc_status       TEXT,                     -- doc_status (код/строка статуса утверждения)
+    doc_status_name  TEXT,                     -- читаемый статус ('действующий'/'утверждён'/...)
+    project_name     TEXT,                     -- project_name WFS
+    dmd_actual_year  INTEGER,                  -- год актуальности документа (dmd_actual_year)
+    geom             geometry(Geometry, 4326) NOT NULL,  -- охват документа (WFS geoloc, уже 4326)
+    raw_props        JSONB,                    -- полный properties фичи (forward-compat / ТЭП-резолв)
+    fetched_at       timestamptz NOT NULL DEFAULT now(),  -- когда harvest снял (freshness)
+    CONSTRAINT planning_projects_type_chk CHECK (project_type IN ('ppt', 'pmt')),
+    CONSTRAINT planning_projects_source_key_chk CHECK (source_key > 0)
+);
+
+COMMENT ON TABLE planning_projects IS
+    'ППТ/ПМТ ЕКБ (#1085, future-supply GG-форсайт): утверждённые проекты планировки/межевания из '
+    'ЕКБ-геопортала WFS (portal_geo_dmd_prj_plan/_mej). 1 строка = 1 WFS-фича (документ). UPSERT '
+    'harvest-воркером по uq_planning_projects (project_type, source_key). analyze читает overlap '
+    'участка через ST_Intersects(geom). ТЭП/очередность из PDF — отдельный проход (follow-up).';
+COMMENT ON COLUMN planning_projects.source_key IS 'Стабильный `key` фичи ЕКБ-геопортала — дедуп-ключ внутри project_type';
+COMMENT ON COLUMN planning_projects.doc_status_name IS 'Читаемый статус утверждения документа планировки (действующий/утверждён/...)';
+
+-- ── 2. Дедуп/UPSERT-ключ (project_type, source_key) ────────────────────────────
+DO $$
+BEGIN
+    IF NOT EXISTS (
+        SELECT 1 FROM pg_constraint WHERE conname = 'uq_planning_projects'
+    ) THEN
+        ALTER TABLE planning_projects
+            ADD CONSTRAINT uq_planning_projects
+            UNIQUE (project_type, source_key);
+    END IF;
+END
+$$;
+
+-- ── 3. Индексы ─────────────────────────────────────────────────────────────────
+-- Главный read-path: spatial overlap участка (analyze: ST_Intersects(parcel.geom, pp.geom)).
+CREATE INDEX IF NOT EXISTS idx_planning_projects_geom
+    ON planning_projects USING GIST (geom);
+-- Фильтр по типу проекта (выборка «все ППТ» / «все ПМТ»).
+CREATE INDEX IF NOT EXISTS idx_planning_projects_type
+    ON planning_projects (project_type);
+
+COMMIT;