716 lines
34 KiB
Python
716 lines
34 KiB
Python
"""Загрузчик OSM инженерных сетей («инженерные сети») из Overpass API (#1746).
|
||
|
||
No-B2B open-data путь: тянет ЛЭП / подстанции / трубопроводы (газ/вода/теплотрассы) /
|
||
ЦТП / вышки связи через Overpass для bbox ЕКБ и UPSERT-ит в
|
||
``osm_utility_infrastructure_ekb``. Живой тест дал 3081+ utility-элементов ЕКБ.
|
||
|
||
Структура зеркалит ``noise_loader.py``: httpx-клиент с явным UA + таймаутом,
|
||
геометрия way → LineString / node → Point, per-element SAVEPOINT при UPSERT
|
||
(битый элемент не валит батч).
|
||
|
||
Hardening (#1746 follow-up, prod-инцидент с 429/504): per-query retry-with-backoff
|
||
(уважает ``Retry-After``) + фолбэк на зеркала Overpass (``OVERPASS_ENDPOINTS``) +
|
||
более мягкая пауза между query-types (``_INTER_QUERY_DELAY_SECONDS``). Раньше
|
||
публичный overpass-api.de троттлил поздние тяжёлые query-types → грузилось
|
||
~1099/3000 элементов; теперь покрытие полное.
|
||
|
||
Запускается раз в неделю через Celery beat. Геометрия включает И линии (ЛЭП,
|
||
трубопроводы — way), И точки (опоры, подстанции, узлы ВКХ, ЦТП, вышки — node).
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import random
|
||
import re
|
||
|
||
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__)
|
||
|
||
# Primary endpoint первым, далее зеркала. Публичный overpass-api.de агрессивно
|
||
# троттлит (429) и отдаёт 504 на тяжёлых query-types (#1746 prod-инцидент:
|
||
# power=tower/water_works/gas/heat/man_made=tower/wastewater_plant молча терялись,
|
||
# загрузилось 1099/~3000). При исчерпании ретраев на текущем endpoint переходим
|
||
# к следующему зеркалу. kumi.systems / private.coffee — общеизвестные mirror'ы.
|
||
OVERPASS_ENDPOINTS: list[str] = [
|
||
"https://overpass-api.de/api/interpreter",
|
||
"https://overpass.kumi.systems/api/interpreter",
|
||
"https://overpass.private.coffee/api/interpreter",
|
||
]
|
||
|
||
# Первый endpoint — для обратной совместимости (тесты/импорты на OVERPASS_URL).
|
||
OVERPASS_URL = OVERPASS_ENDPOINTS[0]
|
||
|
||
# Retry-with-backoff на 429 / 5xx (особенно 504). Попыток на ОДИН endpoint;
|
||
# при их исчерпании уходим к следующему зеркалу из OVERPASS_ENDPOINTS.
|
||
_MAX_RETRIES_PER_ENDPOINT = 3
|
||
# Экспоненциальный backoff (сек) между попытками на одном endpoint; индекс =
|
||
# номер уже сделанной неуспешной попытки. Если Retry-After в ответе есть — он в
|
||
# приоритете над этой таблицей.
|
||
_RETRY_BACKOFF_SECONDS: tuple[float, ...] = (10.0, 30.0, 60.0)
|
||
# Джиттер (сек) поверх backoff — размазываем повторные попытки, не бьём в ритме.
|
||
_RETRY_JITTER_SECONDS = 5.0
|
||
# HTTP-статусы, по которым ретраимся (троттлинг + временные сбои шлюза).
|
||
_RETRYABLE_STATUS = frozenset({429, 500, 502, 503, 504})
|
||
# Пауза между РАЗНЫМИ query-types. 1с был слишком агрессивен для публичного
|
||
# endpoint (#1746) — подняли до 3с (gentler на overpass usage policy).
|
||
_INTER_QUERY_DELAY_SECONDS = 3.0
|
||
|
||
# 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: generic man_made=pipeline фильтруем+классифицируем по substance —
|
||
# газопровод vs теплотрасса, см. _SUBSTANCE_FILTER / _PIPELINE_SUBSTANCE_KIND.
|
||
# Виды сети (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"),
|
||
# generic man_made=pipeline тянет ВСЕ трубы; substance разводит их на
|
||
# gas / heat (см. _PIPELINE_SUBSTANCE_KIND). kind ниже — дефолтный для строки
|
||
# (gas), реальный вид переопределяется по substance в _upsert_elements.
|
||
("man_made", "pipeline", "way", "gas"),
|
||
# heat — теплотрассы (man_made=pipeline substance∈{heat,hot_water,steam},
|
||
# разводятся из строки выше) + ЦТП / теплопункты (man_made=heat_substation,
|
||
# в ЕКБ sparse/0 — это нормально; линии-теплотрассы = единственный гео-источник §3).
|
||
("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`)
|
||
# определяет ВИД сети: газопровод (gas) vs теплотрасса (heat). Раньше брали только
|
||
# газ, остальное скипали — теплотрассы (единственный гео-источник §3, ~135 way в
|
||
# ЕКБ) терялись. Теперь substance → kind. Труба без известного substance
|
||
# по-прежнему пропускается (не под газ), чтобы нефтепроводы/химию не примешивать.
|
||
# gas — газоснабжение: gas / natural_gas / cng
|
||
# heat — теплоснабжение: heat / hot_water / steam
|
||
_PIPELINE_SUBSTANCE_KIND: dict[str, str] = {
|
||
"gas": "gas",
|
||
"natural_gas": "gas",
|
||
"cng": "gas",
|
||
"heat": "heat",
|
||
"hot_water": "heat",
|
||
"steam": "heat",
|
||
}
|
||
|
||
# Substance-фильтр для generic man_made=pipeline: пропускаем трубу только если её
|
||
# substance ведёт в известный вид сети (_PIPELINE_SUBSTANCE_KIND). Ключ — (key,
|
||
# value), значение — допустимые substance. Собирается из маппинга выше, чтобы
|
||
# фильтр и классификация не разошлись.
|
||
_SUBSTANCE_FILTER: dict[tuple[str, str], frozenset[str]] = {
|
||
("man_made", "pipeline"): frozenset(_PIPELINE_SUBSTANCE_KIND),
|
||
}
|
||
|
||
|
||
def _substance_of(tags: dict) -> str:
|
||
"""Вещество трубы: тег `substance` (или legacy `type`), lower-case, «» если нет."""
|
||
return (tags.get("substance") or tags.get("type") or "").lower()
|
||
|
||
|
||
def _pipeline_kind_from_substance(tags: dict, default_kind: str) -> str:
|
||
"""Вид сети для generic man_made=pipeline по substance.
|
||
|
||
substance∈{gas,natural_gas,cng} → gas; {heat,hot_water,steam} → heat.
|
||
Неизвестный/пустой substance → ``default_kind`` (труба уже прошла
|
||
_passes_tag_filters, до сюда без совпадения не доходит).
|
||
"""
|
||
return _PIPELINE_SUBSTANCE_KIND.get(_substance_of(tags), default_kind)
|
||
|
||
|
||
# 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];nwr["{key}"="{value}"]{bbox};out geom;'
|
||
if el_type == "way":
|
||
return f'[out:json][timeout:30];way["{key}"="{value}"]{bbox};out geom;'
|
||
return f'[out:json][timeout:30];node["{key}"="{value}"]{bbox};out body;'
|
||
|
||
|
||
def _passes_tag_filters(key: str, value: str, tags: dict) -> bool:
|
||
"""Доп-фильтр по тегам элемента (substance / tower:type).
|
||
|
||
Для generic ``man_made=pipeline`` оставляем только трубы с известным веществом
|
||
(газ или теплоноситель — см. _PIPELINE_SUBSTANCE_KIND); нефтепроводы/химию и
|
||
трубы без substance пропускаем. Для ``man_made=tower`` — только вышки связи
|
||
(tower:type). Остальные ключи проходят без фильтра.
|
||
"""
|
||
substance_ok = _SUBSTANCE_FILTER.get((key, value))
|
||
if substance_ok is not None:
|
||
if _substance_of(tags) 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
|
||
|
||
|
||
def _parse_retry_after(value: str | None) -> float | None:
|
||
"""Парсит заголовок ``Retry-After`` (только delta-seconds — самый частый
|
||
формат Overpass). HTTP-date форму игнорируем (Overpass её не отдаёт) — вернём
|
||
None и используем табличный backoff. Отрицательные / мусорные → None.
|
||
"""
|
||
if not value:
|
||
return None
|
||
try:
|
||
seconds = float(value.strip())
|
||
except (TypeError, ValueError):
|
||
return None
|
||
return seconds if seconds >= 0 else None
|
||
|
||
|
||
def _backoff_delay(attempt_idx: int, retry_after: float | None) -> float:
|
||
"""Задержка перед следующей попыткой.
|
||
|
||
``Retry-After`` (если сервер прислал) приоритетнее таблицы. Иначе —
|
||
экспоненциальный backoff из ``_RETRY_BACKOFF_SECONDS`` по индексу попытки
|
||
(с клампом на последний элемент). Всегда + рандомный джиттер, чтобы повторы
|
||
не били синхронно.
|
||
"""
|
||
if retry_after is not None:
|
||
base = retry_after
|
||
else:
|
||
idx = min(attempt_idx, len(_RETRY_BACKOFF_SECONDS) - 1)
|
||
base = _RETRY_BACKOFF_SECONDS[idx]
|
||
# random.uniform — это джиттер планировщика ретраев, не криптография.
|
||
return base + random.uniform(0.0, _RETRY_JITTER_SECONDS)
|
||
|
||
|
||
async def _fetch_one_query(client: httpx.AsyncClient, query: str, label: str) -> list[dict] | None:
|
||
"""Один Overpass-запрос с retry-backoff и фолбэком на зеркала (#1746).
|
||
|
||
На каждом endpoint из ``OVERPASS_ENDPOINTS`` делаем до
|
||
``_MAX_RETRIES_PER_ENDPOINT`` попыток, ретраясь по 429/5xx (особенно 504) с
|
||
экспоненциальным backoff + джиттер, уважая ``Retry-After``. Исчерпав попытки
|
||
на endpoint — переходим к следующему зеркалу. Возвращает list элементов при
|
||
успехе или None, если все endpoint'ы исчерпаны (caller логирует + продолжает,
|
||
partial-load, не abort — поведение #1746 сохранено).
|
||
"""
|
||
for endpoint in OVERPASS_ENDPOINTS:
|
||
for attempt in range(_MAX_RETRIES_PER_ENDPOINT):
|
||
try:
|
||
r = await client.post(endpoint, data={"data": query})
|
||
except httpx.HTTPError as e:
|
||
# Сетевой сбой / таймаут — это тоже повод ретраить/сменить зеркало.
|
||
last_attempt = attempt == _MAX_RETRIES_PER_ENDPOINT - 1
|
||
logger.warning(
|
||
"Overpass %s [%s] transport error (attempt %d/%d): %s",
|
||
label,
|
||
endpoint,
|
||
attempt + 1,
|
||
_MAX_RETRIES_PER_ENDPOINT,
|
||
e,
|
||
)
|
||
if last_attempt:
|
||
break # к следующему зеркалу
|
||
await asyncio.sleep(_backoff_delay(attempt, None))
|
||
continue
|
||
|
||
if r.status_code in _RETRYABLE_STATUS:
|
||
retry_after = _parse_retry_after(r.headers.get("Retry-After"))
|
||
last_attempt = attempt == _MAX_RETRIES_PER_ENDPOINT - 1
|
||
logger.warning(
|
||
"Overpass %s [%s] HTTP %d (attempt %d/%d, Retry-After=%s)",
|
||
label,
|
||
endpoint,
|
||
r.status_code,
|
||
attempt + 1,
|
||
_MAX_RETRIES_PER_ENDPOINT,
|
||
retry_after,
|
||
)
|
||
if last_attempt:
|
||
break # к следующему зеркалу
|
||
await asyncio.sleep(_backoff_delay(attempt, retry_after))
|
||
continue
|
||
|
||
# Не-ретраибл статус (4xx кроме 429 и т.п.) — поднимем как есть.
|
||
r.raise_for_status()
|
||
return r.json().get("elements", [])
|
||
|
||
logger.warning(
|
||
"Overpass %s: endpoint %s exhausted after %d attempts → next mirror",
|
||
label,
|
||
endpoint,
|
||
_MAX_RETRIES_PER_ENDPOINT,
|
||
)
|
||
|
||
return None
|
||
|
||
|
||
async def fetch_overpass_utility() -> list[dict]:
|
||
"""Запрашивает Overpass API для всех инж.-сетевых типов ЕКБ.
|
||
|
||
Возвращает enriched list dict'ов с доп-полями:
|
||
``_infrastructure_kind``, ``_source_tag`` — для UPSERT.
|
||
|
||
Каждый query-type идёт через ``_fetch_one_query`` (retry-backoff + фолбэк на
|
||
зеркала, #1746). Между РАЗНЫМИ query-types — пауза
|
||
``_INTER_QUERY_DELAY_SECONDS`` (gentler на публичный endpoint, чем прежний 1с).
|
||
Query, провалившийся на всех зеркалах, логируется и пропускается (partial-load,
|
||
не abort) — остальные query-types это не валит.
|
||
"""
|
||
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)
|
||
label = f"{key}={value} ({kind})"
|
||
try:
|
||
elements = await _fetch_one_query(client, query, label)
|
||
except Exception as e:
|
||
# Не-ретраибл HTTP-ошибка (raise_for_status) или неожиданный сбой —
|
||
# логируем и продолжаем, чтобы один битый query-type не валил весь sync.
|
||
logger.warning("Overpass utility failed for %s=%s: %s", key, value, e)
|
||
elements = None
|
||
|
||
if elements is None:
|
||
logger.warning(
|
||
"Overpass utility: %s skipped — all endpoints failed (retries+mirrors)",
|
||
label,
|
||
)
|
||
else:
|
||
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)
|
||
|
||
await asyncio.sleep(_INTER_QUERY_DELAY_SECONDS)
|
||
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")
|
||
# generic man_made=pipeline: реальный вид (gas vs heat) по substance —
|
||
# одна overpass-строка тянет и газопроводы, и теплотрассы (#2119).
|
||
if (key, value) == ("man_made", "pipeline"):
|
||
kind = _pipeline_kind_from_substance(tags, kind)
|
||
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)
|
||
|
||
|
||
# Junk-значения OSM tags, которые не несут смысла для попапа.
|
||
_JUNK_TAG_VALUES = frozenset({"", "yes", "no", "fixme", "unknown", "-"})
|
||
|
||
|
||
def _clean_utility_tag(value: object) -> str | None:
|
||
"""Нормализовать значение OSM tag: strip, None для пустого / junk."""
|
||
if value is None:
|
||
return None
|
||
s = str(value).strip()
|
||
return s if s and s.lower() not in _JUNK_TAG_VALUES else None
|
||
|
||
|
||
# OSM material (en) → RU («человеческим языком», правило ui-microcopy). Неизвестное
|
||
# значение остаётся как есть (лучше сырое, чем потерять).
|
||
_MATERIAL_RU: dict[str, str] = {
|
||
"metal": "металл",
|
||
"steel": "сталь",
|
||
"iron": "железо",
|
||
"cast_iron": "чугун",
|
||
"ductile_iron": "высокопрочный чугун",
|
||
"aluminium": "алюминий",
|
||
"aluminum": "алюминий",
|
||
"copper": "медь",
|
||
"concrete": "бетон",
|
||
"reinforced_concrete": "железобетон",
|
||
"wood": "дерево",
|
||
"wooden": "дерево",
|
||
"plastic": "пластик",
|
||
"pvc": "ПВХ",
|
||
"polyethylene": "полиэтилен",
|
||
"pe": "полиэтилен",
|
||
"asbestos_cement": "асбестоцемент",
|
||
"ceramic": "керамика",
|
||
}
|
||
|
||
|
||
def _format_voltage(raw: str) -> str | None:
|
||
"""OSM voltage (вольты, возможно multi «110000;10000») → «110 кВ/10 кВ».
|
||
|
||
≥1000 В → «<kV> кВ», иначе «<V> В». Multi-токены через «;» join'ятся «/».
|
||
Нечисловые токены пропускаются. None — если ни одного валидного.
|
||
"""
|
||
parts: list[str] = []
|
||
for token in raw.split(";"):
|
||
tok = token.strip()
|
||
if not tok:
|
||
continue
|
||
try:
|
||
volts = int(float(tok))
|
||
except (TypeError, ValueError):
|
||
continue # нечисловой мусор — пропускаем
|
||
if volts >= 1000:
|
||
kv = volts / 1000
|
||
kv_str = f"{kv:g}" # 110.0 → «110», 0.4 → «0.4»
|
||
parts.append(f"{kv_str} кВ")
|
||
else:
|
||
parts.append(f"{volts} В")
|
||
return "/".join(parts) if parts else None
|
||
|
||
|
||
def _format_diameter(raw: str) -> str | None:
|
||
"""Трубопровод diameter → «⌀ <n> мм». Берём ПЕРВЫЙ числовой run (отбрасываем
|
||
суффиксы «mm»/«мм» и не склеиваем мульти-размеры «500x300» → «500»). None — если
|
||
цифр нет.
|
||
"""
|
||
m = re.search(r"\d+", raw)
|
||
return f"⌀ {m.group()} мм" if m else None
|
||
|
||
|
||
# Теплоноситель теплотрассы (OSM substance) → RU для попапа. «heat» — обобщённая
|
||
# тепловая сеть (без уточнения носителя) — подписи не даём (kind=heat уже читается
|
||
# фронтом как «Теплотрасса»), чтобы не дублировать. hot_water / steam — уточняют.
|
||
_HEAT_SUBSTANCE_RU: dict[str, str] = {
|
||
"hot_water": "горячая вода",
|
||
"steam": "пар",
|
||
}
|
||
|
||
|
||
def _heat_substance_label(tags: dict) -> str | None:
|
||
"""Теплоноситель теплотрассы (hot_water→«горячая вода», steam→«пар»).
|
||
|
||
None для substance=heat (обобщённо — не уточняем) или неизвестного — попап и
|
||
так знает, что это теплотрасса (kind=heat → «Теплотрасса» на фронте).
|
||
"""
|
||
return _HEAT_SUBSTANCE_RU.get(_substance_of(tags))
|
||
|
||
|
||
def _utility_characteristics(kind: str, tags: dict | None) -> str | None:
|
||
"""Одна гуманизированная RU-строка ключевых характеристик OSM-сети.
|
||
|
||
Стиль совпадает с connection-point ``_build_structure_characteristics``
|
||
(quarter_dump_lookup): « · »-join непустых частей. Включает (при наличии):
|
||
теплоноситель (для heat), напряжение (multi «/»), оператор, диаметр (⌀ мм),
|
||
материал, кол-во цепей. None — если ничего пригодного.
|
||
|
||
Examples:
|
||
power voltage=110000;10000 operator=«МРСК Урала» →
|
||
«110 кВ/10 кВ · МРСК Урала»
|
||
gas diameter=500mm material=steel → «⌀ 500 мм · сталь»
|
||
heat substance=hot_water diameter=200 → «горячая вода · ⌀ 200 мм»
|
||
voltage=400 → «400 В»; voltage=yes → None
|
||
"""
|
||
if not isinstance(tags, dict):
|
||
return None
|
||
parts: list[str] = []
|
||
|
||
# Теплоноситель — только для теплотрасс (kind=heat), первым: «горячая вода» / «пар».
|
||
if kind == "heat":
|
||
heat_label = _heat_substance_label(tags)
|
||
if heat_label:
|
||
parts.append(heat_label)
|
||
|
||
voltage = _clean_utility_tag(tags.get("voltage"))
|
||
if voltage:
|
||
vstr = _format_voltage(voltage)
|
||
if vstr:
|
||
parts.append(vstr)
|
||
|
||
operator = _clean_utility_tag(tags.get("operator"))
|
||
if operator:
|
||
parts.append(operator)
|
||
|
||
diameter = _clean_utility_tag(tags.get("diameter"))
|
||
if diameter:
|
||
dstr = _format_diameter(diameter)
|
||
if dstr:
|
||
parts.append(dstr)
|
||
|
||
material = _clean_utility_tag(tags.get("material"))
|
||
if material:
|
||
parts.append(_MATERIAL_RU.get(material.lower(), material))
|
||
|
||
# Кол-во цепей/кабелей ЛЭП — только если значение чисто числовое.
|
||
circuits = _clean_utility_tag(tags.get("circuits")) or _clean_utility_tag(tags.get("cables"))
|
||
if circuits and circuits.isdigit():
|
||
parts.append(f"{circuits} цеп.")
|
||
|
||
return " · ".join(parts) if parts else None
|
||
|
||
|
||
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,
|
||
u.tags AS tags,
|
||
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,
|
||
"characteristics": _utility_characteristics(kind, r["tags"]),
|
||
"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,
|
||
},
|
||
}
|