"""Nominatim client — server-side only; browsers never call Nominatim directly.""" from __future__ import annotations import logging import re from typing import Any import requests from django.conf import settings logger = logging.getLogger(__name__) # ISO3166-2-lvl4 "US-OH" → "OH"; fall back to common full-name map. _US_STATE_ABBREV = { "alabama": "AL", "alaska": "AK", "arizona": "AZ", "arkansas": "AR", "california": "CA", "colorado": "CO", "connecticut": "CT", "delaware": "DE", "district of columbia": "DC", "florida": "FL", "georgia": "GA", "hawaii": "HI", "idaho": "ID", "illinois": "IL", "indiana": "IN", "iowa": "IA", "kansas": "KS", "kentucky": "KY", "louisiana": "LA", "maine": "ME", "maryland": "MD", "massachusetts": "MA", "michigan": "MI", "minnesota": "MN", "mississippi": "MS", "missouri": "MO", "montana": "MT", "nebraska": "NE", "nevada": "NV", "new hampshire": "NH", "new jersey": "NJ", "new mexico": "NM", "new york": "NY", "north carolina": "NC", "north dakota": "ND", "ohio": "OH", "oklahoma": "OK", "oregon": "OR", "pennsylvania": "PA", "rhode island": "RI", "south carolina": "SC", "south dakota": "SD", "tennessee": "TN", "texas": "TX", "utah": "UT", "vermont": "VT", "virginia": "VA", "washington": "WA", "west virginia": "WV", "wisconsin": "WI", "wyoming": "WY", } class NominatimError(RuntimeError): pass # Leading house / unit number from user query (e.g. "1968", "12A", "100-102"). _HOUSE_FROM_QUERY = re.compile(r"^(\d+[A-Za-z]?(?:-\d+[A-Za-z]?)?)\b") def _house_from_query(query: str) -> str: match = _HOUSE_FROM_QUERY.match((query or "").strip()) return match.group(1) if match else "" def _state_code(addr: dict[str, Any]) -> str: iso = (addr.get("ISO3166-2-lvl4") or "").strip() if iso.startswith("US-") and len(iso) == 5: return iso[3:] raw = (addr.get("state") or "").strip() if len(raw) == 2: return raw.upper() return _US_STATE_ABBREV.get(raw.lower(), raw) def _city(addr: dict[str, Any]) -> str: for key in ("city", "town", "village", "hamlet", "municipality", "suburb"): val = (addr.get(key) or "").strip() if val: return val return "" def _line1(addr: dict[str, Any], display_name: str, *, query: str = "") -> str: house = (addr.get("house_number") or "").strip() road = (addr.get("road") or addr.get("pedestrian") or "").strip() # Nominatim often returns road-level hits with no house_number even when the # user typed one — keep that number so mailing street isn't incomplete. if not house: house = _house_from_query(query) if house and road: return f"{house} {road}" if road: return road # Place-level hits (city only) — leave street empty for the user to fill. if house or road: return " ".join(p for p in (house, road) if p) first = (display_name or "").split(",")[0].strip() # Avoid stuffing "Akron" into street when it's a city result. if first and first.lower() != _city(addr).lower(): return first return "" def normalize_hit(raw: dict[str, Any], *, query: str = "") -> dict[str, str]: addr = raw.get("address") or {} if not isinstance(addr, dict): addr = {} country_code = (addr.get("country_code") or "us").upper() if country_code == "US": country = "US" else: country = country_code[:2] or "US" display = (raw.get("display_name") or "").strip() line1 = _line1(addr, display, query=query) label = display # Surface recovered house number in the dropdown when OSM omitted it. house = (addr.get("house_number") or "").strip() or _house_from_query(query) if house and label and not re.match(rf"^{re.escape(house)}\b", label, re.I): label = f"{house} {label}" return { "label": label, "line1": line1, "line2": "", "city": _city(addr), "state": _state_code(addr), "zip": (addr.get("postcode") or "").strip().split(";")[0].strip(), "country": country, } def suggest_addresses(query: str, *, limit: int = 5) -> list[dict[str, str]]: """ Proxy Nominatim /search. Returns normalized address dicts for the UI. Nominatim itself has no API-key auth — LAN firewall + this Django proxy gate access. Optional NOMINATIM_API_KEY is sent as X-API-Key if you put a gateway in front of Nominatim later. """ base = (settings.NOMINATIM_BASE_URL or "").rstrip("/") if not base: raise NominatimError("NOMINATIM_BASE_URL is not configured") q = (query or "").strip() if len(q) < 3: return [] limit = max(1, min(int(limit or 5), 8)) params: dict[str, str | int] = { "q": q, "format": "json", "addressdetails": 1, "limit": limit, } countrycodes = (settings.NOMINATIM_COUNTRY_CODES or "").strip() if countrycodes: params["countrycodes"] = countrycodes headers = { "User-Agent": settings.NOMINATIM_USER_AGENT, "Accept": "application/json", } api_key = (settings.NOMINATIM_API_KEY or "").strip() if api_key: headers["X-API-Key"] = api_key url = f"{base}/search" try: response = requests.get( url, params=params, headers=headers, timeout=settings.NOMINATIM_TIMEOUT_SECONDS, ) response.raise_for_status() payload = response.json() except requests.RequestException as exc: logger.exception("Nominatim request failed") raise NominatimError(f"Nominatim unreachable at {url}: {exc}") from exc except ValueError as exc: raise NominatimError("Nominatim returned invalid JSON") from exc if not isinstance(payload, list): return [] results: list[dict[str, str]] = [] seen: set[str] = set() for item in payload: if not isinstance(item, dict): continue normalized = normalize_hit(item, query=q) key = re.sub(r"\s+", " ", normalized["label"].lower()) if not key or key in seen: continue seen.add(key) results.append(normalized) return results