205 lines
8 KiB
Python
205 lines
8 KiB
Python
"""Загрузчик OSM POI из Overpass API для site-finder.
|
||
|
||
Запускается раз в неделю через Celery beat. Поддерживает фильтр
|
||
"не старше 2 лет" (требование Максима) — last_osm_edit_date.
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
from datetime import date, datetime, timedelta
|
||
|
||
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)
|
||
|
||
# Маппинг OSM-тег → нормализованная category
|
||
OSM_CATEGORIES: dict[tuple[str, str], str] = {
|
||
# amenity tags — школы расширены (school/college/university)
|
||
("amenity", "school"): "school",
|
||
("amenity", "college"): "school",
|
||
("amenity", "university"): "school",
|
||
("amenity", "kindergarten"): "kindergarten",
|
||
("amenity", "pharmacy"): "pharmacy",
|
||
("amenity", "hospital"): "hospital",
|
||
("amenity", "clinic"): "hospital",
|
||
# shop tags — supermarket расширен
|
||
("shop", "mall"): "shop_mall",
|
||
("shop", "supermarket"): "shop_supermarket",
|
||
("shop", "hypermarket"): "shop_supermarket",
|
||
("shop", "convenience"): "shop_small",
|
||
("shop", "bakery"): "shop_small",
|
||
# leisure
|
||
("leisure", "park"): "park",
|
||
# transit
|
||
("railway", "tram_stop"): "tram_stop",
|
||
("highway", "bus_stop"): "bus_stop",
|
||
# метро (одна линия в ЕКБ, но добавляем для полноты)
|
||
("station", "subway"): "metro_stop",
|
||
}
|
||
|
||
|
||
def _build_overpass_query_single(key: str, value: str) -> str:
|
||
"""Запрос для одной пары tag → нормированной категории.
|
||
|
||
Раньше делали один большой запрос на все 14 категорий — Overpass возвращал
|
||
504 Gateway Timeout (запрос слишком тяжёлый). Сплит на per-category даёт
|
||
14 быстрых запросов вместо одного 60+ секундного.
|
||
"""
|
||
south, west, north, east = EKB_BBOX
|
||
bbox = f"({south},{west},{north},{east})"
|
||
return (
|
||
f"[out:json][timeout:30];"
|
||
f'(node["{key}"="{value}"]{bbox};way["{key}"="{value}"]{bbox};);'
|
||
f"out center meta;"
|
||
)
|
||
|
||
|
||
def _classify(tags: dict[str, str]) -> str | None:
|
||
"""Определить category из OSM-тегов. None если не соответствует ни одной."""
|
||
for (k, v), cat in OSM_CATEGORIES.items():
|
||
if tags.get(k) == v:
|
||
return cat
|
||
return None
|
||
|
||
|
||
async def fetch_overpass() -> list[dict]:
|
||
"""Запросить Overpass API per category, вернуть combined список elements.
|
||
|
||
Делаем 14 отдельных запросов вместо одного гигантского — большой запрос
|
||
отдаёт 504 Gateway Timeout. Между запросами sleep 1с (Overpass usage
|
||
policy: max 2 concurrent, лучше 1 req/s).
|
||
|
||
Overpass блокирует default `python-httpx/*` User-Agent (406) — поэтому
|
||
явный UA с контактом проекта.
|
||
"""
|
||
headers = {
|
||
"User-Agent": "GenDesign-SiteFinder/1.0 (+https://gendsgn.ru)",
|
||
"Accept": "application/json",
|
||
}
|
||
all_elements: list[dict] = []
|
||
async with httpx.AsyncClient(timeout=60, headers=headers) as client:
|
||
for (key, value), category in OSM_CATEGORIES.items():
|
||
query = _build_overpass_query_single(key, value)
|
||
try:
|
||
r = await client.post(OVERPASS_URL, data={"data": query})
|
||
r.raise_for_status()
|
||
elements: list[dict] = r.json().get("elements", [])
|
||
logger.info("Overpass: %s=%s (%s) → %d", key, value, category, len(elements))
|
||
all_elements.extend(elements)
|
||
except Exception as e:
|
||
# Не падаем на одной категории — логируем и продолжаем
|
||
logger.warning("Overpass failed for %s=%s: %s", key, value, e)
|
||
await asyncio.sleep(1.0)
|
||
logger.info(
|
||
"Overpass: total %d elements across %d categories", len(all_elements), len(OSM_CATEGORIES)
|
||
)
|
||
return all_elements
|
||
|
||
|
||
def sync_poi_to_db() -> dict[str, int]:
|
||
"""Синхронизирует POI из Overpass в osm_poi_ekb.
|
||
|
||
UPSERT по UNIQUE(osm_type, osm_id, category).
|
||
Returns: counters {fetched, inserted, updated, skipped_old}.
|
||
"""
|
||
elements = asyncio.run(fetch_overpass())
|
||
# 730 дней ≈ 2 года: избегаем ValueError 29 февраля (year-2 не високосный → нет 29.02).
|
||
# Точность ±1 день несущественна для фильтра "не старше 2 лет" (требование Максима). См. #1232.
|
||
two_years_ago = date.today() - timedelta(days=730)
|
||
|
||
inserted = 0
|
||
updated = 0
|
||
skipped_old = 0
|
||
fetched = len(elements)
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
for el in elements:
|
||
tags: dict[str, str] = el.get("tags") or {}
|
||
category = _classify(tags)
|
||
if not category:
|
||
continue
|
||
|
||
osm_id: int = el["id"]
|
||
osm_type: str = el["type"] # 'node' | 'way' | 'relation'
|
||
|
||
# node имеет lat/lon напрямую, way — через center
|
||
if osm_type == "node":
|
||
lat = el.get("lat")
|
||
lon = el.get("lon")
|
||
else:
|
||
center: dict = el.get("center") or {}
|
||
lat = center.get("lat")
|
||
lon = center.get("lon")
|
||
|
||
if lat is None or lon is None:
|
||
continue
|
||
|
||
# timestamp последней правки элемента
|
||
ts: str | None = el.get("timestamp")
|
||
last_edit: date | None = None
|
||
if ts:
|
||
try:
|
||
last_edit = datetime.fromisoformat(ts.replace("Z", "+00:00")).date()
|
||
except Exception:
|
||
last_edit = None
|
||
|
||
# Мягкий фильтр "не старше 2 лет" — данные сохраняем, помечаем для API
|
||
if last_edit and last_edit < two_years_ago:
|
||
skipped_old += 1
|
||
|
||
result = db.execute(
|
||
text("""
|
||
INSERT INTO osm_poi_ekb
|
||
(osm_id, osm_type, category, name, lat, lon, geom,
|
||
last_osm_edit_date, tags, fetched_at)
|
||
VALUES (:osm_id, :osm_type, :category, :name, :lat, :lon,
|
||
ST_SetSRID(ST_MakePoint(:lon, :lat), 4326),
|
||
:last_edit, CAST(:tags AS jsonb), NOW())
|
||
ON CONFLICT (osm_type, osm_id, category) DO UPDATE
|
||
SET name = EXCLUDED.name,
|
||
lat = EXCLUDED.lat,
|
||
lon = EXCLUDED.lon,
|
||
geom = EXCLUDED.geom,
|
||
last_osm_edit_date = EXCLUDED.last_osm_edit_date,
|
||
tags = EXCLUDED.tags,
|
||
fetched_at = NOW()
|
||
RETURNING (xmax = 0) AS is_insert
|
||
"""),
|
||
{
|
||
"osm_id": osm_id,
|
||
"osm_type": osm_type,
|
||
"category": category,
|
||
"name": tags.get("name"),
|
||
"lat": lat,
|
||
"lon": lon,
|
||
"last_edit": last_edit,
|
||
"tags": json.dumps(tags, ensure_ascii=False),
|
||
},
|
||
).scalar()
|
||
|
||
if result:
|
||
inserted += 1
|
||
else:
|
||
updated += 1
|
||
|
||
db.commit()
|
||
except Exception:
|
||
db.rollback()
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
return {
|
||
"fetched": fetched,
|
||
"inserted": inserted,
|
||
"updated": updated,
|
||
"skipped_old": skipped_old,
|
||
}
|