feat(sf): ИРД-harvest Celery task → ird_overlays (#1067 B5)
All checks were successful
CI / changes (pull_request) Successful in 8s
CI / changes (push) Successful in 7s
CI / backend-tests (push) Successful in 5m51s
CI / backend-tests (pull_request) Successful in 5m45s

Periodic harvest готовых НСПД ИРД-слоёв (ЗОУИТ 37577-581, ПЗЗ-границы 875838,
ОКН 875840, ООПТ/леса/вода 875845/866/832/835) по кварталам ЕКБ (66:41:* из
cad_parcels) → grid-walk → upsert ird_overlays (м.132) по (source_layer_id,
geom_data_id). Геометрия 3857→4326. content_* НЕ трогаем (резолвер D9). SAVEPOINT
per-feature; сбой квартала/слоя изолирован. Beat: ird-harvest-weekly (Mon 05:00 МСК).

Refs #1067 (B5). Тесты: 3 passed.
This commit is contained in:
bot-backend 2026-06-06 21:45:32 +03:00
parent 4e41f9be73
commit d7e5f53789
4 changed files with 365 additions and 0 deletions

View file

@ -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

View file

@ -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"

View file

@ -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(38574326).
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()

View file

@ -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 == []