guideCity/backend/app/services/nearby_search.py

"""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)