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

"""
enqueue.py — Daemon qui attend des job_triggers de type 'geocoder_enqueue'
et peuple band_locations depuis bands.location_text.

Déclenché depuis l'UI admin → POST /admin/api/job-triggers {job_type: "geocoder_enqueue"}
Tourne en continu (restart: unless-stopped), poll toutes les 15s.
"""

import os
import time
import psycopg2

from parser import (
    parse_location_text,
    country_centroid,
    COUNTRY_NAME_TO_ISO2,
)

POLL_INTERVAL = int(os.environ.get("GEOCODE_ENQUEUE_POLL", "15"))
BATCH         = int(os.environ.get("GEOCODE_ENQUEUE_BATCH", "100000"))


def resolve_iso2(location_raw: str, band_country: str | None) -> str | None:
    t = location_raw.strip()
    if len(t) == 2 and t.upper().isalpha():
        return t.upper()
    iso = COUNTRY_NAME_TO_ISO2.get(t.lower())
    if iso:
        return iso
    if band_country:
        return band_country.strip().upper()
    return None


def run_enqueue(cur) -> tuple[int, int, int]:
    """Parse tous les bands et peuple band_locations. Retourne (inserted, country_fixed, skipped)."""
    cur.execute(
        "SELECT ma_id, country, location_text FROM bands "
        "WHERE location_text IS NOT NULL AND location_text <> '' "
        "ORDER BY ma_id ASC LIMIT %s",
        (BATCH,),
    )
    bands = cur.fetchall()
    print(f"[enqueue] {len(bands)} bands à parser")

    inserted = country_fixed = skipped = failed = 0
    err_samples: list[str] = []

    for ma_id, country, location_text in bands:
        steps = parse_location_text(location_text, country)
        if not steps:
            skipped += 1
            continue

        for step in steps:
            try:
                cur.execute(
                    """
                    INSERT INTO band_locations
                      (ma_id, step_order, step_label, location_raw,
                       is_country_only, geocode_status, geocode_next_at)
                    VALUES (%s, %s, %s, %s, %s, 'queued', now())
                    ON CONFLICT (ma_id, step_order, location_raw) DO NOTHING
                    """,
                    (ma_id, step['step_order'], step['step_label'],
                     step['location_raw'], step['is_country_only']),
                )
                if cur.rowcount > 0:
                    inserted += 1
            except Exception as exc:
                failed += 1
                if len(err_samples) < 5:
                    err_samples.append(f"ma_id={ma_id}: {exc}")
                print(f"[enqueue] WARN insert ma_id={ma_id}: {exc}")

    if failed:
        print(f"[enqueue] {failed} inserts échoués — échantillon: {err_samples}")

    # Résoudre immédiatement les entrées pays-seulement
    cur.execute(
        """
        SELECT bl.id, bl.location_raw, b.country
        FROM band_locations bl
        JOIN bands b ON b.ma_id = bl.ma_id
        WHERE bl.geocode_status = 'queued'
          AND bl.is_country_only = TRUE
        """
    )
    for loc_id, location_raw, band_country in cur.fetchall():
        iso2   = resolve_iso2(location_raw, band_country)
        coords = country_centroid(iso2) if iso2 else None
        if coords:
            lat, lon = coords
            cur.execute(
                """
                UPDATE band_locations
                SET lat=%s, lon=%s,
                    geocode_status='country_only',
                    geocode_provider='country_centroid',
                    geocode_confidence=0.1,
                    geocode_query=%s,
                    updated_at=now()
                WHERE id=%s
                """,
                (lat, lon, iso2, loc_id),
            )
            country_fixed += 1
        else:
            print(f"[enqueue] no centroid id={loc_id} raw='{location_raw}'")

    return inserted, country_fixed, skipped, failed


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

    print(f"[enqueue] daemon démarré, poll toutes les {POLL_INTERVAL}s")

    with conn.cursor() as cur:
        while True:
            # Chercher un job en attente
            cur.execute(
                """
                SELECT id FROM job_triggers
                WHERE job_type = 'geocoder_enqueue' AND status = 'pending'
                ORDER BY created_at ASC
                LIMIT 1
                FOR UPDATE SKIP LOCKED
                """
            )
            row = cur.fetchone()

            if not row:
                time.sleep(POLL_INTERVAL)
                continue

            job_id = row[0]
            cur.execute(
                "UPDATE job_triggers SET status='running', started_at=now() WHERE id=%s",
                (job_id,),
            )
            cur.execute(
                "INSERT INTO crawl_log (level, message) VALUES ('info', %s)",
                (f"[geocoder_enqueue] job #{job_id} démarré",),
            )
            print(f"[enqueue] job #{job_id} démarré")

            try:
                inserted, country_fixed, skipped, failed = run_enqueue(cur)
                summary = (
                    f"[geocoder_enqueue] job #{job_id} terminé — "
                    f"inserted={inserted} country_fixed={country_fixed} "
                    f"skipped={skipped} failed={failed}"
                )
                cur.execute(
                    "UPDATE job_triggers SET status='done', finished_at=now() WHERE id=%s",
                    (job_id,),
                )
                cur.execute(
                    "INSERT INTO crawl_log (level, message) VALUES ('info', %s)",
                    (summary,),
                )
                print(f"[enqueue] {summary}")
            except Exception as exc:
                err = str(exc)[:400]
                cur.execute(
                    "UPDATE job_triggers SET status='error', finished_at=now(), error=%s WHERE id=%s",
                    (err, job_id),
                )
                cur.execute(
                    "INSERT INTO crawl_log (level, message) VALUES ('error', %s)",
                    (f"[geocoder_enqueue] job #{job_id} erreur: {err}",),
                )
                print(f"[enqueue] job #{job_id} erreur: {err}")

    conn.close()


if __name__ == "__main__":
    main()