diff --git a/backend/app/workers/beat_schedule.py b/backend/app/workers/beat_schedule.py index 58378bf7..b332fd2c 100644 --- a/backend/app/workers/beat_schedule.py +++ b/backend/app/workers/beat_schedule.py @@ -322,4 +322,14 @@ def build_beat_schedule() -> dict: "options": {"queue": "celery"}, } + # ИРД-overlay harvest (#1067 B5): grid-walk готовых НСПД ИРД-слоёв по кварталам ЕКБ → + # ird_overlays (м.132). Понедельник 05:00 МСК — до supply-layers (06:00), не конкурируя + # с scrape_kn (04:15). Геометрия зон стабильна, еженедельно достаточно. Техническая, + # не в job_settings (как cbr/poi/supply-layers). + schedule["ird-harvest-weekly"] = { + "task": "tasks.ird_harvest.harvest_ird_overlays", + "schedule": _parse_cron("0 5 * * mon"), + "options": {"queue": "celery"}, + } + return schedule diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index 59189d4d..cd51c76a 100644 --- a/backend/app/workers/celery_app.py +++ b/backend/app/workers/celery_app.py @@ -63,6 +63,7 @@ celery_app = Celery( "app.workers.tasks.cbr_macro_sync", "app.workers.tasks.supply_layers_refresh", "app.workers.tasks.forecast", + "app.workers.tasks.ird_harvest", ], ) celery_app.conf.timezone = "Europe/Moscow" diff --git a/backend/app/workers/tasks/ird_harvest.py b/backend/app/workers/tasks/ird_harvest.py new file mode 100644 index 00000000..c721ed83 --- /dev/null +++ b/backend/app/workers/tasks/ird_harvest.py @@ -0,0 +1,221 @@ +"""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() diff --git a/backend/tests/workers/test_ird_harvest.py b/backend/tests/workers/test_ird_harvest.py new file mode 100644 index 00000000..1d075055 --- /dev/null +++ b/backend/tests/workers/test_ird_harvest.py @@ -0,0 +1,133 @@ +"""Тесты harvest_ird_overlays (#1067 B5) — НСПД ИРД-harvest → ird_overlays. + +Сеть/БД не дёргаются: мокаем NSPDClient + SessionLocal, проверяем grid-walk loop, +skip битых фич и upsert-параметры. +""" + +from __future__ import annotations + +from contextlib import contextmanager +from typing import Any + +from app.workers.tasks import ird_harvest + + +class _Feat: + def __init__(self, geometry: dict[str, Any] | None, properties: dict[str, Any]) -> None: + self.geometry = geometry + self.properties = properties + + +class _SearchRes: + def __init__(self, feat: _Feat | None) -> None: + self.first = feat + + +class _FakeClient: + """NSPDClient-двойник: квартал → полигон (3857), grid-walk → 1 фича на слой.""" + + def __init__(self, *a: Any, **kw: Any) -> None: + self.grid_calls: list[int] = [] + + def search_by_cad(self, cad: str, thematic_id: int = 1) -> _SearchRes: + ring = [[0, 0], [100, 0], [100, 100], [0, 100], [0, 0]] + geom = {"type": "Polygon", "coordinates": [ring]} + return _SearchRes(_Feat(geom, {"cad": cad})) + + def get_features_in_bbox_grid(self, layer_id: int, bbox: Any, **kw: Any) -> list[_Feat]: + self.grid_calls.append(layer_id) + return [ + _Feat( + {"type": "Polygon", "coordinates": [[[1, 1], [2, 1], [2, 2], [1, 1]]]}, + { + "geom_data_id": 1000 + layer_id, + "reg_numb_border": "66:41-6.1", + "category": 36940, + "subcategory": 5, + }, + ) + ] + + +class _FakeDB: + """Session-двойник: пишет execute-вызовы, begin_nested как no-op CM.""" + + 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(ird_harvest, "NSPDClient", lambda *a, **kw: client) + monkeypatch.setattr(ird_harvest, "SessionLocal", lambda: db) + return client + + +def test_harvest_explicit_quarters_upserts_all_layers(monkeypatch: Any) -> None: + """1 квартал × N ИРД-слоёв → upsert по фиче на слой; commit + close.""" + db = _FakeDB() + client = _patch(monkeypatch, db) + n_layers = len(ird_harvest.IRD_HARVEST_LAYER_KEYS) + + result = ird_harvest.harvest_ird_overlays(quarters=["66:41:0204016"]) + + assert result == {"quarters": 1, "features": n_layers} + assert len(client.grid_calls) == n_layers # grid-walk по каждому слою + assert len(db.executed) == n_layers # upsert по фиче на слой + assert db.committed is True + assert db.closed is True + # параметры upsert: geom как JSON-строка, layer_kind из ключей + _, params = db.executed[0] + assert params is not None + assert params["layer_kind"] in ird_harvest.IRD_HARVEST_LAYER_KEYS + assert isinstance(params["geojson"], str) + assert params["reg_numb_border"] == "66:41-6.1" + + +def test_harvest_skips_empty_quarter(monkeypatch: Any) -> None: + """Квартал без геометрии в НСПД → 0 кварталов/фич, без падения.""" + db = _FakeDB() + client = _patch(monkeypatch, db) + monkeypatch.setattr(client, "search_by_cad", lambda cad, thematic_id=1: _SearchRes(None)) + + result = ird_harvest.harvest_ird_overlays(quarters=["66:41:9999999"]) + + assert result == {"quarters": 0, "features": 0} + assert db.executed == [] + assert db.committed is True + + +def test_upsert_feature_skips_missing_geom_data_id(monkeypatch: Any) -> None: + """Фича без geom_data_id / geometry → не пишется (defensive).""" + db = _FakeDB() + no_id = _Feat({"type": "Polygon", "coordinates": []}, {"reg_numb_border": "x"}) + no_geom = _Feat(None, {"geom_data_id": 5}) + assert ( + ird_harvest._upsert_feature( + db, layer_id=37577, layer_kind="zouit_okn", feature=no_id, fetched_at="t" + ) + is False + ) + assert ( + ird_harvest._upsert_feature( + db, layer_id=37577, layer_kind="zouit_okn", feature=no_geom, fetched_at="t" + ) + is False + ) + assert db.executed == []