From b5c61d295b8ef4d02fc2d903687e30e0192f3027 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sat, 6 Jun 2026 23:18:23 +0300 Subject: [PATCH] =?UTF-8?q?feat(sf):=20=D0=9F=D0=9F=D0=A2/=D0=9F=D0=9C?= =?UTF-8?q?=D0=A2=20WFS=20ingest=20=E2=86=92=20planning=5Fprojects=20(futu?= =?UTF-8?q?re-supply,=20#1085)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/workers/beat_schedule.py | 9 + backend/app/workers/celery_app.py | 1 + backend/app/workers/tasks/planning_harvest.py | 157 ++++++++++++++++++ .../tests/workers/test_planning_harvest.py | 111 +++++++++++++ data/sql/134_planning_projects.sql | 78 +++++++++ 5 files changed, 356 insertions(+) create mode 100644 backend/app/workers/tasks/planning_harvest.py create mode 100644 backend/tests/workers/test_planning_harvest.py create mode 100644 data/sql/134_planning_projects.sql diff --git a/backend/app/workers/beat_schedule.py b/backend/app/workers/beat_schedule.py index 6af8dedd..a3057f27 100644 --- a/backend/app/workers/beat_schedule.py +++ b/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 diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index 849917f1..6d9aba2b 100644 --- a/backend/app/workers/celery_app.py +++ b/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" diff --git a/backend/app/workers/tasks/planning_harvest.py b/backend/app/workers/tasks/planning_harvest.py new file mode 100644 index 00000000..53f02252 --- /dev/null +++ b/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() diff --git a/backend/tests/workers/test_planning_harvest.py b/backend/tests/workers/test_planning_harvest.py new file mode 100644 index 00000000..6dc8ee98 --- /dev/null +++ b/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 diff --git a/data/sql/134_planning_projects.sql b/data/sql/134_planning_projects.sql new file mode 100644 index 00000000..aa218336 --- /dev/null +++ b/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;