"""Nominatim-backed online geocoder. The actual HTTP call is encapsulated in :func:`build_default_nominatim_geocoder` which returns an ``lru_cache``-wrapped function. Domain modules typically: 1. Build a default geocoder via :func:`build_default_nominatim_geocoder`. 2. Re-export it under a stable module-level name (e.g. ``_geocode_online``). 3. Pass a *late-binding lambda* (``lambda q: _geocode_online(q)``) to :class:`NominatimResolver`. This ensures tests that ``monkeypatch.setattr(module, "_geocode_online", ...)`` can swap the geocoder behavior without touching pipeline construction. """ from __future__ import annotations import time from functools import lru_cache from typing import Any, Callable import httpx from ..models import LocationCandidate, LocationQuery, ResolverOutput from ..text import ( coerce_str, normalize_country_text, normalize_text, parse_float, ) NOMINATIM_SEARCH_URL = "https://nominatim.openstreetmap.org/search" DEFAULT_USER_AGENT = "planet-earth-location-resolver/1.0" DEFAULT_MIN_INTERVAL_SECONDS = 1.1 DEFAULT_TIMEOUT_SECONDS = 8.0 def build_default_nominatim_geocoder( *, user_agent: str = DEFAULT_USER_AGENT, min_interval_seconds: float = DEFAULT_MIN_INTERVAL_SECONDS, timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, cache_size: int = 512, ) -> Callable[[str], dict[str, Any] | None]: """Return a cached, rate-limited Nominatim geocoder.""" last_request_at = [0.0] @lru_cache(maxsize=cache_size) def geocode(query: str) -> dict[str, Any] | None: if not query: return None elapsed = time.monotonic() - last_request_at[0] if elapsed < min_interval_seconds: time.sleep(min_interval_seconds - elapsed) last_request_at[0] = time.monotonic() response = httpx.get( NOMINATIM_SEARCH_URL, params={ "q": query, "format": "jsonv2", "limit": 1, "addressdetails": 1, }, headers={"User-Agent": user_agent}, timeout=timeout_seconds, ) response.raise_for_status() payload = response.json() if not isinstance(payload, list) or not payload: return None result = payload[0] if not isinstance(result, dict): return None return result return geocode _DEFAULT_SITE_CATEGORIES = frozenset( { "amenity", "office", "building", "industrial", "research", "university", "education", "tourism", "shop", "man_made", "campus", "research_institute", } ) def interpret_geocode_result( result: dict[str, Any], *, matched_fields: tuple[str, ...], context_country: str | None, site_categories: frozenset[str] = _DEFAULT_SITE_CATEGORIES, site_promoting_match_fields: frozenset[str] = frozenset( {"site", "operator", "name"} ), ) -> tuple[float, float, dict[str, Any], str] | None: """Validate a Nominatim raw result. Returns (lat, lon, address, classification).""" latitude = parse_float(result.get("lat")) longitude = parse_float(result.get("lon")) if latitude in (None, 0.0) or longitude in (None, 0.0): return None address = result.get("address") if isinstance(result.get("address"), dict) else {} if not isinstance(address, dict): address = {} has_city_level = bool( address.get("city") or address.get("town") or address.get("village") or address.get("municipality") or address.get("hamlet") or address.get("suburb") ) osm_class = str(result.get("class") or "").lower() osm_type = str(result.get("type") or "").lower() is_site_like = osm_class in site_categories or osm_type in site_categories if not has_city_level and not is_site_like: return None if context_country: normalized_context = normalize_text(normalize_country_text(context_country)) normalized_result = normalize_text( normalize_country_text(address.get("country")) ) if ( normalized_context and normalized_result and normalized_context != normalized_result ): return None classification = ( "site" if ( is_site_like and has_city_level and any(field in site_promoting_match_fields for field in matched_fields) ) else "city" ) return float(latitude), float(longitude), address, classification def _candidate_from_geocode( *, query: LocationQuery, geocode_query: str, matched_fields: tuple[str, ...], raw_result: dict[str, Any], interpret: Callable[..., tuple[float, float, dict[str, Any], str] | None], source: str, site_confidence: float, city_confidence: float, ) -> LocationCandidate | None: interpreted = interpret( raw_result, matched_fields=matched_fields, context_country=query.country, ) if not interpreted: return None latitude, longitude, address, classification = interpreted city = ( address.get("city") or address.get("town") or address.get("village") or address.get("municipality") or query.city or None ) region = address.get("state") or address.get("region") country = address.get("country") or query.country or None display_name = raw_result.get("display_name") or geocode_query confidence = city_confidence if classification == "city" else site_confidence extra = query.extra or {} suggested_registry_entry = { "canonical_name": ( (query.aliases[0] if query.aliases else None) or query.name or display_name ), "aliases": list( { value for value in [ query.name, *query.aliases, coerce_str(extra.get("operator")), coerce_str(extra.get("site")), ] if value } ), "operator": coerce_str(extra.get("operator")) or None, "site": coerce_str(extra.get("site")) or coerce_str(extra.get("organization")) or None, "country": country, "city": city, "region": region, "latitude": latitude, "longitude": longitude, "precision": classification, "confidence": confidence, "source_note": ( f"Resolved via Nominatim query '{geocode_query}' → {display_name}" ), } return LocationCandidate( latitude=latitude, longitude=longitude, display_name=display_name, precision=classification, confidence=confidence, query=geocode_query, source=source, source_note=f"Nominatim search result: {display_name}", matched_fields=matched_fields, needs_confirmation=True, city=city, region=region, country=country, matched_location_name=display_name, location_verified_at=None, suggested_registry_entry=suggested_registry_entry, ) class NominatimResolver: """Run a domain-specific query plan against Nominatim.""" def __init__( self, *, query_plan_builder: Callable[ [LocationQuery], list[tuple[str, tuple[str, ...]]] ], geocoder: Callable[[str], dict[str, Any] | None], name: str = "nominatim_online_geocode", site_confidence: float = 0.72, city_confidence: float = 0.62, interpret: Callable[..., tuple[float, float, dict[str, Any], str] | None] = ( interpret_geocode_result ), ) -> None: self.name = name self._query_plan_builder = query_plan_builder self._geocoder = geocoder self._site_confidence = site_confidence self._city_confidence = city_confidence self._interpret = interpret def resolve(self, query: LocationQuery) -> ResolverOutput: plan = self._query_plan_builder(query) candidates: list[LocationCandidate] = [] attempted: list[str] = [] for geocode_query, matched_fields in plan: attempted.append(geocode_query) try: raw_result = self._geocoder(geocode_query) except Exception: continue if not raw_result: continue candidate = _candidate_from_geocode( query=query, geocode_query=geocode_query, matched_fields=matched_fields, raw_result=raw_result, interpret=self._interpret, source=self.name, site_confidence=self._site_confidence, city_confidence=self._city_confidence, ) if candidate is not None: candidates.append(candidate) return ResolverOutput( candidates=tuple(candidates), attempted_queries=tuple(attempted), )