web_django_template/site/contacts/nominatim.py

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