Batch Geocoding with Nominatim Without Getting Blocked Jump to heading#

from __future__ import annotations

import logging
import re
import sqlite3
import time
import unicodedata
from dataclasses import dataclass
from pathlib import Path

import requests

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logger = logging.getLogger("osm.nominatim.batch")

ENDPOINT = "https://nominatim.openstreetmap.org/search"
HEADERS = {"User-Agent": "osm-pipeline-example/1.0 (contact@example.org)"}
MIN_INTERVAL = 1.1          # seconds between requests; policy is ~1/sec
_WS = re.compile(r"\s+")
_PUNCT = re.compile(r"[.,;]+")


@dataclass(frozen=True)
class Address:
    street: str
    city: str
    postalcode: str
    country: str

    def key(self) -> str:
        """A normalised, order-stable key used for dedupe and for the cache."""
        parts = (self.street, self.city, self.postalcode, self.country)
        cleaned = []
        for part in parts:
            # NFKC folds compatibility forms; casefold is stronger than lower().
            text = unicodedata.normalize("NFKC", part).casefold()
            text = _PUNCT.sub(" ", text)
            cleaned.append(_WS.sub(" ", text).strip())
        return "|".join(cleaned)


def open_store(path: Path) -> sqlite3.Connection:
    conn = sqlite3.connect(path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS geocode (
            key         TEXT PRIMARY KEY,
            lat         REAL,
            lon         REAL,
            place_rank  INTEGER,
            osm_type    TEXT,
            osm_id      INTEGER,
            resolved_at TEXT NOT NULL,
            status      TEXT NOT NULL      -- 'ok' or 'no_match'
        )
    """)
    conn.commit()
    return conn


def already_done(conn: sqlite3.Connection, key: str) -> bool:
    row = conn.execute("SELECT 1 FROM geocode WHERE key = ?", (key,)).fetchone()
    return row is not None


def geocode_one(session: requests.Session, addr: Address) -> dict | None:
    """One structured, country-constrained lookup. Returns the top candidate."""
    params = {
        "street": addr.street,
        "city": addr.city,
        "postalcode": addr.postalcode,
        "countrycodes": addr.country,      # removes the wrong-country class entirely
        "format": "jsonv2",
        "addressdetails": 1,
        "limit": 1,
    }
    response = session.get(ENDPOINT, params=params, timeout=30)
    if response.status_code in (429, 503):
        wait = float(response.headers.get("Retry-After", 30))
        logger.warning("throttled, sleeping %.0fs", wait)
        time.sleep(wait)
        return geocode_one(session, addr)
    response.raise_for_status()
    results = response.json()
    return results[0] if results else None


def run_batch(addresses: list[Address], store: Path) -> None:
    conn = open_store(store)
    session = requests.Session()
    session.headers.update(HEADERS)

    # Dedupe on the normalised key, preserving one representative per key.
    unique: dict[str, Address] = {}
    for addr in addresses:
        unique.setdefault(addr.key(), addr)
    logger.info("%d row(s) collapsed to %d distinct address(es)",
                len(addresses), len(unique))

    pending = [(k, a) for k, a in unique.items() if not already_done(conn, k)]
    logger.info("%d already cached, %d to fetch", len(unique) - len(pending), len(pending))

    last = 0.0
    for i, (key, addr) in enumerate(pending, start=1):
        # Throttle on the wall clock, not with a fixed sleep: local work is free.
        gap = MIN_INTERVAL - (time.monotonic() - last)
        if gap > 0:
            time.sleep(gap)
        last = time.monotonic()

        hit = geocode_one(session, addr)
        now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        if hit is None:
            conn.execute(
                "INSERT INTO geocode (key, resolved_at, status) VALUES (?, ?, 'no_match')",
                (key, now))
        else:
            conn.execute(
                "INSERT INTO geocode (key, lat, lon, place_rank, osm_type, osm_id,"
                " resolved_at, status) VALUES (?, ?, ?, ?, ?, ?, ?, 'ok')",
                (key, float(hit["lat"]), float(hit["lon"]),
                 int(hit.get("place_rank", -1)), hit.get("osm_type"),
                 int(hit.get("osm_id", 0)), now))
        conn.commit()          # commit per row: a crash loses at most one lookup
        if i % 100 == 0:
            logger.info("%d/%d fetched", i, len(pending))

    conn.close()


if __name__ == "__main__":
    sample = [
        Address("Rynek Główny 1", "Kraków", "31-042", "pl"),
        Address("rynek glowny 1 ", "KRAKÓW", "31-042", "pl"),   # same key
    ]
    run_batch(sample, Path("geocode.sqlite"))