Light1YT
d694e735cd
Loop UPSERT в sync_noise_sources_to_db без begin_nested: один замкнутый 3-точечный natural=water way [A,B,A] даёт POLYGON((A,B,A)), PostGIS отвергает (< 4 точек в ring) → outer tx rollback + raise → весь weekly noise/water/utility sync падает, тот же way отравляет каждый прогон. - Оборачиваем каждый UPSERT в `with db.begin_nested():` + per-row try/except → logger.warning + skipped++ (канон pzz_loader.py:111). - В _way_to_polygon_wkt проверяем итоговое кольцо ≥ 4 точек (fail-safe). - Outer except: добавлен logger.exception для видимости. Closes #1231
269 lines
12 KiB
Python
269 lines
12 KiB
Python
"""Загрузчик OSM источников шума из Overpass API для site-finder.
|
||
|
||
Запускается раз в неделю через Celery beat. Хранит дороги (way), ж/д пути,
|
||
промзоны и аэродромы в таблице osm_noise_sources_ekb.
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
|
||
import httpx
|
||
from sqlalchemy import text
|
||
|
||
from app.core.db import SessionLocal
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
OVERPASS_URL = "https://overpass-api.de/api/interpreter"
|
||
EKB_BBOX = (56.7, 60.5, 56.95, 60.75) # (south, west, north, east)
|
||
|
||
# Overpass блокирует дефолтный User-Agent (406) — явный UA с контактом.
|
||
_HEADERS = {
|
||
"User-Agent": "GenDesign-SiteFinder/1.0 (+https://gendsgn.ru)",
|
||
"Accept": "application/json",
|
||
}
|
||
|
||
# Описание запросов: (overpass_key, overpass_value, source_type, road_class_from_value)
|
||
_NOISE_QUERIES: list[tuple[str, str, str, str | None]] = [
|
||
("highway", "motorway", "highway", "motorway"),
|
||
("highway", "trunk", "highway", "trunk"),
|
||
("highway", "primary", "highway", "primary"),
|
||
("highway", "secondary", "highway", "secondary"),
|
||
("highway", "tertiary", "highway", "tertiary"),
|
||
("highway", "residential", "highway", "residential"),
|
||
("railway", "rail", "railway", None),
|
||
("landuse", "industrial", "industrial", None),
|
||
("aeroway", "aerodrome", "aerodrome", None),
|
||
# Гидрология — реки, ручьи, каналы (way-линии) и водоёмы (way-полигоны)
|
||
("waterway", "river", "water", "river"),
|
||
("waterway", "stream", "water", "stream"),
|
||
("waterway", "canal", "water", "canal"),
|
||
("natural", "water", "water", "lake_or_pond"),
|
||
# Инженерные сети — магистральные линии (для proximity-проверки доступности
|
||
# подключения / распознавания ЛЭП-охранных зон через ЗОУИТ типа 5/9)
|
||
("power", "line", "utility", "power_line"),
|
||
("power", "minor_line", "utility", "power_line"),
|
||
("man_made", "pipeline", "utility", "pipeline"),
|
||
# Сети ВКХ (водоснабжение / канализация) — узлы, не магистрали
|
||
("man_made", "water_works", "utility", "water_works"),
|
||
("man_made", "wastewater_plant", "utility", "wastewater_plant"),
|
||
("man_made", "water_tower", "utility", "water_tower"),
|
||
("man_made", "storage_tank", "utility", "storage_tank"),
|
||
# Электроподстанции — потенциальные точки подключения 10/35/110 кВ
|
||
("power", "substation", "utility", "substation"),
|
||
]
|
||
|
||
|
||
def _build_overpass_query(key: str, value: str, el_type: str) -> str:
|
||
"""Запрос Overpass для одного типа элемента.
|
||
|
||
Для way используем `out geom` — получаем полные координаты узлов (нужно
|
||
для построения LineString геометрии).
|
||
Для node — `out body`.
|
||
"""
|
||
south, west, north, east = EKB_BBOX
|
||
bbox = f"({south},{west},{north},{east})"
|
||
if el_type == "way":
|
||
return f"[out:json][timeout:30];" f'way["{key}"="{value}"]{bbox};' f"out geom;"
|
||
else:
|
||
return f"[out:json][timeout:30];" f'node["{key}"="{value}"]{bbox};' f"out body;"
|
||
|
||
|
||
async def fetch_overpass_noise() -> list[dict]:
|
||
"""Запрашивает Overpass API для всех шумовых источников ЕКБ.
|
||
|
||
Возвращает enriched list dict'ов с дополнительными полями:
|
||
_source_type, _road_class — для использования при UPSERT.
|
||
|
||
Между запросами sleep 1с (Overpass usage policy).
|
||
"""
|
||
all_elements: list[dict] = []
|
||
async with httpx.AsyncClient(timeout=60, headers=_HEADERS) as client:
|
||
for key, value, source_type, road_class in _NOISE_QUERIES:
|
||
# aerodrome — node; остальное — way
|
||
el_type = "node" if key == "aeroway" else "way"
|
||
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 noise: %s=%s (%s) → %d elements",
|
||
key,
|
||
value,
|
||
source_type,
|
||
len(elements),
|
||
)
|
||
for el in elements:
|
||
el["_source_type"] = source_type
|
||
el["_road_class"] = road_class
|
||
all_elements.extend(elements)
|
||
except Exception as e:
|
||
logger.warning("Overpass noise failed for %s=%s: %s", key, value, e)
|
||
await asyncio.sleep(1.0)
|
||
logger.info("Overpass noise: total %d elements", len(all_elements))
|
||
return all_elements
|
||
|
||
|
||
def _way_to_linestring_wkt(nodes: list[dict]) -> str | None:
|
||
"""Построить WKT LineString из списка узлов Overpass (out geom).
|
||
|
||
Каждый узел — {'lat': ..., 'lon': ...}. Минимум 2 точки для валидной линии.
|
||
"""
|
||
if len(nodes) < 2:
|
||
return None
|
||
# WKT LINESTRING требует запятую между парами точек: "lon1 lat1, lon2 lat2".
|
||
# Раньше joined через пробел → "lon1 lat1 lon2 lat2" → invalid geometry.
|
||
points = [f"{n['lon']} {n['lat']}" for n in nodes if "lon" in n and "lat" in n]
|
||
if len(points) < 2:
|
||
return None
|
||
return f"LINESTRING({', '.join(points)})"
|
||
|
||
|
||
def _way_to_polygon_wkt(nodes: list[dict]) -> str | None:
|
||
"""Построить WKT Polygon из замкнутого way Overpass (out geom).
|
||
|
||
Используется для natural=water (озёра, пруды). Если кольцо не замкнуто
|
||
автоматически — добавляем первую точку в конец.
|
||
|
||
PostGIS отвергает linear ring < 4 точек. Замкнутый way из 3 точек
|
||
[A, B, A] (points[0]==points[-1]) даёт POLYGON((A,B,A)) — "geometry
|
||
requires more points". Возвращаем None, чтобы caller сделал fallback
|
||
на LineString или пропустил элемент.
|
||
"""
|
||
points = [f"{n['lon']} {n['lat']}" for n in nodes if "lon" in n and "lat" in n]
|
||
if len(points) < 3:
|
||
return None
|
||
# Замкнуть кольцо если нужно
|
||
if points[0] != points[-1]:
|
||
points.append(points[0])
|
||
# Итоговое кольцо должно содержать ≥4 точек (PostGIS требование).
|
||
if len(points) < 4:
|
||
return None
|
||
return f"POLYGON(({', '.join(points)}))"
|
||
|
||
|
||
def sync_noise_sources_to_db() -> dict[str, int]:
|
||
"""UPSERT шумовых источников из Overpass в osm_noise_sources_ekb.
|
||
|
||
UPSERT по UNIQUE(osm_type, osm_id).
|
||
Геометрия:
|
||
- way с nodes → LineString WKT через ST_GeomFromText(..., 4326)
|
||
- node → ST_SetSRID(ST_MakePoint(lon, lat), 4326)
|
||
Записывает breadcrumb в nspd_geo_log (stage='noise_sync').
|
||
"""
|
||
elements = asyncio.run(fetch_overpass_noise())
|
||
|
||
inserted = 0
|
||
updated = 0
|
||
skipped = 0
|
||
fetched = len(elements)
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
for el in elements:
|
||
osm_id: int = el["id"]
|
||
osm_type: str = el["type"] # 'way' | 'node'
|
||
source_type: str = el.get("_source_type", "unknown")
|
||
road_class: str | None = el.get("_road_class")
|
||
tags: dict = el.get("tags") or {}
|
||
name: str | None = tags.get("name")
|
||
|
||
# Геометрия зависит от типа элемента
|
||
geom_sql: str
|
||
geom_params: dict
|
||
if osm_type == "way":
|
||
nodes: list[dict] = el.get("geometry") or []
|
||
# natural=water → замкнутый полигон; waterway → линия
|
||
if source_type == "water" and road_class == "lake_or_pond":
|
||
wkt = _way_to_polygon_wkt(nodes)
|
||
if wkt is None:
|
||
# Fallback: попробовать как линию (частично загруженный bbox-обрезок)
|
||
wkt = _way_to_linestring_wkt(nodes)
|
||
else:
|
||
wkt = _way_to_linestring_wkt(nodes)
|
||
if wkt is None:
|
||
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,
|
||
"source_type": source_type,
|
||
"road_class": road_class,
|
||
"name": name,
|
||
"tags": json.dumps(tags, ensure_ascii=False),
|
||
**geom_params,
|
||
}
|
||
|
||
try:
|
||
with db.begin_nested(): # SAVEPOINT — откат только этой записи
|
||
result = db.execute(
|
||
text(f"""
|
||
INSERT INTO osm_noise_sources_ekb
|
||
(osm_type, osm_id, source_type, road_class, name, geom, tags,
|
||
fetched_at)
|
||
VALUES (
|
||
:osm_type, :osm_id, :source_type, :road_class, :name,
|
||
{geom_sql},
|
||
CAST(:tags AS jsonb), NOW()
|
||
)
|
||
ON CONFLICT (osm_type, osm_id) DO UPDATE
|
||
SET source_type = EXCLUDED.source_type,
|
||
road_class = EXCLUDED.road_class,
|
||
name = EXCLUDED.name,
|
||
geom = EXCLUDED.geom,
|
||
tags = EXCLUDED.tags,
|
||
fetched_at = NOW()
|
||
RETURNING (xmax = 0) AS is_insert
|
||
"""),
|
||
params,
|
||
).scalar()
|
||
if result:
|
||
inserted += 1
|
||
else:
|
||
updated += 1
|
||
except Exception as e:
|
||
# Дефектный way (например, 3-точечное замкнутое кольцо POLYGON((A,B,A))
|
||
# отвергается PostGIS) не должен валить весь weekly-sync. SAVEPOINT
|
||
# откатывает только эту строку, продолжаем с остальными.
|
||
logger.warning(
|
||
"noise_sync insert failed for %s/%s (source_type=%s): %s",
|
||
osm_type,
|
||
osm_id,
|
||
source_type,
|
||
e,
|
||
)
|
||
skipped += 1
|
||
|
||
db.commit()
|
||
except Exception as e:
|
||
db.rollback()
|
||
logger.exception("noise_sync: unexpected error, outer tx rolled back: %s", e)
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
result_dict = {
|
||
"fetched": fetched,
|
||
"inserted": inserted,
|
||
"updated": updated,
|
||
"skipped": skipped,
|
||
}
|
||
logger.info("noise_sync done: %s", result_dict)
|
||
return result_dict
|