bot-backend
86b9a5a197
Co-authored-by: bot-backend <bot-backend@gendsgn.local> Co-committed-by: bot-backend <bot-backend@gendsgn.local>
122 lines
4.3 KiB
Python
122 lines
4.3 KiB
Python
"""Celery task: еженедельный harvest КРТ-геометрий из геопортала ЕКБ (#1130).
|
||
|
||
Слой portal_geo_oks_krt («Здания в границах КРТ», ~48 WFS-фич) — UPSERT в ekb_krt_geometry.
|
||
После harvest analyze._krt_at читает из БД (ST_Intersects) вместо живого WFS-вызова.
|
||
|
||
UPSERT ON CONFLICT (feature_id): повторный запуск идемпотентен — обновляет геометрию/атрибуты.
|
||
SAVEPOINT per-row: одна упавшая фича не откатывает всю транзакцию.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
|
||
from sqlalchemy import text
|
||
|
||
from app.core.db import SessionLocal
|
||
from app.services.scrapers.ekb_geoportal_client import EKBGeoportalClient
|
||
from app.workers.celery_app import celery_app
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Bbox ЕКБ-агломерации (EPSG:4326): minlon, minlat, maxlon, maxlat.
|
||
# Охватывает весь муниципальный округ Екатеринбург с запасом; все ~48 КРТ-фич попадают.
|
||
_EKB_BBOX: tuple[float, float, float, float] = (60.3, 56.6, 61.1, 57.1)
|
||
|
||
_UPSERT_SQL = text("""
|
||
INSERT INTO ekb_krt_geometry (
|
||
feature_id, num_oks_krt, full_name, status,
|
||
geom, area, raw_props, fetched_at
|
||
)
|
||
VALUES (
|
||
:feature_id,
|
||
CAST(:num_oks_krt AS integer),
|
||
:full_name,
|
||
:status,
|
||
ST_SetSRID(ST_GeomFromGeoJSON(CAST(:geom_json AS text)), 4326),
|
||
CAST(:area AS numeric),
|
||
CAST(:raw_props AS jsonb),
|
||
now()
|
||
)
|
||
ON CONFLICT (feature_id) DO UPDATE SET
|
||
num_oks_krt = EXCLUDED.num_oks_krt,
|
||
full_name = EXCLUDED.full_name,
|
||
status = EXCLUDED.status,
|
||
geom = EXCLUDED.geom,
|
||
area = EXCLUDED.area,
|
||
raw_props = EXCLUDED.raw_props,
|
||
fetched_at = now()
|
||
""")
|
||
|
||
|
||
@celery_app.task(name="tasks.krt_geometry_sync.sync_krt_geometry")
|
||
def sync_krt_geometry() -> dict[str, int]:
|
||
"""Fetch + UPSERT геометрий КРТ-слоя геопортала ЕКБ → ekb_krt_geometry.
|
||
|
||
Один bbox-запрос по всей агломерации ЕКБ (~48 фич). ON CONFLICT idempotent.
|
||
Returns: {"krt": N} — число обработанных фич.
|
||
"""
|
||
logger.info("krt_geometry_sync: start (bbox=%s)", _EKB_BBOX)
|
||
|
||
client = EKBGeoportalClient()
|
||
try:
|
||
features = client.features_in_bbox("krt", _EKB_BBOX)
|
||
except Exception as exc:
|
||
logger.error("krt_geometry_sync: WFS fetch failed: %s", exc)
|
||
raise
|
||
|
||
if not features:
|
||
logger.warning(
|
||
"krt_geometry_sync: WFS returned 0 features — слой пуст или геопортал недоступен"
|
||
)
|
||
return {"krt": 0}
|
||
|
||
logger.info("krt_geometry_sync: fetched %d features from WFS", len(features))
|
||
|
||
upserted = 0
|
||
failed = 0
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
for feat in features:
|
||
if feat.geometry is None:
|
||
logger.warning(
|
||
"krt_geometry_sync: feature_id=%s has no geometry — пропускаем", feat.feature_id
|
||
)
|
||
failed += 1
|
||
continue
|
||
props = feat.properties or {}
|
||
params = {
|
||
"feature_id": feat.feature_id,
|
||
"num_oks_krt": props.get("num_oks_krt"),
|
||
"full_name": props.get("full_name"),
|
||
"status": props.get("status"),
|
||
"geom_json": json.dumps(feat.geometry, ensure_ascii=False),
|
||
"area": props.get("area"),
|
||
"raw_props": json.dumps(props, ensure_ascii=False),
|
||
}
|
||
try:
|
||
with db.begin_nested():
|
||
db.execute(_UPSERT_SQL, params)
|
||
upserted += 1
|
||
except Exception as exc:
|
||
failed += 1
|
||
logger.warning(
|
||
"krt_geometry_sync: UPSERT failed for feature_id=%s: %s",
|
||
feat.feature_id,
|
||
exc,
|
||
)
|
||
db.commit()
|
||
except Exception:
|
||
db.rollback()
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
logger.info(
|
||
"krt_geometry_sync: done — upserted=%d failed=%d",
|
||
upserted,
|
||
failed,
|
||
)
|
||
return {"krt": upserted}
|