gendesign/backend/app/workers/tasks/krt_geometry_sync.py
bot-backend bbb0879948
Some checks failed
CI / changes (push) Successful in 7s
CI / frontend-tests (push) Has been skipped
CI / changes (pull_request) Successful in 6s
CI / frontend-tests (pull_request) Has been skipped
CI / backend-tests (push) Has been cancelled
CI / backend-tests (pull_request) Successful in 6m22s
perf(sf): КРТ-геометрия в БД, _krt_at из БД вместо живого WFS (#1130)
2026-06-07 15:07:17 +03:00

122 lines
4.3 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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