metalfrom.eu/apps/geocoder/src/worker.py

import os
import time
import random
import json
import requests
import psycopg2

NOMINATIM_BASE = os.environ.get("NOMINATIM_BASE", "https://nominatim.openstreetmap.org").rstrip("/")
NOMINATIM_EMAIL = os.environ.get("NOMINATIM_EMAIL", "").strip()
USER_AGENT = os.environ.get("NOMINATIM_USER_AGENT", "bm-geocoder/0.1 (contact: you@example.com)").strip()

MIN_DELAY = float(os.environ.get("NOMINATIM_MIN_DELAY", "1.05"))  # >= 1s
JITTER = float(os.environ.get("NOMINATIM_JITTER", "0.35"))
MAX_PER_RUN = int(os.environ.get("GEOCODE_MAX_PER_RUN", "100000"))

_last = 0.0

def polite_sleep():
    global _last
    elapsed = time.monotonic() - _last
    wait = max(0.0, MIN_DELAY - elapsed) + random.uniform(0.0, JITTER)
    time.sleep(wait)
    _last = time.monotonic()

def nominatim_search(query: str) -> dict | None:
    # doc: /search + email conseillé pour gros volume :contentReference[oaicite:1]{index=1}
    params = {
        "q": query,
        "format": "jsonv2",
        "limit": 1,
        "addressdetails": 1,
    }
    if NOMINATIM_EMAIL:
        params["email"] = NOMINATIM_EMAIL

    headers = {"User-Agent": USER_AGENT}  # requis par policy :contentReference[oaicite:2]{index=2}

    polite_sleep()
    r = requests.get(f"{NOMINATIM_BASE}/search", params=params, headers=headers, timeout=30)
    if r.status_code in (403, 429, 503):
        raise RuntimeError(f"nominatim_throttle status={r.status_code} body={r.text[:200]}")
    r.raise_for_status()
    arr = r.json()
    if not arr:
        return None
    return arr[0]

def main():
    dsn = os.environ["DATABASE_URL"]
    conn = psycopg2.connect(dsn)
    conn.autocommit = True

    processed = 0
    with conn.cursor() as cur:
        while processed < MAX_PER_RUN:
            # prend 1 job à faire
            cur.execute(
                """
                SELECT ma_id, query, country, tries
                FROM geocode_queue
                WHERE status IN ('queued','error')
                  AND next_run_at <= now()
                ORDER BY next_run_at ASC, ma_id ASC
                LIMIT 1
                FOR UPDATE SKIP LOCKED
                """
            )
            row = cur.fetchone()
            if not row:
                print("[worker] nothing to do. sleeping 60s")
                time.sleep(60)
                continue

            ma_id, query, country, tries = row

            # marque processing
            cur.execute(
                """
                UPDATE geocode_queue
                SET status='processing', updated_at=now()
                WHERE ma_id=%s
                """,
                (ma_id,),
            )

            # 1) cache ?
            cur.execute("SELECT lat, lon, raw FROM geocode_cache WHERE query=%s", (query,))
            cached = cur.fetchone()
            if cached and cached[0] is not None and cached[1] is not None:
                lat, lon, raw = cached
                cur.execute(
                    """
                    UPDATE bands
                    SET lat=%s, lon=%s,
                        geom=ST_SetSRID(ST_MakePoint(%s,%s),4326)::geography,
                        geocoded_at=now(),
                        geocode_provider='nominatim-cache',
                        geocode_query=%s,
                        geocode_raw=%s,
                        geocode_error=NULL,
                        geocode_error_at=NULL
                    WHERE ma_id=%s
                    """,
                    (lat, lon, lon, lat, query, json.dumps(raw), ma_id),
                )
                cur.execute("UPDATE geocode_queue SET status='done', updated_at=now() WHERE ma_id=%s", (ma_id,))
                processed += 1
                print(f"[worker] ma_id={ma_id} cache HIT -> done")
                continue

            # 2) requête nominatim
            try:
                res = nominatim_search(query)
                if not res:
                    cur.execute(
                        """
                        UPDATE bands
                        SET geocode_error=%s, geocode_error_at=now()
                        WHERE ma_id=%s
                        """,
                        ("no_result", ma_id),
                    )
                    cur.execute(
                        """
                        UPDATE geocode_queue
                        SET status='error', tries=tries+1, last_error=%s,
                            next_run_at=now() + interval '7 days',
                            updated_at=now()
                        WHERE ma_id=%s
                        """,
                        ("no_result", ma_id),
                    )
                    processed += 1
                    print(f"[worker] ma_id={ma_id} no_result -> postpone")
                    continue

                lat = float(res["lat"])
                lon = float(res["lon"])

                # écrit cache
                cur.execute(
                    """
                    INSERT INTO geocode_cache(query, provider, lat, lon, geom, raw, updated_at)
                    VALUES (%s,'nominatim',%s,%s,ST_SetSRID(ST_MakePoint(%s,%s),4326)::geography,%s,now())
                    ON CONFLICT (query) DO UPDATE
                      SET lat=EXCLUDED.lat, lon=EXCLUDED.lon, geom=EXCLUDED.geom,
                          raw=EXCLUDED.raw, updated_at=now()
                    """,
                    (query, lat, lon, lon, lat, json.dumps(res)),
                )

                # écrit bands
                cur.execute(
                    """
                    UPDATE bands
                    SET lat=%s, lon=%s,
                        geom=ST_SetSRID(ST_MakePoint(%s,%s),4326)::geography,
                        geocoded_at=now(),
                        geocode_provider='nominatim',
                        geocode_query=%s,
                        geocode_raw=%s,
                        geocode_error=NULL,
                        geocode_error_at=NULL
                    WHERE ma_id=%s
                    """,
                    (lat, lon, lon, lat, query, json.dumps(res), ma_id),
                )

                cur.execute("UPDATE geocode_queue SET status='done', updated_at=now() WHERE ma_id=%s", (ma_id,))
                processed += 1
                print(f"[worker] ma_id={ma_id} OK lat={lat} lon={lon}")

            except Exception as e:
                # backoff progressif
                backoff_minutes = min(60, 5 * (tries + 1))
                cur.execute(
                    """
                    UPDATE bands
                    SET geocode_error=%s, geocode_error_at=now()
                    WHERE ma_id=%s
                    """,
                    (str(e)[:400], ma_id),
                )
                cur.execute(
                    """
                    UPDATE geocode_queue
                    SET status='error', tries=tries+1, last_error=%s,
                        next_run_at=now() + (%s || ' minutes')::interval,
                        updated_at=now()
                    WHERE ma_id=%s
                    """,
                    (str(e)[:400], backoff_minutes, ma_id),
                )
                processed += 1
                print(f"[worker] ma_id={ma_id} ERROR {e} -> retry in {backoff_minutes}m")

    conn.close()

if __name__ == "__main__":
    main()