Light1YT
86e9ea2937
Многоагентный аудит + имплементация: один воркер на файл, точечные правки. Верификация: py_compile (47/47 .py) + tsc --noEmit (0 ошибок). Unit-тесты не прогонялись (окружение не поднято: rollup native dep / нет pytest-venv). Полностью исправлено (169): #1336, #1337, #1339, #1340, #1341, #1342, #1343, #1345, #1346, #1348, #1349, #1350, #1351, #1354, #1356, #1358, #1359, #1360, #1362, #1364, #1365, #1366, #1367, #1368, #1369, #1370, #1371, #1372, #1373, #1374, #1375, #1376, #1377, #1378, #1379, #1380, #1381, #1382, #1384, #1385, #1386, #1387, #1388, #1389, #1390, #1391, #1392, #1394, #1395, #1396, #1397, #1399, #1400, #1401, #1402, #1403, #1404, #1408, #1409, #1410, #1411, #1412, #1413, #1414, #1415, #1416, #1417, #1418, #1420, #1423, #1425, #1426, #1427, #1428, #1429, #1430, #1431, #1432, #1433, #1434, #1435, #1437, #1438, #1439, #1440, #1441, #1442, #1443, #1444, #1445, #1446, #1447, #1448, #1449, #1450, #1451, #1452, #1453, #1454, #1455, #1456, #1457, #1458, #1459, #1460, #1461, #1462, #1463, #1464, #1465, #1466, #1467, #1468, #1469, #1471, #1472, #1473, #1474, #1476, #1478, #1479, #1481, #1482, #1483, #1484, #1485, #1487, #1488, #1489, #1490, #1491, #1492, #1493, #1494, #1495, #1496, #1497, #1499, #1500, #1501, #1502, #1504, #1505, #1506, #1507, #1510, #1514, #1515, #1516, #1517, #1518, #1519, #1521, #1522, #1523, #1524, #1525, #1526, #1527, #1528, #1529, #1531, #1532, #1533, #1534, #1535, #1536, #1537, #1538 Частично (9, in-file часть, остаток cross-file): #1361, #1419, #1422, #1424, #1470, #1475, #1477, #1480, #1498 Требуют cross-file (3, не тронуты): #1338, #1363, #1421 Пропущено (1): #1539 Не входило в партию: 22 needs-Leha issue (нужны решения владельца). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
232 lines
10 KiB
Python
232 lines
10 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))
|
||
# Привязываем category именно к тому per-category запросу, под который
|
||
# элемент реально пришёл. Элемент с двумя целевыми тегами (например
|
||
# amenity=pharmacy + shop=supermarket) приходит дважды — каждая копия
|
||
# несёт свою category. Иначе _classify по dict-порядку молча терял бы
|
||
# вторую категорию при UPSERT по UNIQUE(osm_type, osm_id, category). См. #1372.
|
||
for el in elements:
|
||
el["_gd_category"] = category
|
||
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
|
||
skipped = 0
|
||
fetched = len(elements)
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
for el in elements:
|
||
tags: dict[str, str] = el.get("tags") or {}
|
||
# category проставлена в fetch_overpass под тот per-category запрос, под
|
||
# который элемент пришёл (multi-tag элемент приходит несколько раз, каждая
|
||
# копия со своей category). Fallback на _classify для прямых вызовов. См. #1372.
|
||
category = el.get("_gd_category") or _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
|
||
|
||
try:
|
||
with db.begin_nested(): # SAVEPOINT — откат только этой записи
|
||
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
|
||
except Exception as e:
|
||
# Дефектный OSM-элемент (битый tags-jsonb, нарушение constraint, PostGIS-
|
||
# ошибка у way/relation без корректного center) не должен валить весь
|
||
# weekly-sync. SAVEPOINT откатывает только эту строку, продолжаем
|
||
# с остальными. См. #1343 и .claude/rules/backend.md (SAVEPOINT pattern).
|
||
logger.warning(
|
||
"poi_sync insert failed for %s/%s (category=%s): %s",
|
||
osm_type,
|
||
osm_id,
|
||
category,
|
||
e,
|
||
)
|
||
skipped += 1
|
||
|
||
db.commit()
|
||
except Exception as e:
|
||
db.rollback()
|
||
logger.exception("poi_sync: unexpected error, outer tx rolled back: %s", e)
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
return {
|
||
"fetched": fetched,
|
||
"inserted": inserted,
|
||
"updated": updated,
|
||
"skipped_old": skipped_old,
|
||
"skipped": skipped,
|
||
}
|