225 lines
9.8 KiB
Python
225 lines
9.8 KiB
Python
"""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
|
||
# Durable per-quarter commit: длинный grid-walk (часы) не теряет прогресс при
|
||
# краше/таймауте/WorkerLost середины прогона (commit раз-в-конце терял ВСЁ).
|
||
db.commit()
|
||
except Exception as exc:
|
||
logger.warning("ird_harvest: квартал %s failed: %s", quarter, exc)
|
||
db.rollback() # сбросить незакоммиченный tx квартала перед следующим
|
||
continue
|
||
db.commit() # финальный — на случай пустого списка / подстраховка
|
||
logger.info("ird_harvest: готово, кварталов=%d фич=%d", n_quarters, n_features)
|
||
return {"quarters": n_quarters, "features": n_features}
|
||
finally:
|
||
db.close()
|