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

"""
worker.py — Géocodeur Geoapify pour band_locations.

Boucle infinie, traite geocode_status='queued' un par un avec SKIP LOCKED.
Stratégie progressive :
  1. Vérifier geocode_cache (requête déjà faite → gratuit)
  2. Essayer les requêtes fallback du parser (du plus précis au plus vague)
  3. Après MAX_GEO_TRIES échecs → status='llm_needed' pour groq_worker
  4. Succès → mise à jour band_locations + sync bands.lat/lon
"""

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

from parser import build_fallback_queries

GEOAPIFY_API_KEY = os.environ.get("GEOAPIFY_API_KEY", "").strip()
GEOAPIFY_BASE    = "https://api.geoapify.com/v1/geocode/search"

MIN_DELAY     = float(os.environ.get("GEOCODER_MIN_DELAY", "0.22"))
JITTER        = float(os.environ.get("GEOCODER_JITTER", "0.10"))
MAX_PER_RUN   = int(os.environ.get("GEOCODE_MAX_PER_RUN", "100000"))
MAX_GEO_TRIES = int(os.environ.get("GEOCODE_MAX_GEO_TRIES", "3"))

_last_call = 0.0


def _polite_sleep():
    global _last_call
    elapsed = time.monotonic() - _last_call
    wait    = max(0.0, MIN_DELAY - elapsed) + random.uniform(0.0, JITTER)
    if wait > 0:
        time.sleep(wait)
    _last_call = time.monotonic()


def geoapify_search(query: str) -> dict | None:
    if not GEOAPIFY_API_KEY:
        raise RuntimeError("GEOAPIFY_API_KEY not set")
    params = {"text": query, "apiKey": GEOAPIFY_API_KEY, "limit": 1}
    _polite_sleep()
    r = requests.get(GEOAPIFY_BASE, params=params, timeout=30)
    if r.status_code == 429:
        raise RuntimeError("geoapify_throttle:429")
    r.raise_for_status()
    data     = r.json()
    features = data.get("features") or []
    if not features:
        return None
    props  = features[0].get("properties", {})
    coords = features[0].get("geometry", {}).get("coordinates", [])
    if len(coords) < 2:
        return None
    rank       = props.get("rank") or {}
    confidence = rank.get("confidence", 0.5) if isinstance(rank, dict) else 0.5
    return {
        "lat":        props.get("lat") or coords[1],
        "lon":        props.get("lon") or coords[0],
        "confidence": confidence,
        "raw":        data,
    }


