fix(tradein/geocoder): раз-хардкод геокодера/DaData на всю Свердловскую обл. (region 66)
All checks were successful
CI / changes (pull_request) Successful in 7s
CI / backend-tests (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 50s
CI Trade-In / changes (pull_request) Successful in 8s
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
CI Trade-In / frontend-checks (pull_request) Has been skipped

Геокодер был жёстко EKB-bound и ВЫБРАСЫВАЛ корректные non-EKB геокоды (bbox-accept-фильтры) → блокировал оценку любого адреса области. Обобщено до region-66.

- OBLAST66_BBOX + is_within_oblast66_bbox (superset EKB-tight, EKB не затронут); EKB Yandex Tier1/2 fast-path сохранён.
- Все accept-фильтры (Yandex/Nominatim forward+suggest, geocode() accept+typo) → oblast-66.
- City-prefix по word-boundary _has_oblast_marker (не substring): EKB-улицы 'Серова 27'/'Ирбитская 5' и др. больше не роняют 'Екатеринбург,'-префикс (был silent wrong-city баг).
- Two-pass tie-break в accept-циклах: tight-EKB кандидат приоритетнее → EKB-результат идентичен прежнему.
- Region cross-check (Nominatim address.state / Yandex AdministrativeAreaName) отсекает соседние области (Тюмень и др.); bbox-fallback когда region отсутствует.
- DaData suggest: region-параметр (hard-filter), убран no-op restrict_value.
- +26 тестов: OBLAST66 superset TIGHT, far-town accept, Тюмень reject, marker +/- (street-collision), DaData region body.

Foundation для A2 (геокодинг non-EKB сделок) и live-оценки адресов области.
This commit is contained in:
bot-backend 2026-07-12 18:33:05 +03:00
parent 4f205bd7da
commit c0cbdc2f99
5 changed files with 535 additions and 59 deletions

View file

@ -56,10 +56,12 @@ async def suggest_addresses(
limit: Annotated[int, Query(ge=1, le=15)] = 8,
db: Annotated[Session, Depends(get_db)] = None, # type: ignore[assignment]
) -> SuggestResponse:
"""Автокомплит адресов в пределах ЕКБ.
"""Автокомплит адресов в Свердловской области (region 66; ЕКБ — основной трафик,
остаётся быстрым fast-path).
Используется в EstimateForm для подсказок пока пользователь печатает.
Bounded viewbox = ЕКБ (lon 60.40-60.85, lat 56.65-56.95).
Bounded viewbox генеральный по всей области (см. geocoder.OBLAST66_VIEWBOX),
ЕКБ (lon 60.40-60.85, lat 56.65-56.95) внутри него остаётся быстрым fast-path.
Пример:
/api/v1/geocode/suggest?q=Малышева

View file

@ -299,7 +299,10 @@ def _parse_suggestion(item: dict[str, Any]) -> DadataSuggestion | None:
async def suggest_addresses(
query: str, limit: int = 8, city: str = "Екатеринбург"
query: str,
limit: int = 8,
city: str | None = "Екатеринбург",
region: str | None = None,
) -> list[DadataSuggestion]:
"""Автокомплит адресов через DaData /suggest/address.
@ -307,6 +310,13 @@ async def suggest_addresses(
query: текст для подсказки (минимум 2 символа).
limit: сколько вариантов вернуть (DaData параметр `count`, max 20).
city: locality для constraint (передаётся в `locations`). По умолчанию ЕКБ.
Игнорируется, если задан `region`.
region: область для hard-constraint в `locations` используется вместо
`city`, когда нужен охват шире одного города (например, вся
Свердловская область / region 66). DaData `locations` сам по себе уже
жёсткий фильтр (не boost) на уровне указанного admin-поля доп.
параметров не требуется. По умолчанию не задан поведение (и body
запроса) для существующих вызовов не меняется.
Returns:
list[DadataSuggestion] пустой список если:
@ -334,7 +344,13 @@ async def suggest_addresses(
"query": query.strip(),
"count": max(1, min(int(limit), 20)),
}
if city:
if region:
# `locations` с полем region — уже жёсткий фильтр сам по себе (DaData
# ограничивает выдачу этим регионом). `restrict_value` — ТОП-LEVEL параметр
# body (не ключ внутри объекта locations) — здесь он был бы silent no-op,
# поэтому не добавляем.
body["locations"] = [{"region": region}]
elif city:
body["locations"] = [{"city": city}]
try:
@ -352,9 +368,7 @@ async def suggest_addresses(
logger.warning("dadata suggest: HTTP 429 — quota exceeded (10k/день free?)")
return []
if status in (401, 403):
logger.error(
"dadata suggest: HTTP %d — auth rejected. Проверь DADATA_API_TOKEN.", status
)
logger.error("dadata suggest: HTTP %d — auth rejected. Проверь DADATA_API_TOKEN.", status)
return []
if status >= 500:
logger.warning("dadata suggest: HTTP %d — transient server error", status)

View file

@ -72,6 +72,112 @@ def is_within_ekb_bbox_wide(lat: float, lon: float) -> bool:
return is_within_ekb_bbox(lat, lon, EKB_BBOX_WIDE)
# ── Sverdlovsk oblast (region 66) ────────────────────────────────────────────
# Генеральный bbox всей области — от южной границы (ЕКБ/Полевской) до
# северных городов (Ивдель/Североуральск/Краснотурьинск). Специально щедрый:
# используется как fallback-accept для провайдеров БЕЗ структурного region-поля
# (сырые Point.pos от Yandex/Nominatim) — когда нет ФИАС/КЛАДР для точной
# проверки region==66. Дороже false-positive у самой границы соседних регионов,
# дешевле false-negative (выбросить корректный областной адрес — исходный баг).
# bbox = (lat_min, lat_max, lon_min, lon_max).
OBLAST66_BBOX = (55.8, 62.2, 56.7, 66.6)
# Nominatim/Yandex viewport для областного (не только ЕКБ) поиска — тот же
# формат что EKB_BBOX ниже, но покрывает всю область. viewbox — Nominatim
# (lon1,lat1,lon2,lat2, NW→SE); ll/spn — Yandex центр+охват (lon,lat).
OBLAST66_VIEWBOX = {
"viewbox": "56.7,62.2,66.6,55.8",
"ll": "61.65,59.0",
"spn": "9.9,6.4",
}
def is_within_oblast66_bbox(lat: float, lon: float) -> bool:
"""True если (lat, lon) в генеральном bbox Свердловской области (регион 66)."""
return is_within_ekb_bbox(lat, lon, OBLAST66_BBOX)
# Крупные города области (кроме ЕКБ) — НЕ исчерпывающий список (в области ~30
# городов), покрывает основные региональные центры. Используется чтобы:
# (a) не навязывать "Екатеринбург, " в запрос, если пользователь уже указал
# другой город региона (геокодер forward/suggest prefix-логика);
# (b) как known-locality set в _yandex_reverse при перестановке "улица, город".
SVERDLOVSK_OBLAST_CITIES = frozenset(
{
"екатеринбург",
"нижний тагил",
"каменск-уральский",
"первоуральск",
"серов",
"новоуральск",
"ревда",
"полевской",
"асбест",
"богданович",
"ирбит",
"красноуфимск",
"верхняя пышма",
"берёзовский",
"березовский",
"заречный",
"качканар",
"краснотурьинск",
"североуральск",
"ивдель",
"тавда",
"туринск",
"сысерть",
"среднеуральск",
"дегтярск",
"верхняя салда",
"нижняя салда",
"невьянск",
"артёмовский",
"артемовский",
"камышлов",
}
)
SVERDLOVSK_OBLAST_REGION = "Свердловская область"
# Word/phrase-boundary regex — НЕ substring — чтобы «Серова 27» не матчил город
# «Серов», «Ирбитская 5» — «Ирбит», «Асбестовский пер.» — «Асбест», «Невьянский
# пер.» — «Невьянск», «Богдановича» — «Богданович» (все реальные EKB-адреса,
# substring-матч ложно ронял "Екатеринбург, "-префикс на них → неверный город
# → неверная цена). Python `\b` для str-паттернов юникод-осведомлён (Cyrillic
# входит в `\w`), поэтому «серова» корректно НЕ матчит `\bсеров\b` (после
# «серов» идёт «а» — тоже \w, границы нет).
_OBLAST_CITY_RE = re.compile(
r"\b(?:" + "|".join(re.escape(city) for city in SVERDLOVSK_OBLAST_CITIES) + r")\b"
)
_OBLAST_MARKER_RE = re.compile(r"\bсвердловск\w*\b")
# Приставки, после которых токен обозначает РАЙОН/микрорайон города, а не сам
# город региона: «мкр Заречный» — микрорайон ВНУТРИ другого города (не ЗАТО
# Заречный). Без этого исключения точное совпадение слова "заречный" всё равно
# ложно сработало бы (это не substring-баг, а реальная омонимия имени).
_DISTRICT_PREFIXES = frozenset({"мкр", "мкр.", "микрорайон", "р", "р-он", "район", "жк"})
def _has_oblast_marker(text_lower: str) -> bool:
"""True если текст уже содержит упоминание области/города региона 66.
Используется чтобы НЕ навязывать "Екатеринбург, " в запрос, когда адрес
уже привязан к другому городу/области иначе получим двойной город
("Екатеринбург, Нижний Тагил, Ленина 10") и провайдер вернёт мусор/пусто.
Матчинг по границе слова/фразы (см. `_OBLAST_CITY_RE`), НЕ substring
и с исключением "мкр/микрорайон/р-н <город>" (район ВНУТРИ другого города).
"""
normalized = " ".join(text_lower.split())
if _OBLAST_MARKER_RE.search(normalized):
return True
for m in _OBLAST_CITY_RE.finditer(normalized):
prefix_words = normalized[: m.start()].split()
if prefix_words and prefix_words[-1] in _DISTRICT_PREFIXES:
continue # «мкр Заречный» — район, не город-ЗАТО Заречный
return True
return False
# ── Address normalisation ───────────────────────────────────────────────────
def normalize_address(address: str) -> str:
"""Нормализация для cache lookup: lowercase + trim + collapse whitespace.
@ -166,11 +272,35 @@ def _cache_put(db: Session, address_norm: str, result: GeocodeResult) -> None:
# ── Provider: Nominatim (OSM, без ключа) ────────────────────────────────────
async def _nominatim_query(client: httpx.AsyncClient, address: str) -> dict | None:
"""Single Nominatim search. Возвращает первый item или None.
def _nominatim_region_ok(item: dict) -> bool | None:
"""Кросс-чек региона по Nominatim `address.state` (доступно т.к. addressdetails=1).
ВАЖНО: фильтруем результаты по ЕКБ bbox прямо тут, чтобы при опечатках
не возвращать Пермский край / Челябинск.
True/False если state однозначно про/не про Свердловскую область. None если
поле отсутствует/не строка тогда accept-логика падает обратно на bbox.
Ловит Тюмень/Шадринск/Кунгур/Снежинск они внутри генерального OBLAST66_BBOX
(специально щедрого), но их state явно другой регион.
"""
addr = item.get("address")
if not isinstance(addr, dict):
return None
state = addr.get("state")
if not isinstance(state, str) or not state:
return None
return "свердловск" in state.lower()
async def _nominatim_query(client: httpx.AsyncClient, address: str) -> dict | None:
"""Single Nominatim search. Возвращает лучший item или None.
ВАЖНО: фильтруем результаты по bbox области (region 66) прямо тут, чтобы при
опечатках не возвращать Пермский край / Челябинск но не резать легитимные
Нижний Тагил / Серов и т.д. (генеральный bbox всей Свердловской области).
Two-pass tie-break: среди кандидатов предпочитаем того, кто попадает в TIGHT
ЕКБ-bbox (byte-identical для ЕКБ-запросов, даже если Nominatim ранжировал его
не первым) иначе первый кандидат внутри OBLAST66. Плюс region cross-check
(`address.state`) отсекает кандидатов ЯВНО из другого региона (Тюмень и
т.п.), даже если координаты попали в генеральный bbox.
"""
response = await client.get(
"https://nominatim.openstreetmap.org/search",
@ -180,21 +310,26 @@ async def _nominatim_query(client: httpx.AsyncClient, address: str) -> dict | No
"limit": "3",
"countrycodes": "ru",
"addressdetails": "1",
"viewbox": EKB_BBOX["viewbox"],
"bounded": "1", # строго в ЕКБ bbox
"viewbox": OBLAST66_VIEWBOX["viewbox"],
"bounded": "1", # строго в пределах области (region 66)
},
)
response.raise_for_status()
data = response.json()
oblast_fallback: dict | None = None
for item in data:
try:
lat_f = float(item["lat"])
lon_f = float(item["lon"])
if is_within_ekb_bbox(lat_f, lon_f):
return item
except Exception:
continue
return None
if _nominatim_region_ok(item) is False:
continue # регион явно не Свердловская область — не рассматриваем
if is_within_ekb_bbox(lat_f, lon_f):
return item # tight-ЕКБ приоритетнее — тот же результат, что и раньше
if oblast_fallback is None and is_within_oblast66_bbox(lat_f, lon_f):
oblast_fallback = item
return oblast_fallback
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=8))
@ -202,8 +337,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-варианты (Цвилинга Цвиллинга).
Tier 1: bounded область (region 66) на оригинальный адрес.
Tier 2: bounded область (region 66) на typo-варианты (Цвилинга Цвиллинга).
"""
headers = {
"User-Agent": f"TradeInMVP/0.1 (contact: {settings.contact_email})",
@ -237,6 +372,30 @@ async def _nominatim_lookup(address: str) -> GeocodeResult | None:
)
def _yandex_region_ok(geo_object: dict) -> bool | None:
"""Кросс-чек региона по Yandex `AdministrativeAreaName`.
True/False если Yandex вернул поле и оно однозначно про/не про Свердловскую
область. None если поля нет / оно не строка тогда accept-логика падает
обратно на bbox. Ловит Тюмень/Шадринск/Кунгур/Снежинск они внутри
генерального OBLAST66_BBOX (специально щедрого), но их регион явно другой.
"""
try:
admin_area = (
geo_object.get("metaDataProperty", {})
.get("GeocoderMetaData", {})
.get("AddressDetails", {})
.get("Country", {})
.get("AdministrativeArea", {})
.get("AdministrativeAreaName")
)
except AttributeError:
return None
if not isinstance(admin_area, str) or not admin_area:
return None
return "свердловск" in admin_area.lower()
# ── 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:
@ -247,8 +406,9 @@ async def _yandex_lookup(address: str, api_key: str) -> GeocodeResult | None:
Запрашиваем с ll+spn (центр ЕКБ) для приоритизации местных результатов,
но БЕЗ rspn чтобы fuzzy matching работал при опечатках.
"""
# Не запихиваем "Екатеринбург" если оно уже есть в адресе (типичный кейс из suggest)
geocode_query = address if "екатеринбург" in address.lower() else f"Екатеринбург, {address}"
# Не навязываем "Екатеринбург, " если в адресе уже есть город/область региона 66
# (типичный кейс из suggest, либо явный запрос по другому городу области).
geocode_query = address if _has_oblast_marker(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/",
@ -269,22 +429,44 @@ async def _yandex_lookup(address: str, api_key: str) -> GeocodeResult | None:
if not members:
return None
# Фильтруем top-5 по ЕКБ bbox — игнорируем Челябинск/Уфу/Москву при опечатке
# Фильтруем top-5 по bbox области (region 66) — игнорируем Челябинск/Уфу/Москву
# при опечатке, но принимаем легитимные Нижний Тагил/Серов/etc. Two-pass
# tie-break: tight-ЕКБ приоритетнее (byte-identical для ЕКБ-запросов, даже
# если Yandex ранжировал его не первым), иначе первый oblast-bbox кандидат.
# Region cross-check (AdministrativeAreaName) отсекает кандидатов ЯВНО из
# другого региона (Тюмень и т.п. попадают в генеральный bbox координатно).
best = None
oblast_fallback = 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)
except Exception:
continue
if _yandex_region_ok(obj) is False:
continue # регион явно не Свердловская область — не рассматриваем
if is_within_ekb_bbox(lat_f, lon_f):
best = obj
break
except Exception:
continue
if oblast_fallback is None and is_within_oblast66_bbox(lat_f, lon_f):
oblast_fallback = obj
if best is None:
# Никто из top-5 не попал в ЕКБ → берем первый «как есть» (вне ЕКБ — но хоть что-то)
best = members[0]["GeoObject"]
best = oblast_fallback
if best is None:
# Никто из top-5 не попал ни в tight, ни в oblast bbox (или все явно
# отсеяны по региону) → берём первый «как есть», НО только если его
# регион не опровергнут явно — иначе весь смысл region cross-check
# теряется на garbage-query, где top-1 — гарантированно не тот регион
# (например, Тюмень).
first_obj = members[0]["GeoObject"]
if _yandex_region_ok(first_obj) is not False:
best = first_obj
if best is None:
return None
lon_str, lat_str = best["Point"]["pos"].split()
precision_raw = (
@ -310,6 +492,8 @@ async def _yandex_lookup(address: str, api_key: str) -> GeocodeResult | None:
# ── Suggest (автокомплит) ───────────────────────────────────────────────────
# ЕКБ bounding box (приблизительно): юг 56.65, запад 60.40, север 56.95, восток 60.85
# Это ЕКБ fast-path (Yandex rspn=1 bounded tier + soft-bias ll/spn). Для областного
# охвата (region 66) — см. OBLAST66_VIEWBOX выше.
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)
@ -375,9 +559,13 @@ async def _dadata_suggest(query: str, limit: int = 8) -> list[GeocodeSuggestion]
типа город/район, для autocomplete с привязкой к карте они бесполезны).
Label собирается из DaData `value` (короткая форма «ул Малышева, д 30»).
Без city suffix query уже ограничен `locations=[{city: 'Екатеринбург'}]`.
Constraint вся область (region='Свердловская область', hard-filter внутри
`suggest_addresses`), а не один город ЕКБ иначе Нижний Тагил/Серов/etc
никогда не появились бы в подсказках.
"""
raw = await dadata.suggest_addresses(query, limit=limit)
raw = await dadata.suggest_addresses(
query, limit=limit, city=None, region=SVERDLOVSK_OBLAST_REGION
)
out: list[GeocodeSuggestion] = []
for s in raw:
if s.lat is None or s.lon is None:
@ -424,16 +612,19 @@ async def _yandex_geocode_request(
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 1: bounded ЕКБ (rspn=1) быстрый путь для основного (ЕКБ) трафика.
Tier 2: bounded ЕКБ на typo-variants (удвоение согласных).
Tier 3: без rspn fuzzy по всей стране, фильтр результатов по ЕКБ bbox.
Tier 3: без rspn fuzzy по всей стране, фильтр результатов по bbox области
(region 66) ловит легитимные Нижний Тагил/Серов/etc, которые Tier 1/2
(bounded строго ЕКБ) структурно вернуть не могут.
"""
prefixed_query = query if _has_oblast_marker(query.lower()) else f"Екатеринбург, {query}"
async with httpx.AsyncClient(timeout=8.0) as client:
# Tier 1: strict bounded на оригинал
# Tier 1: strict bounded на оригинал (ЕКБ fast path)
members = await _yandex_geocode_request(
client,
api_key,
f"Екатеринбург, {query}",
prefixed_query,
limit,
bounded=True,
)
@ -441,12 +632,15 @@ async def _yandex_suggest(query: str, api_key: str, limit: int = 8) -> list[Geoc
if results:
return results
# Tier 2: bounded на typo-варианты
# Tier 2: bounded на typo-варианты (тот же ЕКБ fast path)
for variant in _typo_variants(query, limit=4):
variant_query = (
variant if _has_oblast_marker(variant.lower()) else f"Екатеринбург, {variant}"
)
members = await _yandex_geocode_request(
client,
api_key,
f"Екатеринбург, {variant}",
variant_query,
limit,
bounded=True,
)
@ -454,21 +648,21 @@ async def _yandex_suggest(query: str, api_key: str, limit: int = 8) -> list[Geoc
if results:
return results
# Tier 3: без rspn — даём fuzzy сделать своё дело, фильтр по bbox
# Tier 3: без rspn — даём fuzzy сделать своё дело, фильтр по bbox области
members = await _yandex_geocode_request(
client,
api_key,
f"Екатеринбург, {query}",
prefixed_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
in_oblast = [r for r in results if is_within_oblast66_bbox(r.lat, r.lon)]
return in_oblast
async def _nominatim_query_multi(client: httpx.AsyncClient, query: str, limit: int) -> list[dict]:
"""Один Nominatim search с фильтром по ЕКБ bbox. Возвращает up to N items."""
"""Один Nominatim search с фильтром по bbox области (region 66). Возвращает up to N items."""
response = await client.get(
"https://nominatim.openstreetmap.org/search",
params={
@ -476,7 +670,7 @@ async def _nominatim_query_multi(client: httpx.AsyncClient, query: str, limit: i
"format": "json",
"limit": str(limit),
"countrycodes": "ru",
"viewbox": EKB_BBOX["viewbox"],
"viewbox": OBLAST66_VIEWBOX["viewbox"],
"bounded": "1",
"addressdetails": "1",
},
@ -488,21 +682,29 @@ async def _nominatim_query_multi(client: httpx.AsyncClient, query: str, limit: i
@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 недоступен)."""
"""Nominatim в режиме suggest. С typo-fallback (для случаев когда Yandex недоступен).
Суффикс ", Екатеринбург" навязывается ТОЛЬКО если в запросе ещё нет города/области
региона 66 иначе не режем явные запросы по другим городам области.
"""
headers = {
"User-Agent": f"TradeInMVP/0.1 (contact: {settings.contact_email})",
"Accept": "application/json",
"Accept-Language": "ru,en;q=0.8",
}
suffixed_query = query if _has_oblast_marker(query.lower()) else f"{query}, Екатеринбург"
async with httpx.AsyncClient(timeout=8.0, headers=headers) as client:
# Tier 1: оригинальный query
data = await _nominatim_query_multi(client, f"{query}, Екатеринбург", limit)
data = await _nominatim_query_multi(client, suffixed_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)
variant_query = (
variant if _has_oblast_marker(variant.lower()) else f"{variant}, Екатеринбург"
)
data = await _nominatim_query_multi(client, variant_query, limit)
if data:
logger.info("nominatim suggest typo-fixed: %s%s", query, variant)
break
@ -832,7 +1034,8 @@ def _cadastral_reverse_sync(db: Session, lat: float, lon: float, radius_m: int =
async def suggest(query: str, db: Session | None = None, limit: int = 8) -> list[GeocodeSuggestion]:
"""Автокомплит адресов в ЕКБ. Cadastral FDW → DaData → Yandex → Nominatim → [].
"""Автокомплит адресов в Свердловской области (region 66; ЕКБ — основной трафик,
остаётся быстрым fast-path). Cadastral FDW DaData Yandex Nominatim [].
db: если передан cadastral lookup через gendesign_cad_buildings (первый tier).
DaData /suggest (PR Q2) token-only, 10k/день, заменяет Yandex который
@ -978,11 +1181,9 @@ async def geocode(address: str, db: Session) -> GeocodeResult | None:
if settings.yandex_geocoder_api_key:
try:
result = await _yandex_lookup(address, settings.yandex_geocoder_api_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:
# Если результат вне области (region 66) — пробуем typo-варианты
in_oblast = result is not None and is_within_oblast66_bbox(result.lat, result.lon)
if result is not None and in_oblast:
await asyncio.to_thread(_cache_put, db, addr_norm, result)
logger.info("geocode yandex: %s → (%.5f, %.5f)", addr_norm, result.lat, result.lon)
return result
@ -994,7 +1195,7 @@ async def geocode(address: str, db: Session) -> GeocodeResult | None:
continue
if result is None:
continue
if 60.40 <= result.lon <= 60.85 and 56.65 <= result.lat <= 56.95:
if is_within_oblast66_bbox(result.lat, result.lon):
await asyncio.to_thread(_cache_put, db, addr_norm, result)
logger.info(
"geocode yandex typo-fixed: %s%s → (%.5f, %.5f)",
@ -1128,11 +1329,12 @@ async def _yandex_reverse(lat: float, lon: float, api_key: str) -> ReverseGeocod
for p in parts
if p not in {"Россия", "Свердловская область"} and not p.startswith("городской округ")
]
# Найдём locality (Екатеринбург / Берёзовский / …) и переставим в конец
# Найдём locality (Екатеринбург / Нижний Тагил / Берёзовский / …) и переставим
# в конец — любой город региона 66, не только ЕКБ (case-insensitive match).
locality = None
rest: list[str] = []
for p in filtered:
if locality is None and p in {"Екатеринбург"}:
if locality is None and p.lower() in SVERDLOVSK_OBLAST_CITIES:
locality = p
else:
rest.append(p)

View file

@ -652,6 +652,55 @@ async def test_suggest_addresses_sends_proper_request() -> None:
assert captured["x_secret"] is None
async def test_suggest_addresses_region_body_shape_no_restrict_value() -> None:
"""region= constrains `locations` by region, WITHOUT a restrict_value key.
`restrict_value` is a top-level DaData body param (not a per-location dict
key) putting it inside the locations object would be a silent no-op.
The region constraint in `locations` is already a hard filter on its own.
"""
from app.services import dadata
captured: dict[str, object] = {}
def handler(request: httpx.Request) -> httpx.Response:
import json as _json
captured["body"] = _json.loads(request.content.decode("utf-8"))
return httpx.Response(200, json=SAMPLE_SUGGEST_PAYLOAD)
transport = httpx.MockTransport(handler)
with _patch_settings(token="my-token"), _patch_async_client(transport):
await dadata.suggest_addresses("Ленина", limit=5, city=None, region="Свердловская область")
body = captured["body"]
assert isinstance(body, dict)
assert body["locations"] == [{"region": "Свердловская область"}]
assert "restrict_value" not in body["locations"][0]
assert "restrict_value" not in body
async def test_suggest_addresses_region_overrides_city() -> None:
"""Когда задан и `city`, и `region` — region побеждает, city игнорируется."""
from app.services import dadata
captured: dict[str, object] = {}
def handler(request: httpx.Request) -> httpx.Response:
import json as _json
captured["body"] = _json.loads(request.content.decode("utf-8"))
return httpx.Response(200, json=SAMPLE_SUGGEST_PAYLOAD)
transport = httpx.MockTransport(handler)
with _patch_settings(), _patch_async_client(transport):
await dadata.suggest_addresses("Ленина", city="Екатеринбург", region="Свердловская область")
body = captured["body"]
assert isinstance(body, dict)
assert body["locations"] == [{"region": "Свердловская область"}]
async def test_suggest_addresses_clamps_limit() -> None:
"""count должен быть clamped в [1, 20] — DaData ограничение."""
from app.services import dadata

View file

@ -1,18 +1,32 @@
"""Тесты для ЕКБ bbox-хелперов geocoder (#1871).
"""Тесты для ЕКБ/oblast66 bbox-хелперов geocoder (#1871, oblast-66 expansion).
Покрывают:
- is_within_ekb_bbox_wide ingest-guard (avito_detail координаты);
- is_within_ekb_bbox geocoder tight-фильтр (поведение НЕ изменилось после DRY-рефактора);
- инвариант WIDE TIGHT.
- инвариант WIDE TIGHT;
- OBLAST66_BBOX / is_within_oblast66_bbox генеральный bbox всей области (region 66),
инвариант OBLAST66 TIGHT, дальние города области, известное ограничение (Тюмень
внутри bbox координатно компенсируется region cross-check на accept-сайтах);
- _has_oblast_marker word-boundary matching (не substring);
- _nominatim_query / _yandex_lookup accept-сайты two-pass tie-break (tight ЕКБ
приоритетнее) + region cross-check (отсекает Тюмень и т.п. даже внутри bbox).
"""
from unittest.mock import patch
import httpx
import pytest
from app.services.geocoder import (
EKB_BBOX_TIGHT,
EKB_BBOX_WIDE,
OBLAST66_BBOX,
_has_oblast_marker,
_nominatim_query,
_yandex_lookup,
is_within_ekb_bbox,
is_within_ekb_bbox_wide,
is_within_oblast66_bbox,
)
# Центр ЕКБ (Плотинка) — внутри обоих bbox.
@ -86,3 +100,198 @@ def test_wide_strictly_contains_tight() -> None:
for lon in (t_lon_min, t_lon_max):
assert is_within_ekb_bbox(lat, lon) is True
assert is_within_ekb_bbox_wide(lat, lon) is True
# ── OBLAST66_BBOX (region 66 expansion) ─────────────────────────────────────
def test_oblast66_strictly_contains_tight() -> None:
"""Инвариант: OBLAST66_BBOX ⊇ EKB_BBOX_TIGHT — генерализация не режет ЕКБ.
Мирроор test_wide_strictly_contains_tight, но для областного bbox.
"""
t_lat_min, t_lat_max, t_lon_min, t_lon_max = EKB_BBOX_TIGHT
o_lat_min, o_lat_max, o_lon_min, o_lon_max = OBLAST66_BBOX
assert o_lat_min <= t_lat_min
assert o_lat_max >= t_lat_max
assert o_lon_min <= t_lon_min
assert o_lon_max >= t_lon_max
for lat in (t_lat_min, t_lat_max):
for lon in (t_lon_min, t_lon_max):
assert is_within_ekb_bbox(lat, lon) is True
assert is_within_oblast66_bbox(lat, lon) is True
@pytest.mark.parametrize(
"lat,lon,label",
[
(60.6935, 60.4235, "Ивдель"),
(59.6297, 60.5541, "Серов"),
(58.0378, 65.2753, "Тавда"),
(56.6152, 57.7666, "Красноуфимск"),
],
)
def test_oblast66_accepts_far_towns(lat: float, lon: float, label: str) -> None:
"""Дальние города области (вне EKB tight/wide) — приняты oblast66-bbox."""
assert is_within_oblast66_bbox(lat, lon) is True, f"{label} должен быть внутри OBLAST66_BBOX"
def test_oblast66_bbox_admits_tyumen_by_design() -> None:
"""OBLAST66_BBOX — генеральный (щедрый) bbox, координатно admits Тюмень.
Это ЗНАЕМОЕ и намеренное ограничение самого bbox-хелпера (дешевле
false-positive у границы, чем false-negative на корректном адресе области).
Реальное отсечение Тюмени происходит на accept-сайтах через region
cross-check см. test_nominatim_query_rejects_out_of_region_only_candidate /
test_yandex_lookup_rejects_out_of_region_only_candidate ниже.
"""
assert is_within_oblast66_bbox(57.1530, 65.5343) is True
# ── _has_oblast_marker — word/phrase-boundary matching (не substring) ───────
@pytest.mark.parametrize(
"text,expected",
[
("екатеринбург, малышева 30", True),
("нижний тагил, ленина 10", True),
("свердловская область, серова 27", True),
("каменск-уральский, мира 1", True),
("заречный, мира 5", True), # ЗАТО-город (без district-префикса)
# false-positives, которые substring-match ловил ошибочно (#oblast66 bug):
("серова 27", False), # НЕ город "серов"
("ирбитская 5", False), # НЕ город "ирбит"
("асбестовский пер., 3", False), # НЕ город "асбест"
("невьянский пер., 10", False), # НЕ город "невьянск"
("богдановича, 22", False), # НЕ город "богданович"
("мкр заречный, мира 5", False), # район внутри города, не город-ЗАТО
("микрорайон заречный, мира 5", False),
("малышева 30", False), # обычный ЕКБ bare-адрес
],
)
def test_has_oblast_marker(text: str, expected: bool) -> None:
assert _has_oblast_marker(text) is expected
# ── Accept-site two-pass tie-break + region cross-check ─────────────────────
async def test_nominatim_query_prefers_tight_ekb_over_oblast_rank0() -> None:
"""rank0 — Серов (внутри OBLAST66, вне tight), rank1 — ЕКБ (tight).
Two-pass tie-break должен выбрать rank1 (tight), а не rank0, несмотря на то
что Nominatim ранжировал его вторым.
"""
payload = [
{"lat": "59.6297", "lon": "60.5541", "address": {"state": "Свердловская область"}},
{"lat": "56.838", "lon": "60.605", "address": {"state": "Свердловская область"}},
]
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json=payload)
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport) as client:
item = await _nominatim_query(client, "Серова 27")
assert item is not None
assert float(item["lat"]) == pytest.approx(56.838)
assert float(item["lon"]) == pytest.approx(60.605)
async def test_nominatim_query_rejects_out_of_region_only_candidate() -> None:
"""Единственный кандидат — Тюмень (координатно внутри OBLAST66, но регион другой)
region cross-check отсекает его None (не просто "первый как есть")."""
payload = [{"lat": "57.1522", "lon": "65.5272", "address": {"state": "Тюменская область"}}]
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json=payload)
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport) as client:
item = await _nominatim_query(client, "какой-то мусорный адрес")
assert item is None
async def test_nominatim_query_accepts_oblast_wide_when_region_absent() -> None:
"""Нет tight-совпадения, `address.state` отсутствует (Nominatim не всегда его
отдаёт) fallback на oblast-bbox (Серов), region cross-check не блокирует."""
payload = [{"lat": "59.6297", "lon": "60.5541"}]
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json=payload)
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport) as client:
item = await _nominatim_query(client, "Серов, Ленина 1")
assert item is not None
assert float(item["lat"]) == pytest.approx(59.6297)
_REAL_ASYNC_CLIENT = httpx.AsyncClient
def _yandex_client_factory(transport: httpx.MockTransport):
"""Drop-in replacement для httpx.AsyncClient внутри geocoder._yandex_lookup."""
def factory(*_: object, **__: object) -> httpx.AsyncClient:
return _REAL_ASYNC_CLIENT(transport=transport)
return factory
def _yandex_geo_object(lat: str, lon: str, admin_area: str | None) -> dict:
meta: dict = {"precision": "exact", "text": "тест"}
if admin_area is not None:
meta["AddressDetails"] = {
"Country": {"AdministrativeArea": {"AdministrativeAreaName": admin_area}}
}
return {
"GeoObject": {
"Point": {"pos": f"{lon} {lat}"},
"metaDataProperty": {"GeocoderMetaData": meta},
}
}
def _yandex_payload(members: list[dict]) -> dict:
return {"response": {"GeoObjectCollection": {"featureMember": members}}}
async def test_yandex_lookup_prefers_tight_ekb_over_oblast_rank0() -> None:
"""Тот же tie-break сценарий, что и для Nominatim, но для Yandex top-5."""
members = [
_yandex_geo_object("59.6297", "60.5541", "Свердловская область"), # Серов, rank0
_yandex_geo_object("56.838", "60.605", "Свердловская область"), # ЕКБ tight, rank1
]
payload = _yandex_payload(members)
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json=payload)
transport = httpx.MockTransport(handler)
with patch("app.services.geocoder.httpx.AsyncClient", _yandex_client_factory(transport)):
result = await _yandex_lookup("Серова 27", "fake-key")
assert result is not None
assert result.lat == pytest.approx(56.838)
assert result.lon == pytest.approx(60.605)
async def test_yandex_lookup_rejects_out_of_region_only_candidate() -> None:
"""Единственный кандидат — Тюмень (bbox admits координатно, регион другой) →
region cross-check отсекает даже в ultimate as-is fallback None."""
members = [_yandex_geo_object("57.1522", "65.5272", "Тюменская область")]
payload = _yandex_payload(members)
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json=payload)
transport = httpx.MockTransport(handler)
with patch("app.services.geocoder.httpx.AsyncClient", _yandex_client_factory(transport)):
result = await _yandex_lookup("какой-то мусорный адрес", "fake-key")
assert result is None