390 lines
17 KiB
Python
390 lines
17 KiB
Python
"""Загрузчик OSM инженерных сетей («инженерные сети») из Overpass API (#1746).
|
||
|
||
No-B2B open-data путь: тянет ЛЭП / подстанции / трубопроводы (газ/вода) / ЦТП /
|
||
вышки связи через Overpass для bbox ЕКБ и UPSERT-ит в
|
||
``osm_utility_infrastructure_ekb``. Живой тест дал 3081+ utility-элементов ЕКБ.
|
||
|
||
Структура зеркалит ``noise_loader.py``: httpx-клиент с явным UA + таймаутом,
|
||
rate-limit ~1 req/s (Overpass usage policy), геометрия way → LineString /
|
||
node → Point, per-element SAVEPOINT при UPSERT (битый элемент не валит батч).
|
||
|
||
Запускается раз в неделю через Celery beat. Геометрия включает И линии (ЛЭП,
|
||
трубопроводы — way), И точки (опоры, подстанции, узлы ВКХ, ЦТП, вышки — node).
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
|
||
import httpx
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.db import SessionLocal
|
||
from app.services.site_finder.noise_loader import (
|
||
EKB_BBOX,
|
||
_way_to_linestring_wkt,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
OVERPASS_URL = "https://overpass-api.de/api/interpreter"
|
||
|
||
# Overpass блокирует дефолтный User-Agent (406) — явный UA с контактом (как noise_loader).
|
||
_HEADERS = {
|
||
"User-Agent": "GenDesign-SiteFinder/1.0 (+https://gendsgn.ru)",
|
||
"Accept": "application/json",
|
||
}
|
||
|
||
# Описание запросов: (overpass_key, overpass_value, el_type, infrastructure_kind).
|
||
# el_type: 'way' (out geom) | 'node' (out body) | 'nwr' (out geom, node+way).
|
||
# substance: для pipeline-веток газ/вода дополнительно фильтруем по substance,
|
||
# чтобы не смешивать водопровод с газопроводом — см. _SUBSTANCE_FILTER.
|
||
# Виды сети (infrastructure_kind):
|
||
# power | water | gas | heat | communication | sewage
|
||
_UTILITY_QUERIES: list[tuple[str, str, str, str]] = [
|
||
# power — магистральные ЛЭП (линии) + опоры (точки) + подстанции (точка/площадка)
|
||
("power", "line", "way", "power"),
|
||
("power", "minor_line", "way", "power"),
|
||
("power", "tower", "node", "power"),
|
||
("power", "substation", "nwr", "power"),
|
||
# water — водозаборы/узлы ВКХ (точка/площадка) + магистраль pipeline substance=water
|
||
("man_made", "water_works", "nwr", "water"),
|
||
("man_made", "water_tower", "nwr", "water"),
|
||
("pipeline", "water", "way", "water"),
|
||
# gas — магистральные газопроводы (way), pipeline substance=gas
|
||
("pipeline", "gas", "way", "gas"),
|
||
("man_made", "pipeline", "way", "gas"), # substance=gas фильтруется ниже
|
||
# heat — ЦТП / теплопункты (sparse — это нормально)
|
||
("man_made", "heat_substation", "nwr", "heat"),
|
||
# communication — вышки связи (точка/площадка) tower:type=communication
|
||
("man_made", "tower", "nwr", "communication"),
|
||
# sewage — магистральная канализация (pipeline substance=sewerage) + очистные
|
||
("pipeline", "sewerage", "way", "sewage"),
|
||
("man_made", "wastewater_plant", "nwr", "sewage"),
|
||
]
|
||
|
||
# Для generic man_made=pipeline нужно различать вещество по тегу `substance`
|
||
# (или legacy `type`): иначе под газ попадут все трубы. Ключ — (key, value),
|
||
# значение — ожидаемый substance. Элементы без совпадения substance пропускаем.
|
||
_SUBSTANCE_FILTER: dict[tuple[str, str], frozenset[str]] = {
|
||
("man_made", "pipeline"): frozenset({"gas", "natural_gas", "cng"}),
|
||
}
|
||
|
||
# tower:type=communication — для man_made=tower оставляем только вышки связи
|
||
# (иначе под communication попадут смотровые/водонапорные башни).
|
||
_TOWER_TYPE_REQUIRED: dict[tuple[str, str], frozenset[str]] = {
|
||
("man_made", "tower"): frozenset({"communication", "communications"}),
|
||
}
|
||
|
||
|
||
def _build_overpass_query(key: str, value: str, el_type: str) -> str:
|
||
"""Запрос Overpass для одного типа инж.-сетевого элемента.
|
||
|
||
way / nwr → ``out geom`` (нужны координаты узлов для LineString / way-площадок);
|
||
node → ``out body``. Зеркалит noise_loader._build_overpass_query.
|
||
"""
|
||
south, west, north, east = EKB_BBOX
|
||
bbox = f"({south},{west},{north},{east})"
|
||
if el_type == "nwr":
|
||
return f"[out:json][timeout:30];" f'nwr["{key}"="{value}"]{bbox};' f"out geom;"
|
||
if el_type == "way":
|
||
return f"[out:json][timeout:30];" f'way["{key}"="{value}"]{bbox};' f"out geom;"
|
||
return f"[out:json][timeout:30];" f'node["{key}"="{value}"]{bbox};' f"out body;"
|
||
|
||
|
||
def _passes_tag_filters(key: str, value: str, tags: dict) -> bool:
|
||
"""Доп-фильтр по тегам элемента (substance / tower:type).
|
||
|
||
Для generic ``man_made=pipeline`` оставляем только газовые трубы (substance),
|
||
для ``man_made=tower`` — только вышки связи (tower:type). Остальные ключи
|
||
проходят без фильтра.
|
||
"""
|
||
substance_ok = _SUBSTANCE_FILTER.get((key, value))
|
||
if substance_ok is not None:
|
||
substance = (tags.get("substance") or tags.get("type") or "").lower()
|
||
if substance not in substance_ok:
|
||
return False
|
||
tower_ok = _TOWER_TYPE_REQUIRED.get((key, value))
|
||
if tower_ok is not None:
|
||
tower_type = (tags.get("tower:type") or "").lower()
|
||
if tower_type not in tower_ok:
|
||
return False
|
||
return True
|
||
|
||
|
||
async def fetch_overpass_utility() -> list[dict]:
|
||
"""Запрашивает Overpass API для всех инж.-сетевых типов ЕКБ.
|
||
|
||
Возвращает enriched list dict'ов с доп-полями:
|
||
``_infrastructure_kind``, ``_source_tag`` — для UPSERT.
|
||
Между запросами sleep 1с (Overpass usage policy), как в noise_loader.
|
||
"""
|
||
all_elements: list[dict] = []
|
||
async with httpx.AsyncClient(timeout=60, headers=_HEADERS) as client:
|
||
for key, value, el_type, kind in _UTILITY_QUERIES:
|
||
query = _build_overpass_query(key, value, el_type)
|
||
try:
|
||
r = await client.post(OVERPASS_URL, data={"data": query})
|
||
r.raise_for_status()
|
||
elements: list[dict] = r.json().get("elements", [])
|
||
logger.info(
|
||
"Overpass utility: %s=%s (%s) → %d elements",
|
||
key,
|
||
value,
|
||
kind,
|
||
len(elements),
|
||
)
|
||
for el in elements:
|
||
el["_infrastructure_kind"] = kind
|
||
el["_source_tag"] = f"{key}={value}"
|
||
el["_query_key"] = key
|
||
el["_query_value"] = value
|
||
all_elements.extend(elements)
|
||
except Exception as e:
|
||
logger.warning("Overpass utility failed for %s=%s: %s", key, value, e)
|
||
await asyncio.sleep(1.0)
|
||
logger.info("Overpass utility: total %d elements", len(all_elements))
|
||
return all_elements
|
||
|
||
|
||
def sync_utility_infrastructure_to_db() -> dict[str, int]:
|
||
"""UPSERT инж.сетей из Overpass в osm_utility_infrastructure_ekb.
|
||
|
||
UPSERT по UNIQUE(osm_type, osm_id). Геометрия:
|
||
- way с nodes → LineString WKT через ST_GeomFromText(..., 4326)
|
||
- node → ST_SetSRID(ST_MakePoint(lon, lat), 4326)
|
||
Per-element SAVEPOINT — битый элемент не валит весь weekly-sync.
|
||
|
||
Returns: счётчики fetched/inserted/updated/skipped + per-kind counts.
|
||
"""
|
||
elements = asyncio.run(fetch_overpass_utility())
|
||
return _upsert_elements(elements)
|
||
|
||
|
||
def _upsert_elements(elements: list[dict]) -> dict[str, int]:
|
||
"""UPSERT уже полученных Overpass-элементов в таблицу (выделено для тестов).
|
||
|
||
Принимает enriched-элементы (с ``_infrastructure_kind`` / ``_source_tag``),
|
||
строит геометрию и UPSERT-ит под per-element SAVEPOINT. Возвращает счётчики.
|
||
"""
|
||
inserted = 0
|
||
updated = 0
|
||
skipped = 0
|
||
fetched = len(elements)
|
||
by_kind: dict[str, int] = {}
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
for el in elements:
|
||
key: str = el.get("_query_key", "")
|
||
value: str = el.get("_query_value", "")
|
||
tags: dict = el.get("tags") or {}
|
||
|
||
# substance / tower:type фильтры (generic man_made=pipeline / tower).
|
||
if not _passes_tag_filters(key, value, tags):
|
||
skipped += 1
|
||
continue
|
||
|
||
osm_id: int = el["id"]
|
||
osm_type: str = el["type"] # 'way' | 'node'
|
||
kind: str = el.get("_infrastructure_kind", "power")
|
||
source_tag: str | None = el.get("_source_tag")
|
||
name: str | None = tags.get("name")
|
||
|
||
geom_sql: str
|
||
geom_params: dict
|
||
if osm_type == "way":
|
||
nodes: list[dict] = el.get("geometry") or []
|
||
wkt = _way_to_linestring_wkt(nodes)
|
||
if wkt is None:
|
||
# way-площадка (закрытый контур) тоже даёт LineString из geometry —
|
||
# если узлов <2, пропускаем (битый/частичный bbox-обрезок).
|
||
skipped += 1
|
||
continue
|
||
geom_sql = "ST_GeomFromText(:wkt, 4326)"
|
||
geom_params = {"wkt": wkt}
|
||
elif osm_type == "node":
|
||
lat = el.get("lat")
|
||
lon = el.get("lon")
|
||
if lat is None or lon is None:
|
||
skipped += 1
|
||
continue
|
||
geom_sql = "ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)"
|
||
geom_params = {"lat": lat, "lon": lon}
|
||
else:
|
||
# relation — skip (сложная геометрия, не нужна сейчас)
|
||
skipped += 1
|
||
continue
|
||
|
||
params: dict = {
|
||
"osm_id": osm_id,
|
||
"osm_type": osm_type,
|
||
"infrastructure_kind": kind,
|
||
"name": name,
|
||
"source_tag": source_tag,
|
||
"tags": json.dumps(tags, ensure_ascii=False),
|
||
**geom_params,
|
||
}
|
||
|
||
try:
|
||
with db.begin_nested(): # SAVEPOINT — откат только этой записи
|
||
result = db.execute(
|
||
text(f"""
|
||
INSERT INTO osm_utility_infrastructure_ekb
|
||
(osm_type, osm_id, infrastructure_kind, name, geom,
|
||
source_tag, tags, fetched_at)
|
||
VALUES (
|
||
:osm_type, :osm_id, :infrastructure_kind, :name,
|
||
{geom_sql},
|
||
:source_tag, CAST(:tags AS jsonb), NOW()
|
||
)
|
||
ON CONFLICT (osm_type, osm_id) DO UPDATE
|
||
SET infrastructure_kind = EXCLUDED.infrastructure_kind,
|
||
name = EXCLUDED.name,
|
||
geom = EXCLUDED.geom,
|
||
source_tag = EXCLUDED.source_tag,
|
||
tags = EXCLUDED.tags,
|
||
fetched_at = NOW()
|
||
RETURNING (xmax = 0) AS is_insert
|
||
"""),
|
||
params,
|
||
).scalar()
|
||
if result:
|
||
inserted += 1
|
||
else:
|
||
updated += 1
|
||
by_kind[kind] = by_kind.get(kind, 0) + 1
|
||
except Exception as e:
|
||
# Дефектный элемент (битая геометрия и т.п.) не должен валить
|
||
# весь weekly-sync. SAVEPOINT откатывает только эту строку.
|
||
logger.warning(
|
||
"utility_sync upsert failed for %s/%s (kind=%s): %s",
|
||
osm_type,
|
||
osm_id,
|
||
kind,
|
||
e,
|
||
)
|
||
skipped += 1
|
||
|
||
db.commit()
|
||
except Exception as e:
|
||
db.rollback()
|
||
logger.exception("utility_sync: unexpected error, outer tx rolled back: %s", e)
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
result_dict: dict[str, int] = {
|
||
"fetched": fetched,
|
||
"inserted": inserted,
|
||
"updated": updated,
|
||
"skipped": skipped,
|
||
}
|
||
# per-kind counts (power/water/gas/heat/communication/sewage) — для отчёта populate.
|
||
for k, c in sorted(by_kind.items()):
|
||
result_dict[f"kind_{k}"] = c
|
||
logger.info("utility_sync done: %s", result_dict)
|
||
return result_dict
|
||
|
||
|
||
async def load_utility_infrastructure(db: Session | None = None) -> dict[str, int]:
|
||
"""Async-обёртка для прямого вызова из async-контекста (#1746 spec).
|
||
|
||
Тянет Overpass и UPSERT-ит в таблицу. ``db`` принимается для сигнатурной
|
||
совместимости со спекой, но UPSERT идёт через собственную SessionLocal
|
||
(как noise/poi sync) — Overpass-fetch async, запись в БД sync. Возвращает
|
||
те же per-kind counts, что ``sync_utility_infrastructure_to_db``.
|
||
"""
|
||
elements = await fetch_overpass_utility()
|
||
return _upsert_elements(elements)
|
||
|
||
|
||
def get_nearby_utility_infrastructure(
|
||
db: Session,
|
||
cad_num: str,
|
||
radius_m: int = 500,
|
||
limit: int = 200,
|
||
) -> dict:
|
||
"""Инженерные сети из OSM в radius_m от центроида участка (#1746 endpoint).
|
||
|
||
Источник: ``osm_utility_infrastructure_ekb`` (наполняется weekly-sync из
|
||
Overpass). Расстояние считается от ЦЕНТРОИДА участка через ST_DWithin /
|
||
ST_Distance в geography. Возвращает feature-list + summary с ближайшими
|
||
расстояниями по видам сети.
|
||
|
||
Args:
|
||
db: SQLAlchemy session.
|
||
cad_num: кадастровый номер участка (e.g. '66:41:0204016:10').
|
||
radius_m: радиус поиска, м (caller валидирует диапазон).
|
||
limit: максимум фич в ответе (ближайшие).
|
||
|
||
Returns:
|
||
{"features": [...], "summary": {...}}
|
||
|
||
Raises:
|
||
ValueError: участок не найден в БД (нет geom).
|
||
"""
|
||
# Импорт здесь, чтобы избежать циклического импорта на module-load
|
||
# (quarter_dump_lookup → noise_loader → ... ).
|
||
from app.services.site_finder.quarter_dump_lookup import _get_parcel_wkt
|
||
|
||
parcel_wkt = _get_parcel_wkt(db, cad_num)
|
||
if parcel_wkt is None:
|
||
raise ValueError(f"Участок {cad_num!r} не найден в БД")
|
||
|
||
rows = (
|
||
db.execute(
|
||
text("""
|
||
SELECT osm_id, osm_type, infrastructure_kind, name, source_tag,
|
||
ST_Distance(
|
||
u.geom::geography,
|
||
ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography
|
||
) AS distance_m,
|
||
ST_AsGeoJSON(u.geom) AS geojson
|
||
FROM osm_utility_infrastructure_ekb u
|
||
WHERE ST_DWithin(
|
||
u.geom::geography,
|
||
ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography,
|
||
CAST(:radius_m AS float)
|
||
)
|
||
ORDER BY distance_m ASC
|
||
LIMIT :lim
|
||
"""),
|
||
{"wkt": parcel_wkt, "radius_m": radius_m, "lim": limit},
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
|
||
features: list[dict] = []
|
||
nearest_by_kind: dict[str, float | None] = {}
|
||
nearest_overall: float | None = None
|
||
for r in rows:
|
||
dist = round(float(r["distance_m"]), 1)
|
||
kind = r["infrastructure_kind"]
|
||
# rows отсортированы по distance ASC → первое попадание вида = ближайшее.
|
||
if kind not in nearest_by_kind:
|
||
nearest_by_kind[kind] = dist
|
||
if nearest_overall is None:
|
||
nearest_overall = dist
|
||
features.append(
|
||
{
|
||
"osm_id": int(r["osm_id"]),
|
||
"osm_type": r["osm_type"],
|
||
"infrastructure_kind": kind,
|
||
"name": r["name"],
|
||
"source_tag": r["source_tag"],
|
||
"distance_m": dist,
|
||
"geometry_geojson": json.loads(r["geojson"]) if r["geojson"] else {},
|
||
}
|
||
)
|
||
|
||
return {
|
||
"features": features,
|
||
"summary": {
|
||
"total_features": len(features),
|
||
"nearest_distance_m": nearest_overall,
|
||
"nearest_by_kind": nearest_by_kind,
|
||
},
|
||
}
|