369 lines
15 KiB
Python
369 lines
15 KiB
Python
"""Denormalize nspd_quarter_dumps.features_json → nspd_parcels / nspd_buildings.
|
||
|
||
Используется:
|
||
- Inline в harvest_quarter task после UPSERT dump'а
|
||
- Backfill task для existing dumps (backfill_all_dumps)
|
||
|
||
Property mapping (NSPD WMS GetFeatureInfo responses):
|
||
Parcels (layer="parcels", layer id 36048):
|
||
cad_num — кадастровый номер ЗУ
|
||
permitted_use — ВРИ
|
||
land_category — категория земель
|
||
cost_value — кадастровая стоимость, руб.
|
||
area — площадь, м²
|
||
address — адрес
|
||
|
||
Buildings (layer="buildings", layer id 36049):
|
||
cad_num — кадастровый номер ОКС
|
||
purpose — назначение ("Многоквартирный дом" / "Нежилое здание" / ...)
|
||
floors_above_ground — надземных этажей
|
||
floors_underground — подземных этажей
|
||
year_built — год постройки
|
||
cost_value — кадастровая стоимость, руб.
|
||
build_record_area — площадь по ГКН, м²
|
||
address — адрес
|
||
|
||
Геометрия в features_json хранится в EPSG:3857 (как возвращает NSPD WMS) —
|
||
трансформация 3857→4326 делается в SQL через ST_Transform.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import datetime
|
||
import json
|
||
import logging
|
||
import re
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── Engineering classifier ─────────────────────────────────────────────────────
|
||
|
||
# Паттерны для классификации инженерных сооружений (layer 36328) по текстовым
|
||
# свойствам. Источник: bug-post-mortem fixes/Bug_NSPD_WMS_NotBulk_2026_May14.md
|
||
# + live test данные 7×7 grid на 1км² центра ЕКБ.
|
||
# Порядок проверки важен: более специфичные паттерны идут первыми.
|
||
|
||
_ENGINEERING_PATTERNS: list[tuple[re.Pattern[str], str]] = [
|
||
# gas — газопроводы, ГРП, ГК (газовые колодцы)
|
||
(re.compile(r"газопровод|газоснабж|ГРП\b|ГК\b", re.IGNORECASE), "gas"),
|
||
# sewage — канализация, сток, ливневые сети
|
||
(re.compile(r"канализ|сточ|ливнев|самотёч|самотеч", re.IGNORECASE), "sewage"),
|
||
# heat — тепловые сети, котельные, ТЭЦ
|
||
(
|
||
re.compile(r"теплов(ая|ой|ые)|теплосеть|теплоснабж|котельн|ТЭЦ\b", re.IGNORECASE),
|
||
"heat",
|
||
),
|
||
# electric — электросети, ВЛ, КЛ, ЛЭП, подстанции, трансформаторы, ТП
|
||
(
|
||
re.compile(
|
||
r"электроэнерг|ВЛ[\s\-]|ВЛ-?\d|КЛ[\s\-]|КЛ-?\d|ЛЭП\b"
|
||
r"|подстанц|трансформ|ТП[\s\-]?\d",
|
||
re.IGNORECASE,
|
||
),
|
||
"electric",
|
||
),
|
||
# water — водопровод, водоснабжение, хозбытовые водосети
|
||
(re.compile(r"водопровод|водоснабж|хозбытов|водовод", re.IGNORECASE), "water"),
|
||
]
|
||
|
||
# Поля из NSPD properties в которых ищем паттерны (по приоритету)
|
||
_ENGINEERING_TEXT_FIELDS = ("params_name", "name", "params_purpose", "purpose", "label")
|
||
|
||
|
||
def classify_engineering_kind(properties: dict[str, Any]) -> str:
|
||
"""Классифицировать инженерное сооружение (layer 36328) по его properties.
|
||
|
||
Проверяет поля `params_name`, `name`, `params_purpose`, `purpose`, `label`
|
||
против regex-паттернов. Возвращает первое совпадение.
|
||
|
||
Args:
|
||
properties: dict свойств NSPDFeature.properties из WMS GetFeatureInfo
|
||
или search/geoportal response.
|
||
|
||
Returns:
|
||
Одно из: ``"water"`` | ``"sewage"`` | ``"gas"`` | ``"heat"``
|
||
| ``"electric"`` | ``"other"``.
|
||
|
||
Examples:
|
||
>>> classify_engineering_kind({"params_name": "Газопровод высокого давления"})
|
||
'gas'
|
||
>>> classify_engineering_kind({"name": "КЛ 10 кВ ТП 64102"})
|
||
'electric'
|
||
>>> classify_engineering_kind({"params_purpose": "Водопровод хозбытовой"})
|
||
'water'
|
||
>>> classify_engineering_kind({"params_name": "Тепловая сеть"})
|
||
'heat'
|
||
>>> classify_engineering_kind({"name": "Канализация"})
|
||
'sewage'
|
||
"""
|
||
# Собираем текст для проверки из всех релевантных полей
|
||
text_parts: list[str] = []
|
||
for field in _ENGINEERING_TEXT_FIELDS:
|
||
val = properties.get(field)
|
||
if val and isinstance(val, str):
|
||
text_parts.append(val)
|
||
|
||
combined = " ".join(text_parts)
|
||
if not combined:
|
||
return "other"
|
||
|
||
for pattern, kind in _ENGINEERING_PATTERNS:
|
||
if pattern.search(combined):
|
||
return kind
|
||
|
||
return "other"
|
||
|
||
|
||
# ── Type coercions ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _coerce_int(v: Any) -> int | None:
|
||
"""NSPD properties могут быть str / int / None."""
|
||
if v is None:
|
||
return None
|
||
try:
|
||
return int(v)
|
||
except (ValueError, TypeError):
|
||
return None
|
||
|
||
|
||
def _coerce_numeric(v: Any) -> float | None:
|
||
"""NSPD properties могут быть str с запятой (европейский формат) или None."""
|
||
if v is None:
|
||
return None
|
||
try:
|
||
return float(str(v).replace(",", ".").strip())
|
||
except (ValueError, TypeError):
|
||
return None
|
||
|
||
|
||
# ── Parcel UPSERT ──────────────────────────────────────────────────────────────
|
||
|
||
_PARCEL_UPSERT_SQL = text(
|
||
"""
|
||
INSERT INTO nspd_parcels (
|
||
cad_num, quarter_cad, permitted_use, land_category,
|
||
cost_value, cost_per_m2, area_sqm, address, geom, snapshot_date
|
||
) VALUES (
|
||
CAST(:cad_num AS text),
|
||
CAST(:quarter_cad AS text),
|
||
CAST(:permitted_use AS text),
|
||
CAST(:land_category AS text),
|
||
CAST(:cost_value AS numeric),
|
||
CAST(:cost_per_m2 AS numeric),
|
||
CAST(:area_sqm AS numeric),
|
||
CAST(:address AS text),
|
||
CASE WHEN :geom_json IS NULL THEN NULL
|
||
ELSE ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:geom_json), 3857), 4326)
|
||
END,
|
||
CAST(:snapshot_date AS date)
|
||
)
|
||
ON CONFLICT (cad_num) DO UPDATE SET
|
||
quarter_cad = EXCLUDED.quarter_cad,
|
||
permitted_use = EXCLUDED.permitted_use,
|
||
land_category = EXCLUDED.land_category,
|
||
cost_value = EXCLUDED.cost_value,
|
||
cost_per_m2 = EXCLUDED.cost_per_m2,
|
||
area_sqm = EXCLUDED.area_sqm,
|
||
address = EXCLUDED.address,
|
||
geom = EXCLUDED.geom,
|
||
snapshot_date = EXCLUDED.snapshot_date,
|
||
updated_at = NOW()
|
||
"""
|
||
)
|
||
|
||
|
||
def denorm_parcel_feature(
|
||
db: Session,
|
||
*,
|
||
feature: dict[str, Any],
|
||
quarter_cad: str,
|
||
snapshot_date: str,
|
||
) -> bool:
|
||
"""UPSERT one parcel feature → nspd_parcels.
|
||
|
||
Returns True если строка вставлена/обновлена, False при пропуске (нет cad_num).
|
||
"""
|
||
props = feature.get("properties") or {}
|
||
cad_num = props.get("cad_num") or props.get("cadastral_number")
|
||
if not cad_num:
|
||
return False
|
||
|
||
area = _coerce_numeric(props.get("area"))
|
||
cost_value = _coerce_numeric(props.get("cost_value"))
|
||
cost_per_m2: float | None = None
|
||
if cost_value is not None and area is not None and area > 0:
|
||
cost_per_m2 = cost_value / area
|
||
|
||
geom = feature.get("geometry")
|
||
geom_json: str | None = json.dumps(geom, ensure_ascii=False) if geom else None
|
||
|
||
try:
|
||
with db.begin_nested():
|
||
db.execute(
|
||
_PARCEL_UPSERT_SQL,
|
||
{
|
||
"cad_num": cad_num,
|
||
"quarter_cad": quarter_cad,
|
||
"permitted_use": props.get("permitted_use"),
|
||
"land_category": props.get("land_category"),
|
||
"cost_value": cost_value,
|
||
"cost_per_m2": cost_per_m2,
|
||
"area_sqm": area,
|
||
"address": props.get("address"),
|
||
"geom_json": geom_json,
|
||
"snapshot_date": snapshot_date,
|
||
},
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
logger.warning("denorm parcel %s failed: %s", cad_num, e)
|
||
return False
|
||
|
||
|
||
# ── Building UPSERT ────────────────────────────────────────────────────────────
|
||
|
||
_BUILDING_UPSERT_SQL = text(
|
||
"""
|
||
INSERT INTO nspd_buildings (
|
||
cad_num, quarter_cad, purpose, floors, floors_underground,
|
||
year_built, cost_value, build_record_area, address, geom, snapshot_date
|
||
) VALUES (
|
||
CAST(:cad_num AS text),
|
||
CAST(:quarter_cad AS text),
|
||
CAST(:purpose AS text),
|
||
CAST(:floors AS integer),
|
||
CAST(:floors_underground AS integer),
|
||
CAST(:year_built AS integer),
|
||
CAST(:cost_value AS numeric),
|
||
CAST(:build_record_area AS numeric),
|
||
CAST(:address AS text),
|
||
CASE WHEN :geom_json IS NULL THEN NULL
|
||
ELSE ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:geom_json), 3857), 4326)
|
||
END,
|
||
CAST(:snapshot_date AS date)
|
||
)
|
||
ON CONFLICT (cad_num) DO UPDATE SET
|
||
quarter_cad = EXCLUDED.quarter_cad,
|
||
purpose = EXCLUDED.purpose,
|
||
floors = EXCLUDED.floors,
|
||
floors_underground = EXCLUDED.floors_underground,
|
||
year_built = EXCLUDED.year_built,
|
||
cost_value = EXCLUDED.cost_value,
|
||
build_record_area = EXCLUDED.build_record_area,
|
||
address = EXCLUDED.address,
|
||
geom = EXCLUDED.geom,
|
||
snapshot_date = EXCLUDED.snapshot_date,
|
||
updated_at = NOW()
|
||
"""
|
||
)
|
||
|
||
|
||
def denorm_building_feature(
|
||
db: Session,
|
||
*,
|
||
feature: dict[str, Any],
|
||
quarter_cad: str,
|
||
snapshot_date: str,
|
||
) -> bool:
|
||
"""UPSERT one building feature → nspd_buildings.
|
||
|
||
Returns True если строка вставлена/обновлена, False при пропуске (нет cad_num).
|
||
"""
|
||
props = feature.get("properties") or {}
|
||
cad_num = props.get("cad_num") or props.get("cadastral_number")
|
||
if not cad_num:
|
||
return False
|
||
|
||
geom = feature.get("geometry")
|
||
geom_json: str | None = json.dumps(geom, ensure_ascii=False) if geom else None
|
||
|
||
try:
|
||
with db.begin_nested():
|
||
db.execute(
|
||
_BUILDING_UPSERT_SQL,
|
||
{
|
||
"cad_num": cad_num,
|
||
"quarter_cad": quarter_cad,
|
||
"purpose": props.get("purpose"),
|
||
"floors": _coerce_int(props.get("floors_above_ground")),
|
||
"floors_underground": _coerce_int(props.get("floors_underground")),
|
||
"year_built": _coerce_int(props.get("year_built")),
|
||
"cost_value": _coerce_numeric(props.get("cost_value")),
|
||
"build_record_area": _coerce_numeric(props.get("build_record_area")),
|
||
"address": props.get("address"),
|
||
"geom_json": geom_json,
|
||
"snapshot_date": snapshot_date,
|
||
},
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
logger.warning("denorm building %s failed: %s", cad_num, e)
|
||
return False
|
||
|
||
|
||
# ── Batch helper ───────────────────────────────────────────────────────────────
|
||
|
||
|
||
def denorm_dump(
|
||
db: Session,
|
||
*,
|
||
quarter_cad: str,
|
||
features: list[dict[str, Any]],
|
||
) -> dict[str, int]:
|
||
"""Denorm all features из одного quarter dump → nspd_parcels / nspd_buildings.
|
||
|
||
Итерирует features_json плоский список. Записи с layer="parcels" идут в
|
||
nspd_parcels, layer="buildings" → nspd_buildings. Остальные layers пропускаются.
|
||
|
||
Каждая строка UPSERT'ится в отдельном SAVEPOINT (begin_nested) — ошибка
|
||
одной строки не откатывает весь batch.
|
||
|
||
Args:
|
||
db: SQLAlchemy Session. Caller отвечает за commit/close после вызова.
|
||
quarter_cad: 3-сегментный кадастровый квартал.
|
||
features: плоский list из features_json JSONB (уже декодированный Python list).
|
||
|
||
Returns:
|
||
dict {"parcels": N, "buildings": M, "errors": K} — количество обработанных строк.
|
||
"""
|
||
snapshot_date = datetime.date.today().isoformat()
|
||
parcels_n = 0
|
||
buildings_n = 0
|
||
errors_n = 0
|
||
|
||
for feat in features:
|
||
layer = feat.get("layer", "")
|
||
try:
|
||
if layer == "parcels":
|
||
if denorm_parcel_feature(
|
||
db, feature=feat, quarter_cad=quarter_cad, snapshot_date=snapshot_date
|
||
):
|
||
parcels_n += 1
|
||
else:
|
||
errors_n += 1
|
||
elif layer == "buildings":
|
||
if denorm_building_feature(
|
||
db, feature=feat, quarter_cad=quarter_cad, snapshot_date=snapshot_date
|
||
):
|
||
buildings_n += 1
|
||
else:
|
||
errors_n += 1
|
||
# Остальные layers (territorial_zones, zouit_*, risk_*, ...) — пропускаем
|
||
except Exception as e:
|
||
logger.warning("denorm feature (layer=%s quarter=%s) failed: %s", layer, quarter_cad, e)
|
||
errors_n += 1
|
||
|
||
db.commit()
|
||
logger.info(
|
||
"denorm_dump quarter=%s parcels=%d buildings=%d errors=%d",
|
||
quarter_cad,
|
||
parcels_n,
|
||
buildings_n,
|
||
errors_n,
|
||
)
|
||
return {"parcels": parcels_n, "buildings": buildings_n, "errors": errors_n}
|