"""Nearby-place search with iterative radius expansion. Sorts candidate places by distance ascending and grows the search radius by ``step_m`` until at least ``min_results`` places are found (or ``max_radius_m`` is hit, as a safety cap against runaway loops in sparsely-covered areas). Reserved for a future v2 extension: ``speed``/``heading`` inputs to bias results toward the user's direction of travel and to pick a shorter content ``length`` automatically when moving fast through a dense cluster of places. Not implemented in this iteration — callers should not pass them yet. """ import logging from dataclasses import dataclass from geoalchemy2 import Geometry from sqlalchemy import Row, and_, cast, func, select from sqlalchemy.orm import Session from app.models.city import City from app.models.enums import ContentLanguage, ContentLength from app.models.place import Place from app.models.place_content import PlaceContent logger = logging.getLogger("guidecity.nearby_search") @dataclass class NearbyResult: search_radius_m: int rows: list[Row] def find_nearby( db: Session, *, lat: float, lon: float, language: ContentLanguage, length: ContentLength, city_slug: str | None = None, min_results: int = 5, initial_radius_m: int = 300, step_m: int = 200, max_radius_m: int = 5000, ) -> NearbyResult: city_id: int | None = None if city_slug is not None: city_id = db.scalar(select(City.id).where(City.slug == city_slug)) if city_id is None: logger.warning("city_slug=%r not found; searching across all cities", city_slug) logger.debug( "nearby search starting: lat=%.5f lon=%.5f city_slug=%s lang=%s length=%s " "min_results=%d initial_radius_m=%d step_m=%d max_radius_m=%d", lat, lon, city_slug, language, length, min_results, initial_radius_m, step_m, max_radius_m, ) origin = func.ST_GeogFromText(f"SRID=4326;POINT({lon} {lat})") distance_expr = func.ST_Distance(Place.location, origin) base_stmt = ( select( Place, PlaceContent, distance_expr.label("distance_m"), func.ST_X(cast(Place.location, Geometry)).label("lon"), func.ST_Y(cast(Place.location, Geometry)).label("lat"), ) .join( PlaceContent, and_( PlaceContent.place_id == Place.id, PlaceContent.language == language, PlaceContent.length == length, ), ) .where(Place.is_active.is_(True)) .order_by(distance_expr) ) if city_id is not None: base_stmt = base_stmt.where(Place.city_id == city_id) radius = initial_radius_m rows: list[Row] = [] while True: stmt = base_stmt.where(func.ST_DWithin(Place.location, origin, radius)) rows = db.execute(stmt).all() logger.debug("radius=%dm -> %d place(s) found", radius, len(rows)) if len(rows) >= min_results or radius >= max_radius_m: break radius += step_m if len(rows) < min_results: logger.warning( "hit max_radius_m=%dm with only %d/%d place(s) found near lat=%.5f lon=%.5f", max_radius_m, len(rows), min_results, lat, lon, ) logger.info("nearby search done: radius=%dm count=%d", radius, len(rows)) return NearbyResult(search_radius_m=radius, rows=rows)