"""Geocode endpoints — debug + frontend Suggest proxy.""" from __future__ import annotations import asyncio import logging from typing import Annotated from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel, Field from sqlalchemy import text from sqlalchemy.orm import Session from app.core.db import get_db from app.services.estimator import _lookup_house_facts from app.services.geocoder import GeocodeResult, geocode, reverse_geocode, suggest logger = logging.getLogger(__name__) router = APIRouter() @router.get("/lookup", response_model=GeocodeResult) async def lookup( address: Annotated[str, Query(min_length=3, max_length=500)], db: Annotated[Session, Depends(get_db)], city_hint: Annotated[ str | None, Query( max_length=100, description=( "Город, если известен вызывающему (например выбран пользователем " "на предыдущем шаге UI). #2576: без него геокодер БОЛЬШЕ НЕ " "подставляет 'Екатеринбург' молча — ответ может помечаться " "city_ambiguous=true." ), ), ] = None, ) -> GeocodeResult: """Геокодинг адреса → lat/lon. Примеры: /api/v1/geocode/lookup?address=ул.+Малышева+30+Екатеринбург /api/v1/geocode/lookup?address=Куйбышева+50+Екатеринбург /api/v1/geocode/lookup?address=Ленина+1&city_hint=Нижний+Тагил """ result = await geocode(address, db, city_hint=city_hint) if result is None: raise HTTPException(status_code=404, detail=f"Address not found: {address}") return result class SuggestItem(BaseModel): label: str full_address: str lat: float lon: float kind: str # ГАР OBJECTGUID (ФИАС) дома — присутствует ТОЛЬКО у house-level кандидатов # DaData-тира. У прочих тиров / street/locality-кандидатов null. Фронт # прокидывает его в POST /estimate как target_fias_id (Tier 0.5 fias_exact). fias_id: str | None = None class SuggestResponse(BaseModel): items: list[SuggestItem] @router.get("/suggest", response_model=SuggestResponse) async def suggest_addresses( q: Annotated[str, Query(min_length=2, max_length=200, description="Запрос для автокомплита")], limit: Annotated[int, Query(ge=1, le=15)] = 8, db: Annotated[Session, Depends(get_db)] = None, # type: ignore[assignment] city_hint: Annotated[ str | None, Query( max_length=100, description=( "Город, если известен вызывающему (#2576) — без него подсказки " "БОЛЬШЕ НЕ ограничиваются молчаливо Екатеринбургом." ), ), ] = None, ) -> SuggestResponse: """Автокомплит адресов в Свердловской области (region 66; ЕКБ — основной трафик, остаётся быстрым fast-path). Используется в EstimateForm для подсказок пока пользователь печатает. Bounded viewbox — генеральный по всей области (см. geocoder.OBLAST66_VIEWBOX), ЕКБ (lon 60.40-60.85, lat 56.65-56.95) внутри него остаётся быстрым fast-path. Пример: /api/v1/geocode/suggest?q=Малышева /api/v1/geocode/suggest?q=Цвиллинга # → пусто, такой улицы в ЕКБ нет /api/v1/geocode/suggest?q=Ленина+1&city_hint=Нижний+Тагил """ items = await suggest(q, db=db, limit=limit, city_hint=city_hint) return SuggestResponse( items=[ SuggestItem( label=s.label, full_address=s.full_address, lat=s.lat, lon=s.lon, kind=s.kind, fias_id=s.fias_id, ) for s in items ] ) class ReverseResponse(BaseModel): """Reverse-geocode response для MapPicker'а. `lat`/`lon` — echo входной точки клика (для логики «отодвинули ли далеко»). `snapped_lat`/`snapped_lon` — центр matched здания от provider'а. Если `precision in ("exact","number")` фронт двигает marker на snapped point — пользователь видит «магнит к дому». Для остальных precision snapped == input. """ address: str lat: float = Field(..., description="Исходная latitude клика") lon: float = Field(..., description="Исходная longitude клика") snapped_lat: float = Field(..., description="Latitude центра matched здания") snapped_lon: float = Field(..., description="Longitude центра matched здания") precision: str = Field( ..., description=( "exact/number/street/range/near/locality/other/cadastral. " "Фронт двигает marker только если exact/number/cadastral." ), ) provider: str = Field(..., description="cadastral | nominatim") @router.get("/reverse", response_model=ReverseResponse) async def reverse( lat: Annotated[float, Query(ge=-90, le=90)], lon: Annotated[float, Query(ge=-180, le=180)], db: Annotated[Session, Depends(get_db)] = None, # type: ignore[assignment] ) -> ReverseResponse: """Обратный геокодинг — координаты с карты → адрес + snapped точка здания. Пример: /api/v1/geocode/reverse?lat=56.8389&lon=60.6057 """ result = await reverse_geocode(lat, lon, db=db) if result is None: raise HTTPException(status_code=404, detail="address not found for coordinates") return ReverseResponse( address=result.address, lat=lat, lon=lon, snapped_lat=result.snapped_lat, snapped_lon=result.snapped_lon, precision=result.precision, provider=result.provider, ) class HouseFactsResponse(BaseModel): """Ответ /house-facts.""" found: bool total_floors: int | None = None year_built: int | None = None house_type: str | None = None source: str | None = Field( default=None, description="'houses' когда факты найдены в справочнике, иначе None" ) def _resolve_house_id_by_fias(db: Session, fias_id: str) -> int | None: """Ищет id дома в `houses` по ФИАС/ГАР guid — по трём алиасам guid-полей справочника (заполняются из разных источников загрузки: DOM.РФ/ГИС ЖКХ/ГАР). Best-effort: не нашли — вызывающий молча падает на geo-фолбэк по lat/lon. """ try: row = ( db.execute( text( """ SELECT id FROM houses WHERE gar_house_guid = CAST(:g AS text) OR house_fias_id = CAST(:g AS text) OR zhkh_house_guid = CAST(:g AS text) LIMIT 1 """ ), {"g": fias_id}, ) .mappings() .first() ) except Exception as exc: # pragma: no cover — defensive # Тот же best-effort, что и у _lookup_house_facts ниже по цепочке: # предзаполнение формы НЕ должно ронять запрос в 500. Откатываем # транзакцию, чтобы не отравить сессию для последующего geo-фолбэка. logger.warning("house-facts: fias lookup failed (graceful): %s", exc) try: db.rollback() except Exception: # pragma: no cover — defensive pass return None if row is None: return None house_id = row["id"] return house_id if isinstance(house_id, int) else None @router.get("/house-facts", response_model=HouseFactsResponse) async def house_facts( lat: Annotated[float, Query(ge=-90, le=90)], lon: Annotated[float, Query(ge=-180, le=180)], db: Annotated[Session, Depends(get_db)], fias_id: Annotated[str | None, Query(max_length=64)] = None, ) -> HouseFactsResponse: """Предзаполнение формы оценки (этажность/год/тип дома) из справочника `houses`. Покрытие справочника: total_floors — 98.3%, year_built — 86.8% (прод, 2026-08). Переиспользует ту же логику поиска, что и estimate_quality() (#3234): по target_house_id если резолвился ФИАС, иначе ближайший дом в радиусе 60м от lat/lon (ST_DWithin). Отдельная ручка, а не поле в /suggest: suggest дёргается на КАЖДОЕ нажатие клавиши автокомплита (debounced, но всё равно несколько запросов на ввод адреса) — тащить туда ещё один запрос к houses на каждый из 8 кандидатов на каждый keystroke было бы неоправданной нагрузкой на БД. /house-facts вызывается ОДИН раз, после того как пользователь выбрал конкретный адрес из подсказок. found=false (200, НЕ 404) означает «дома нет в справочнике» — это не ошибка запроса, а честный ответ об отсутствии данных. Пример: /api/v1/geocode/house-facts?lat=56.838&lon=60.595 /api/v1/geocode/house-facts?lat=56.838&lon=60.595&fias_id=... """ target_house_id: int | None = None if fias_id is not None: target_house_id = await asyncio.to_thread(_resolve_house_id_by_fias, db, fias_id) facts = await asyncio.to_thread( _lookup_house_facts, db, target_house_id=target_house_id, lat=lat, lon=lon, ) if facts is None: return HouseFactsResponse(found=False) return HouseFactsResponse( found=True, total_floors=facts.total_floors, year_built=facts.year_built, house_type=facts.house_type, source="houses", )