def sync_bands_primary(cur, ma_id: int):
    """Met à jour bands.lat/lon avec le step le plus récent géocodé."""
    cur.execute(
        """
        SELECT lat, lon FROM band_locations
        WHERE ma_id = %s
          AND geocode_status IN ('done', 'country_only')
          AND lat IS NOT NULL AND lon IS NOT NULL
        ORDER BY step_order DESC, id DESC
        LIMIT 1
        """,
        (ma_id,),
    )
    row = cur.fetchone()
    if not row:
        return
    lat, lon = row
    cur.execute(
        """
        UPDATE bands
        SET lat=%s, lon=%s,
            geom=ST_SetSRID(ST_MakePoint(%s,%s),4326)::geography,
            geocoded_at=now(),
            geocode_provider='band_locations',
            geocode_error=NULL,
            geocode_error_at=NULL
        WHERE ma_id=%s
        """,
        (lat, lon, lon, lat, ma_id),
    )


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:
            cur.execute(
                """
                SELECT bl.id, bl.ma_id, bl.location_raw,
                       bl.geocode_tries_geo, bl.geocode_query, b.country
                FROM band_locations bl
                JOIN bands b ON b.ma_id = bl.ma_id
                WHERE bl.geocode_status = 'queued'
                  AND bl.geocode_next_at <= now()
                ORDER BY bl.geocode_next_at ASC, bl.id ASC
                LIMIT 1
                FOR UPDATE OF bl SKIP LOCKED
                """
            )
            row = cur.fetchone()
            if not row:
                print("[worker] rien à traiter, attente 60s")
                time.sleep(60)
                continue

            loc_id, ma_id, location_raw, tries, llm_query, country = row

            cur.execute(
                "UPDATE band_locations SET geocode_status='processing', updated_at=now() WHERE id=%s",
                (loc_id,),
            )

            # Liste de requêtes à essayer : requête LLM en premier si disponible
            fallbacks = build_fallback_queries(location_raw, country)
            if llm_query and llm_query not in fallbacks:
                queries = [llm_query] + fallbacks
            else:
                queries = fallbacks

            # Déduplique en gardant l'ordre
            seen: set[str]    = set()
            unique: list[str] = []
            for q in queries:
                if q not in seen:
                    seen.add(q)
                    unique.append(q)

            success    = False
            rate_limit = False
            used_query = None
            lat = lon  = None
            confidence = 0.5
            provider   = 'geoapify'
            last_err   = None

            for query in unique:
                # Cache hit ?
                cur.execute(
                    "SELECT lat, lon FROM geocode_cache WHERE query=%s",
                    (query,),
                )
                cached = cur.fetchone()
                if cached and cached[0] is not None:
                    lat, lon   = cached
                    used_query = query
                    provider   = 'geoapify-cache'
                    success    = True
                    print(f"[worker] id={loc_id} cache HIT '{query}'")
                    break

                try:
                    res = geoapify_search(query)
                    if res:
                        lat        = float(res["lat"])
                        lon        = float(res["lon"])
                        confidence = float(res.get("confidence") or 0.5)
                        used_query = query
                        cur.execute(
                            """
                            INSERT INTO geocode_cache
                              (query, provider, lat, lon, geom, raw, updated_at)
                            VALUES (%s,'geoapify',%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,
                                  provider=EXCLUDED.provider, updated_at=now()
                            """,
                            (query, lat, lon, lon, lat, json.dumps(res["raw"])),
                        )
                        success = True
                        print(f"[worker] id={loc_id} OK '{query}' lat={lat:.4f} lon={lon:.4f}")
                        break
                    else:
                        last_err = f"no_result:'{query}'"
                except RuntimeError as exc:
                    if "429" in str(exc):
                        print("[worker] rate limit Geoapify, attente 60s")
                        cur.execute(
                            """
                            UPDATE band_locations
                            SET geocode_status='queued',
                                geocode_next_at=now() + interval '60 seconds',
                                updated_at=now()
                            WHERE id=%s
                            """,
                            (loc_id,),
                        )
                        rate_limit = True
                        time.sleep(60)
                        break
                    last_err = str(exc)[:300]
                except Exception as exc:
                    last_err = str(exc)[:300]

            if rate_limit:
                continue

            if success:
                cur.execute(
                    """
                    UPDATE band_locations
                    SET lat=%s, lon=%s,
                        geocode_status='done',
                        geocode_query=%s,
                        geocode_provider=%s,
                        geocode_confidence=%s,
                        geocode_error=NULL,
                        updated_at=now()
                    WHERE id=%s
                    """,
                    (lat, lon, used_query, provider, confidence, loc_id),
                )
                sync_bands_primary(cur, ma_id)
            else:
                new_tries = tries + 1
                if new_tries >= MAX_GEO_TRIES:
                    cur.execute(
                        """
                        UPDATE band_locations
                        SET geocode_status='llm_needed',
                            geocode_tries_geo=%s,
                            geocode_error=%s,
                            updated_at=now()
                        WHERE id=%s
                        """,
                        (new_tries, last_err or "all_fallbacks_failed", loc_id),
                    )
                    print(f"[worker] id={loc_id} → llm_needed après {new_tries} essais")
                else:
                    backoff = min(1440, 30 * (2 ** new_tries))
                    cur.execute(
                        """
                        UPDATE band_locations
                        SET geocode_status='queued',
                            geocode_tries_geo=%s,
                            geocode_error=%s,
                            geocode_next_at=now() + (%s || ' minutes')::interval,
                            updated_at=now()
                        WHERE id=%s
                        """,
                        (new_tries, last_err, backoff, loc_id),
                    )
                    print(f"[worker] id={loc_id} retry {new_tries}/{MAX_GEO_TRIES} in {backoff}m: {last_err}")

            processed += 1

    conn.close()
    print(f"[worker] terminé, processed={processed}")


if __name__ == "__main__":
    main()