gendesign/tradein-mvp/backend/app/services/geocoder.py
TradeIn Deploy d456e4fecb feat(tradein): выбор адреса на карте ЕКБ (#415)
Кнопка справа от поля адреса → модалка с картой Екатеринбурга.
Клик по дому → reverse-геокодинг → адрес подставляется в форму.

- geocoder.reverse_geocode + GET /api/v1/geocode/reverse — Nominatim
  /reverse (координаты → адрес).
- MapPicker.tsx — модалка с Leaflet + OSM-тайлами (Leaflet с CDN,
  без npm-зависимости), circleMarker по клику.
- EstimateForm — кнопка-карта справа от AddressInput.

Карта на OSM (без API-ключа). При появлении ключа Yandex (#402)
можно перейти на Я.Карты.

Closes #415
2026-05-22 12:09:35 +05:00

565 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Geocoder service — address → lat/lon.
Стратегия:
- Cache lookup в `geocode_cache` (Postgres) — TTL 90 дней
- Cache miss → Yandex Geocoder (если есть key) → fallback Nominatim
- Результат сохраняется в кэш для последующих вызовов
Используется в:
- /api/v1/trade-in/estimate (вход — адрес от пользователя)
- /api/v1/geocode/lookup (debug endpoint)
- scraper jobs (когда нужно по адресу определить координаты)
"""
from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass
from typing import Literal
import httpx
from sqlalchemy import text
from sqlalchemy.orm import Session
from tenacity import retry, stop_after_attempt, wait_exponential
from app.core.config import settings
logger = logging.getLogger(__name__)
# ── Result type ──────────────────────────────────────────────────────────────
@dataclass(frozen=True, slots=True)
class GeocodeResult:
lat: float
lon: float
full_address: str
provider: Literal["nominatim", "yandex", "cache"]
confidence: Literal["exact", "approximate", "locality"] = "approximate"
# ── Address normalisation ───────────────────────────────────────────────────
def normalize_address(address: str) -> str:
"""Нормализация для cache lookup: lowercase + trim + collapse whitespace.
« Ул. МАЛЫШЕВА, 30 » → «ул. малышева, 30»
"""
return " ".join(address.lower().strip().split())
# Согласные, которые часто пишут с одной буквой вместо двух (RU typos).
_DOUBLE_CONSONANTS = "лнмссккттпп"
def _typo_variants(query: str, limit: int = 6) -> list[str]:
"""Генерация вариантов с удвоением согласных — для случая когда пользователь
написал «Цвилинга» вместо «Цвиллинга», «Толстова» вместо «Толстого» итд.
Каждая позиция где есть одинокая согласная — кандидат на удвоение.
Возвращаем до `limit` вариантов в порядке вероятности (ближе к началу слова → выше).
"""
if len(query) < 3:
return []
variants: list[str] = []
chars = list(query)
for i in range(1, len(chars) - 1):
c = chars[i].lower()
if c not in _DOUBLE_CONSONANTS:
continue
prev_c = chars[i - 1].lower()
next_c = chars[i + 1].lower()
# Удваиваем только если соседи — гласные/'ь'/'ъ' (характерно для русских типо)
if prev_c in "аеёиоуыэюяьъ" and next_c in "аеёиоуыэюяьъ":
variant = query[:i + 1] + chars[i] + query[i + 1:]
if variant != query and variant not in variants:
variants.append(variant)
if len(variants) >= limit:
break
return variants
# ── Cache ────────────────────────────────────────────────────────────────────
def _cache_get(db: Session, address_norm: str) -> GeocodeResult | None:
row = db.execute(
text(
"""
SELECT lat, lon, full_address, provider, confidence
FROM geocode_cache
WHERE address_normalized = :addr
AND expires_at > NOW()
"""
),
{"addr": address_norm},
).fetchone()
if row is None:
return None
return GeocodeResult(
lat=row.lat,
lon=row.lon,
full_address=row.full_address,
provider="cache",
confidence=row.confidence or "approximate",
)
def _cache_put(db: Session, address_norm: str, result: GeocodeResult) -> None:
db.execute(
text(
"""
INSERT INTO geocode_cache
(address_normalized, lat, lon, full_address, provider, confidence)
VALUES (:addr, :lat, :lon, :full, :provider, :conf)
ON CONFLICT (address_normalized) DO UPDATE
SET lat = EXCLUDED.lat,
lon = EXCLUDED.lon,
full_address = EXCLUDED.full_address,
provider = EXCLUDED.provider,
confidence = EXCLUDED.confidence,
created_at = NOW(),
expires_at = NOW() + interval '90 days'
"""
),
{
"addr": address_norm,
"lat": result.lat,
"lon": result.lon,
"full": result.full_address,
"provider": result.provider,
"conf": result.confidence,
},
)
db.commit()
# ── Provider: Nominatim (OSM, без ключа) ────────────────────────────────────
async def _nominatim_query(client: httpx.AsyncClient, address: str) -> dict | None:
"""Single Nominatim search. Возвращает первый item или None.
ВАЖНО: фильтруем результаты по ЕКБ bbox прямо тут, чтобы при опечатках
не возвращать Пермский край / Челябинск.
"""
response = await client.get(
"https://nominatim.openstreetmap.org/search",
params={
"q": address,
"format": "json",
"limit": "3",
"countrycodes": "ru",
"addressdetails": "1",
"viewbox": EKB_BBOX["viewbox"],
"bounded": "1", # строго в ЕКБ bbox
},
)
response.raise_for_status()
data = response.json()
for item in data:
try:
lat_f = float(item["lat"])
lon_f = float(item["lon"])
if 60.40 <= lon_f <= 60.85 and 56.65 <= lat_f <= 56.95:
return item
except Exception: # noqa: BLE001
continue
return None
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=8))
async def _nominatim_lookup(address: str) -> GeocodeResult | None:
"""OSM Nominatim — бесплатно, без ключа, 1 req/sec policy.
Бан-policy: User-Agent с email обязателен.
Tier 1: bounded ЕКБ на оригинальный адрес.
Tier 2: bounded ЕКБ на typo-варианты (Цвилинга → Цвиллинга).
"""
headers = {
"User-Agent": f"TradeInMVP/0.1 (contact: {settings.contact_email})",
"Accept": "application/json",
"Accept-Language": "ru,en;q=0.8",
"Referer": "https://tradein-mvp.local/",
}
async with httpx.AsyncClient(timeout=10.0, headers=headers) as client:
# Tier 1: оригинал
item = await _nominatim_query(client, address)
# Tier 2: typo-variants
if item is None:
for variant in _typo_variants(address, limit=4):
await asyncio.sleep(1.0) # Nominatim 1 req/sec policy
item = await _nominatim_query(client, variant)
if item is not None:
logger.info("nominatim typo-fixed: %s%s", address, variant)
break
if item is None:
return None
confidence = "exact" if item.get("class") == "building" else "approximate"
return GeocodeResult(
lat=float(item["lat"]),
lon=float(item["lon"]),
full_address=item.get("display_name", address),
provider="nominatim",
confidence=confidence,
)
# ── Provider: Yandex Geocoder (требует key, лучшее покрытие РФ) ─────────────
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=8))
async def _yandex_lookup(address: str, api_key: str) -> GeocodeResult | None:
"""Yandex Geocoder — 25K req/day free для самопод, лучше РФ.
Docs: https://yandex.ru/dev/maps/geocoder/doc/desc/concepts/input_params.html
Запрашиваем с ll+spn (центр ЕКБ) для приоритизации местных результатов,
но БЕЗ rspn — чтобы fuzzy matching работал при опечатках.
"""
# Не запихиваем "Екатеринбург" если оно уже есть в адресе (типичный кейс из suggest)
geocode_query = address if "екатеринбург" in address.lower() else f"Екатеринбург, {address}"
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
"https://geocode-maps.yandex.ru/1.x/",
params={
"apikey": api_key,
"geocode": geocode_query,
"format": "json",
"results": 5, # берем top-5, отфильтруем по ЕКБ bbox ниже
"lang": "ru_RU",
"ll": EKB_BBOX["ll"],
"spn": EKB_BBOX["spn"],
},
)
response.raise_for_status()
data = response.json()
members = data.get("response", {}).get("GeoObjectCollection", {}).get("featureMember", [])
if not members:
return None
# Фильтруем top-5 по ЕКБ bbox — игнорируем Челябинск/Уфу/Москву при опечатке
best = None
for m in members:
obj = m.get("GeoObject", {})
try:
lon_str, lat_str = obj["Point"]["pos"].split()
lat_f, lon_f = float(lat_str), float(lon_str)
if 60.40 <= lon_f <= 60.85 and 56.65 <= lat_f <= 56.95:
best = obj
break
except Exception: # noqa: BLE001
continue
if best is None:
# Никто из top-5 не попал в ЕКБ → берем первый «как есть» (вне ЕКБ — но хоть что-то)
best = members[0]["GeoObject"]
lon_str, lat_str = best["Point"]["pos"].split()
precision_raw = (
best.get("metaDataProperty", {})
.get("GeocoderMetaData", {})
.get("precision", "other")
)
confidence_map = {
"exact": "exact",
"number": "exact",
"near": "approximate",
"range": "approximate",
"street": "approximate",
}
return GeocodeResult(
lat=float(lat_str),
lon=float(lon_str),
full_address=best.get("metaDataProperty", {})
.get("GeocoderMetaData", {})
.get("text", address),
provider="yandex",
confidence=confidence_map.get(precision_raw, "approximate"),
)
# ── Suggest (автокомплит) ───────────────────────────────────────────────────
# ЕКБ bounding box (приблизительно): юг 56.65, запад 60.40, север 56.95, восток 60.85
EKB_BBOX = {
"viewbox": "60.40,56.95,60.85,56.65", # Nominatim format: lon1,lat1,lon2,lat2 (NW,SE)
"ll": "60.605,56.838", # Yandex center (lon,lat)
"spn": "0.45,0.30", # Yandex span (lon,lat)
}
@dataclass(frozen=True, slots=True)
class GeocodeSuggestion:
label: str # формат "Малышева 30, Октябрьский район"
full_address: str # полный из геокодера
lat: float
lon: float
kind: str # 'house' / 'street' / 'locality'
def _parse_yandex_members(members: list[dict]) -> list[GeocodeSuggestion]:
"""Yandex geocode_members → list[GeocodeSuggestion]. Чистим описание от мусора."""
out: list[GeocodeSuggestion] = []
for m in members:
obj = m.get("GeoObject", {})
try:
lon_str, lat_str = obj["Point"]["pos"].split()
meta = obj.get("metaDataProperty", {}).get("GeocoderMetaData", {})
kind = meta.get("kind", "other")
full = meta.get("text", obj.get("name", ""))
name = obj.get("name", full)
desc = obj.get("description", "")
desc_parts = [
p.strip() for p in desc.split(",")
if p.strip() and p.strip() not in {"Россия", "Свердловская область"}
]
label = name if not desc_parts else f"{name} · {', '.join(desc_parts)}"
out.append(GeocodeSuggestion(
label=label, full_address=full,
lat=float(lat_str), lon=float(lon_str), kind=kind,
))
except Exception: # noqa: BLE001
continue
return out
async def _yandex_geocode_request(
client: httpx.AsyncClient, api_key: str, query: str, limit: int, bounded: bool
) -> list[dict]:
"""Single Yandex Geocoder request — bounded=True → строго в ЕКБ через rspn=1."""
params: dict[str, str] = {
"apikey": api_key,
"geocode": query,
"format": "json",
"results": str(limit),
"lang": "ru_RU",
"ll": EKB_BBOX["ll"],
"spn": EKB_BBOX["spn"],
}
if bounded:
params["rspn"] = "1"
response = await client.get("https://geocode-maps.yandex.ru/1.x/", params=params)
response.raise_for_status()
data = response.json()
return data.get("response", {}).get("GeoObjectCollection", {}).get("featureMember", [])
@retry(stop=stop_after_attempt(2), wait=wait_exponential(multiplier=1, min=1, max=4))
async def _yandex_suggest(query: str, api_key: str, limit: int = 8) -> list[GeocodeSuggestion]:
"""Yandex Geocoder с авто-fallback на typo-tolerant режим.
Tier 1: bounded ЕКБ (rspn=1) на оригинальный query.
Tier 2: bounded ЕКБ на typo-variants (удвоение согласных).
Tier 3: без rspn — fuzzy по всей стране, фильтр результатов по ЕКБ bbox.
"""
async with httpx.AsyncClient(timeout=8.0) as client:
# Tier 1: strict bounded на оригинал
members = await _yandex_geocode_request(
client, api_key, f"Екатеринбург, {query}", limit, bounded=True,
)
results = _parse_yandex_members(members)
if results:
return results
# Tier 2: bounded на typo-варианты
for variant in _typo_variants(query, limit=4):
members = await _yandex_geocode_request(
client, api_key, f"Екатеринбург, {variant}", limit, bounded=True,
)
results = _parse_yandex_members(members)
if results:
return results
# Tier 3: без rspn — даём fuzzy сделать своё дело, фильтр по bbox
members = await _yandex_geocode_request(
client, api_key, f"Екатеринбург, {query}", limit, bounded=False,
)
results = _parse_yandex_members(members)
in_ekb = [
r for r in results
if 60.40 <= r.lon <= 60.85 and 56.65 <= r.lat <= 56.95
]
return in_ekb
async def _nominatim_query_multi(
client: httpx.AsyncClient, query: str, limit: int
) -> list[dict]:
"""Один Nominatim search с фильтром по ЕКБ bbox. Возвращает up to N items."""
response = await client.get(
"https://nominatim.openstreetmap.org/search",
params={
"q": query,
"format": "json",
"limit": str(limit),
"countrycodes": "ru",
"viewbox": EKB_BBOX["viewbox"],
"bounded": "1",
"addressdetails": "1",
},
)
response.raise_for_status()
data = response.json()
return data if isinstance(data, list) else []
@retry(stop=stop_after_attempt(2), wait=wait_exponential(multiplier=1, min=1, max=4))
async def _nominatim_suggest(query: str, limit: int = 8) -> list[GeocodeSuggestion]:
"""Nominatim в режиме suggest. С typo-fallback (для случаев когда Yandex недоступен)."""
headers = {
"User-Agent": f"TradeInMVP/0.1 (contact: {settings.contact_email})",
"Accept": "application/json",
"Accept-Language": "ru,en;q=0.8",
}
async with httpx.AsyncClient(timeout=8.0, headers=headers) as client:
# Tier 1: оригинальный query
data = await _nominatim_query_multi(client, f"{query}, Екатеринбург", limit)
# Tier 2: typo-варианты если оригинал пустой
if not data:
for variant in _typo_variants(query, limit=3):
await asyncio.sleep(1.0) # Nominatim 1 req/sec
data = await _nominatim_query_multi(client, f"{variant}, Екатеринбург", limit)
if data:
logger.info("nominatim suggest typo-fixed: %s%s", query, variant)
break
out: list[GeocodeSuggestion] = []
for item in data:
display = item.get("display_name", "")
addr = item.get("address", {}) or {}
# Компактный лейбл: street + house_number / locality / district
street = addr.get("road") or addr.get("street") or ""
house = addr.get("house_number", "")
district = addr.get("suburb") or addr.get("city_district") or addr.get("borough") or ""
parts = []
if street:
parts.append(f"{street}{f', {house}' if house else ''}")
elif item.get("name"):
parts.append(item["name"])
if district:
parts.append(district)
label = " · ".join(parts) if parts else display[:80]
kind = "house" if house else ("street" if street else "locality")
out.append(GeocodeSuggestion(
label=label, full_address=display,
lat=float(item["lat"]), lon=float(item["lon"]), kind=kind,
))
return out
async def suggest(query: str, limit: int = 8) -> list[GeocodeSuggestion]:
"""Автокомплит адресов в ЕКБ. Yandex → Nominatim → [].
Без кэша (дешёво, провайдеры толерируют автокомплит-запросы).
"""
if not query or len(query.strip()) < 2:
return []
if settings.yandex_geocoder_key:
try:
results = await _yandex_suggest(query, settings.yandex_geocoder_key, limit)
if results:
return results
except Exception: # noqa: BLE001
logger.exception("yandex suggest failed, falling back to nominatim")
try:
return await _nominatim_suggest(query, limit)
except Exception: # noqa: BLE001
logger.exception("nominatim suggest failed")
return []
# ── Public API ───────────────────────────────────────────────────────────────
async def geocode(address: str, db: Session) -> GeocodeResult | None:
"""Геокодинг с кэшем. Yandex → Nominatim → None.
Args:
address: пользовательский ввод (может быть грязным — нормализуем).
db: сессия Postgres для cache lookup/write.
Returns:
GeocodeResult или None если ни один провайдер не отвечает.
"""
if not address or len(address.strip()) < 3:
return None
addr_norm = normalize_address(address)
# 1. Cache
cached = _cache_get(db, addr_norm)
if cached is not None:
logger.info("geocode cache hit: %s", addr_norm)
return cached
# 2. Yandex (если есть key) с typo-fallback
if settings.yandex_geocoder_key:
try:
result = await _yandex_lookup(address, settings.yandex_geocoder_key)
# Если результат вне ЕКБ — пробуем typo-варианты
in_ekb = result is not None and 60.40 <= result.lon <= 60.85 and 56.65 <= result.lat <= 56.95
if result is not None and in_ekb:
_cache_put(db, addr_norm, result)
logger.info("geocode yandex: %s → (%.5f, %.5f)", addr_norm, result.lat, result.lon)
return result
# Tier 2: typo-variants
for variant in _typo_variants(address, limit=4):
try:
result = await _yandex_lookup(variant, settings.yandex_geocoder_key)
except Exception: # noqa: BLE001
continue
if result is None:
continue
if 60.40 <= result.lon <= 60.85 and 56.65 <= result.lat <= 56.95:
_cache_put(db, addr_norm, result)
logger.info(
"geocode yandex typo-fixed: %s%s → (%.5f, %.5f)",
addr_norm, variant, result.lat, result.lon,
)
return result
except Exception: # noqa: BLE001
logger.exception("yandex geocoder failed, falling back to nominatim")
# 3. Nominatim fallback
try:
result = await _nominatim_lookup(address)
if result is not None:
_cache_put(db, addr_norm, result)
logger.info(
"geocode nominatim: %s → (%.5f, %.5f)", addr_norm, result.lat, result.lon
)
# Nominatim rate-limit policy: 1 req/sec — спим после успешного запроса
await asyncio.sleep(1.0)
return result
except Exception: # noqa: BLE001
logger.exception("nominatim geocoder failed")
return None
# ── Reverse: координаты → адрес (для map-picker'а) ──────────────────────────
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=8))
async def reverse_geocode(lat: float, lon: float) -> str | None:
"""Обратный геокодинг: координаты → адрес (Nominatim /reverse).
Клик по дому на карте ЕКБ → адрес подставляется в форму оценки.
"""
headers = {
"User-Agent": f"TradeInMVP/0.1 (contact: {settings.contact_email})",
"Accept": "application/json",
"Accept-Language": "ru,en;q=0.8",
}
async with httpx.AsyncClient(timeout=10.0, headers=headers) as client:
response = await client.get(
"https://nominatim.openstreetmap.org/reverse",
params={
"lat": str(lat),
"lon": str(lon),
"format": "json",
"addressdetails": "1",
"zoom": "18",
},
)
response.raise_for_status()
data = response.json()
if not isinstance(data, dict) or "error" in data:
return None
return data.get("display_name")