feat(geocoding): boutons UI pour déclencher l'enqueue band_locations

- enqueue.py devient un daemon qui poll job_triggers (type geocoder_enqueue)
  toutes les 15s — plus de commande manuelle à lancer
- API : geocoder_enqueue ajouté aux job types autorisés
- Admin UI page Géocodage : bouton "Lancer l'enqueue" (vert)
- Admin UI Centre de commandes : même bouton dans la section Géocodeur
- docker-compose : geocoder-enqueue passe à restart: unless-stopped

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

apps/admin/site/app.js

@@ -919,28 +919,91 @@ function conflictCard(b) {
 async function renderGeocoding() {
   if (!document.getElementById("geo-stats")) {
     content().innerHTML = `
-      <div class="grid grid-stats" id="geo-stats"></div>
-      <div class="card" style="margin-bottom:16px">
       <div style="display:flex;align-items:center;justify-content:space-between;flex-wrap:wrap;gap:8px;margin-bottom:12px">
-          <h2 style="margin:0">Progression</h2>
+        <div style="font-size:13px;font-weight:700;color:var(--bone)">Pipeline géocodage</div>
-          <div style="display:flex;gap:8px;flex-wrap:wrap;align-items:center">
-            <button class="btn btn-mini" id="geo-reset-errors">🔄 Réinitialiser erreurs</button>
-            <button class="btn btn-mini" id="geo-requeue-done">♻️ Re-géocoder tout (erreurs + fait)</button>
         <span id="geo-refresh-status" style="font-size:11px;color:var(--muted)">Auto-refresh 30s</span>
       </div>
+
+      <div class="card" style="margin-bottom:16px">
+        <div style="display:flex;align-items:center;justify-content:space-between;flex-wrap:wrap;gap:8px;margin-bottom:10px">
+          <h2 style="margin:0">Nouveau pipeline — band_locations</h2>
+          <div style="display:flex;gap:8px;flex-wrap:wrap">
+            <button class="btn btn-mini" id="loc-enqueue" style="border-color:rgba(63,174,90,0.5);color:var(--ok)">Lancer l'enqueue</button>
+            <button class="btn btn-mini" id="loc-reset-errors">Reset erreurs</button>
+            <button class="btn btn-mini" id="loc-reset-llm">Reset LLM needed</button>
+            <button class="btn btn-mini" id="loc-requeue-all">Re-queue tout</button>
           </div>
-        <div style="background:rgba(255,255,255,0.06);border-radius:8px;height:18px;overflow:hidden;margin-bottom:8px">
+        </div>
-          <div id="geo-bar" style="width:0%;height:100%;background:linear-gradient(90deg,var(--blood),rgba(198,26,26,0.6));transition:width 0.4s ease"></div>
+        <div class="grid grid-stats" id="loc-stats" style="margin-bottom:12px"></div>
+        <div style="background:rgba(255,255,255,0.06);border-radius:8px;height:12px;overflow:hidden;margin-bottom:6px">
+          <div id="loc-bar" style="width:0%;height:100%;background:linear-gradient(90deg,var(--blood),rgba(198,26,26,0.6));transition:width 0.4s ease"></div>
+        </div>
+        <div id="loc-pct" style="font-size:12px;color:var(--muted)">—</div>
+        <div id="loc-feedback" style="margin-top:8px;font-size:12px"></div>
+      </div>
+
+      <div class="card" style="margin-bottom:16px">
+        <div style="display:flex;align-items:center;justify-content:space-between;flex-wrap:wrap;gap:8px;margin-bottom:10px">
+          <h2 style="margin:0">Ancien pipeline — geocode_queue</h2>
+          <div style="display:flex;gap:8px;flex-wrap:wrap">
+            <button class="btn btn-mini" id="geo-reset-errors">Reset erreurs</button>
+            <button class="btn btn-mini" id="geo-requeue-done">Re-queue tout</button>
+          </div>
+        </div>
+        <div class="grid grid-stats" id="geo-stats" style="margin-bottom:12px"></div>
+        <div style="background:rgba(255,255,255,0.06);border-radius:8px;height:12px;overflow:hidden;margin-bottom:6px">
+          <div id="geo-bar" style="width:0%;height:100%;background:linear-gradient(90deg,#3fae5a,rgba(63,174,90,0.5));transition:width 0.4s ease"></div>
         </div>
         <div id="geo-pct" style="font-size:12px;color:var(--muted)">—</div>
         <div id="geo-feedback" style="margin-top:8px;font-size:12px"></div>
       </div>
+
       <div class="card">
-        <h2>20 derniers géocodages</h2>
+        <h2>20 derniers géocodages (bands)</h2>
         <div class="table-wrap" id="geo-recent-wrap"><div class="loading">Chargement…</div></div>
       </div>
     `;
 
+    // Nouveau pipeline — actions
+    const makeLocAction = (btnId, fbId, url, body, confirmMsg) => {
+      document.getElementById(btnId).addEventListener("click", async () => {
+        if (confirmMsg && !confirm(confirmMsg)) return;
+        const btn = document.getElementById(btnId);
+        btn.disabled = true;
+        const fb = document.getElementById(fbId);
+        fb.textContent = "En cours…"; fb.style.color = "var(--muted)";
+        try {
+          const r = await api(url, { method: "POST", body: JSON.stringify(body || {}) });
+          fb.textContent = `✓ ${r.count} entrée(s) modifiée(s).`;
+          fb.style.color = "var(--ok)";
+          await loadGeocoding();
+        } catch (e) {
+          fb.textContent = `Erreur: ${e.message}`; fb.style.color = "var(--err)";
+        } finally { btn.disabled = false; }
+      });
+    };
+
+    document.getElementById("loc-enqueue").addEventListener("click", async () => {
+      const btn = document.getElementById("loc-enqueue");
+      btn.disabled = true;
+      const fb = document.getElementById("loc-feedback");
+      fb.textContent = "Job envoyé, le daemon va démarrer l'enqueue (peut prendre quelques secondes)…";
+      fb.style.color = "var(--muted)";
+      try {
+        await api("/admin/api/job-triggers", { method: "POST", body: JSON.stringify({ job_type: "geocoder_enqueue" }) });
+        fb.textContent = "✓ Job geocoder_enqueue créé — le daemon parse les bands en arrière-plan.";
+        fb.style.color = "var(--ok)";
+        setTimeout(() => loadGeocoding(), 3000);
+      } catch (e) {
+        fb.textContent = `Erreur: ${e.message}`; fb.style.color = "var(--err)";
+      } finally { btn.disabled = false; }
+    });
+
+    makeLocAction("loc-reset-errors", "loc-feedback", "/admin/api/locations/reset-errors", {}, null);
+    makeLocAction("loc-reset-llm",    "loc-feedback", "/admin/api/locations/reset-llm",    {}, "Remettre en queue les entrées llm_needed et manual ?");
+    makeLocAction("loc-requeue-all",  "loc-feedback", "/admin/api/locations/requeue-all",  { include_done: true }, "Re-queue TOUTES les band_locations (erreurs + LLM + manual + done) ?");
+
+    // Ancien pipeline — actions
     document.getElementById("geo-reset-errors").addEventListener("click", async () => {
       const btn = document.getElementById("geo-reset-errors");
       btn.disabled = true;
@@ -948,7 +1011,7 @@ async function renderGeocoding() {
       fb.textContent = "Réinitialisation…"; fb.style.color = "var(--muted)";
       try {
         const r = await api("/admin/api/geocoding/reset-errors", { method: "POST", body: JSON.stringify({}) });
-        fb.textContent = `✓ ${r.count} entrée(s) remise(s) en queue — le géocodeur les traite automatiquement.`;
+        fb.textContent = `✓ ${r.count} entrée(s) remise(s) en queue.`;
         fb.style.color = "var(--ok)";
         await loadGeocoding();
       } catch (e) {
@@ -957,7 +1020,7 @@ async function renderGeocoding() {
     });
 
     document.getElementById("geo-requeue-done").addEventListener("click", async () => {
-      if (!confirm("Réinitialiser TOUTES les entrées (erreurs + déjà géocodées) ? Ceci relance le géocodage complet.")) return;
+      if (!confirm("Réinitialiser TOUTES les entrées geocode_queue (erreurs + fait) ?")) return;
       const btn = document.getElementById("geo-requeue-done");
       btn.disabled = true;
       const fb = document.getElementById("geo-feedback");
@@ -978,13 +1041,47 @@ async function renderGeocoding() {
   state.geoTimer = setInterval(async () => {
     await loadGeocoding();
     const el = document.getElementById("geo-refresh-status");
-    if (el) el.textContent = `Actualisé à ${new Date().toLocaleTimeString("fr-FR")}`;
+    if (el) el.textContent = `Actualisé ${new Date().toLocaleTimeString("fr-FR")}`;
   }, 30000);
 }
 
+const LOC_STATUS_LABELS = {
+  queued: "En queue", processing: "En cours", done: "Done",
+  error: "Erreur", llm_needed: "LLM needed", manual: "Manuel",
+  country_only: "Pays (centroïde)", skipped: "Ignoré",
+};
+
 async function loadGeocoding() {
   try {
     const r = await api("/admin/api/geocoding");
+
+    // Nouveau pipeline — band_locations
+    const locs     = r.locations || [];
+    const locTotal = locs.reduce((s, x) => s + x.n, 0);
+    const locDone  = (locs.find((x) => x.status === "done")?.n || 0) +
+                     (locs.find((x) => x.status === "country_only")?.n || 0);
+    const locLlm   = locs.find((x) => x.status === "llm_needed")?.n || 0;
+    const locManual = locs.find((x) => x.status === "manual")?.n || 0;
+    const locErr   = locs.find((x) => x.status === "error")?.n || 0;
+    const locPct   = locTotal ? Math.round((locDone / locTotal) * 100) : 0;
+    const llmCost  = parseFloat(r.llm_cache?.total_cost_usd || 0).toFixed(4);
+
+    const locStatsEl = document.getElementById("loc-stats");
+    if (locStatsEl) locStatsEl.innerHTML = `
+      ${statCard("Total locations", locTotal)}
+      ${statCard("Géocodées", locDone, "ok")}
+      ${statCard("LLM needed", locLlm, locLlm ? "warn" : "")}
+      ${statCard("Manuel", locManual, locManual ? "warn" : "")}
+      ${statCard("Erreurs", locErr, locErr ? "err" : "")}
+      ${statCard("LLM cache", r.llm_cache?.n || 0)}
+      ${statCard("Coût LLM (USD)", "$" + llmCost)}
+    `;
+    const locBar = document.getElementById("loc-bar");
+    if (locBar) locBar.style.width = locPct + "%";
+    const locPctEl = document.getElementById("loc-pct");
+    if (locPctEl) locPctEl.textContent = `${locPct}% — ${locDone.toLocaleString("fr-FR")} / ${locTotal.toLocaleString("fr-FR")} — statuts: ${locs.map((x) => `${LOC_STATUS_LABELS[x.status] || x.status} ${x.n}`).join(", ")}`;
+
+    // Ancien pipeline — geocode_queue
     const total      = r.queue.reduce((s, x) => s + x.n, 0);
     const done       = r.queue.find((x) => x.status === "done")?.n || 0;
     const queued     = r.queue.find((x) => x.status === "queued")?.n || 0;
@@ -995,13 +1092,12 @@ async function loadGeocoding() {
     const statsEl = document.getElementById("geo-stats");
     if (statsEl) statsEl.innerHTML = `
       ${statCard("Total queue", total)}
-      ${statCard("Géocodés ✓", done, "ok")}
+      ${statCard("Géocodés", done, "ok")}
-      ${statCard("En attente", queued, "warn")}
+      ${statCard("En attente", queued, queued ? "warn" : "")}
-      ${statCard("En cours", processing || 0, processing ? "ok" : "")}
+      ${statCard("En cours", processing, processing ? "ok" : "")}
       ${statCard("Erreurs", errored, errored ? "err" : "")}
       ${statCard("Cache Geoapify", r.cache_size)}
     `;
-
     const bar = document.getElementById("geo-bar");
     if (bar) bar.style.width = pct + "%";
     const pctEl = document.getElementById("geo-pct");
@@ -1048,10 +1144,11 @@ async function renderJobs() {
 
       <div class="card">
         <h2>🗺 Géocodeur</h2>
-        <p style="font-size:12px;color:var(--muted);margin:0 0 14px">Le géocodeur tourne en continu — ces actions modifient la queue directement.</p>
+        <p style="font-size:12px;color:var(--muted);margin:0 0 14px">Le géocodeur et le worker Groq tournent en continu. L'enqueue parse les bands en band_locations.</p>
         <div style="display:flex;flex-direction:column;gap:8px">
-          <button class="btn" id="cmd-reset-errors">🔄 Réinitialiser les erreurs de géocodage</button>
+          <button class="btn" id="cmd-enqueue" style="border-color:rgba(63,174,90,0.5);color:var(--ok)">▶ Lancer l'enqueue band_locations</button>
-          <button class="btn" id="cmd-requeue-all" style="border-color:rgba(198,26,26,0.5)">♻️ Re-géocoder tout (erreurs + déjà fait)</button>
+          <button class="btn" id="cmd-reset-errors">🔄 Reset erreurs geocode_queue</button>
+          <button class="btn" id="cmd-requeue-all" style="border-color:rgba(198,26,26,0.5)">♻️ Re-géocoder tout geocode_queue</button>
         </div>
         <div id="geo-cmd-feedback" style="margin-top:10px;font-size:12px;color:var(--muted)"></div>
       </div>
@@ -1098,6 +1195,22 @@ async function renderJobs() {
     });
   });
 
+  // Geocoder enqueue
+  document.getElementById("cmd-enqueue").addEventListener("click", async () => {
+    const btn = document.getElementById("cmd-enqueue");
+    btn.disabled = true;
+    const fb = document.getElementById("geo-cmd-feedback");
+    fb.textContent = "Envoi du job…"; fb.style.color = "var(--muted)";
+    try {
+      await api("/admin/api/job-triggers", { method: "POST", body: JSON.stringify({ job_type: "geocoder_enqueue" }) });
+      fb.textContent = "✓ Job geocoder_enqueue créé — le daemon parse les bands en arrière-plan.";
+      fb.style.color = "var(--ok)";
+      await loadCmdLogTail();
+    } catch (e) {
+      fb.textContent = `Erreur: ${e.message}`; fb.style.color = "var(--err)";
+    } finally { btn.disabled = false; }
+  });
+
   // Geocoder reset errors
   document.getElementById("cmd-reset-errors").addEventListener("click", async () => {
     const btn = document.getElementById("cmd-reset-errors");

apps/api/migrations/008_band_locations.sql

@@ -0,0 +1,54 @@
+-- 008_band_locations.sql
+-- Table relationnelle N steps × M villes par band pour le géocodage multi-périodes.
+-- Remplace la logique mono-entrée de geocode_queue pour les nouveaux traitements.
+
+CREATE TABLE IF NOT EXISTS band_locations (
+  id                 BIGSERIAL PRIMARY KEY,
+  ma_id              INTEGER NOT NULL REFERENCES bands(ma_id) ON DELETE CASCADE,
+  step_order         SMALLINT NOT NULL DEFAULT 0,
+  step_label         TEXT,
+  location_raw       TEXT NOT NULL,
+  is_country_only    BOOLEAN NOT NULL DEFAULT FALSE,
+  lat                DOUBLE PRECISION,
+  lon                DOUBLE PRECISION,
+  geocode_status     TEXT NOT NULL DEFAULT 'queued',
+  geocode_query      TEXT,
+  geocode_provider   TEXT,
+  geocode_confidence REAL,
+  geocode_error      TEXT,
+  geocode_tries_geo  SMALLINT NOT NULL DEFAULT 0,
+  geocode_tries_llm  SMALLINT NOT NULL DEFAULT 0,
+  geocode_next_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
+  created_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
+  UNIQUE (ma_id, step_order, location_raw)
+);
+
+-- Index pour la boucle de travail du geocoder
+CREATE INDEX IF NOT EXISTS idx_bl_geo_work
+  ON band_locations (geocode_next_at ASC, id ASC)
+  WHERE geocode_status = 'queued';
+
+-- Index pour la boucle du groq-worker
+CREATE INDEX IF NOT EXISTS idx_bl_llm_work
+  ON band_locations (geocode_tries_llm ASC, id ASC)
+  WHERE geocode_status = 'llm_needed';
+
+-- Index de navigation par band
+CREATE INDEX IF NOT EXISTS idx_bl_ma_id ON band_locations (ma_id);
+
+-- Cache pour éviter de rappeler le LLM pour la même entrée
+CREATE TABLE IF NOT EXISTS llm_cache (
+  id             BIGSERIAL PRIMARY KEY,
+  input_hash     TEXT NOT NULL UNIQUE,
+  model          TEXT NOT NULL,
+  prompt         TEXT NOT NULL,
+  response       TEXT NOT NULL,
+  parsed_city    TEXT,
+  parsed_country TEXT,
+  is_null        BOOLEAN NOT NULL DEFAULT FALSE,
+  tokens_in      INTEGER,
+  tokens_out     INTEGER,
+  cost_usd       NUMERIC(10, 8),
+  created_at     TIMESTAMPTZ NOT NULL DEFAULT now()
+);

apps/api/src/adminRoutes.js

@@ -432,7 +432,7 @@ export default async function adminRoutes(fastify, opts) {
   // ------------------------------------------------------------------
   fastify.get("/admin/api/geocoding", async (req, reply) => {
     try {
-      const [queue, cache, recent] = await Promise.all([
+      const [queue, cache, recent, locations, llm] = await Promise.all([
         pool.query(`
           SELECT status, count(*)::int AS n FROM geocode_queue GROUP BY status ORDER BY n DESC
         `),
@@ -444,12 +444,23 @@ export default async function adminRoutes(fastify, opts) {
           WHERE b.geocoded_at IS NOT NULL
           ORDER BY b.geocoded_at DESC LIMIT 20
         `),
+        pool.query(`
+          SELECT geocode_status AS status, count(*)::int AS n
+          FROM band_locations
+          GROUP BY geocode_status ORDER BY n DESC
+        `).catch(() => ({ rows: [] })),
+        pool.query(`
+          SELECT count(*)::int AS n, sum(cost_usd)::numeric(10,6) AS total_cost_usd
+          FROM llm_cache
+        `).catch(() => ({ rows: [{ n: 0, total_cost_usd: 0 }] })),
       ]);
       return {
         ok: true,
         queue: queue.rows,
         cache_size: cache.rows[0].n,
         recent: recent.rows,
+        locations: locations.rows,
+        llm_cache: llm.rows[0],
       };
     } catch (err) {
       fastify.log.error(err);
@@ -476,7 +487,7 @@ export default async function adminRoutes(fastify, opts) {
   fastify.post("/admin/api/job-triggers", async (req, reply) => {
     try {
       const { job_type } = req.body || {};
-      const allowed = ["enrich", "incremental", "full_crawl"];
+      const allowed = ["enrich", "incremental", "full_crawl", "geocoder_enqueue"];
       if (!allowed.includes(job_type)) {
         return reply.code(400).send({ ok: false, error: `job_type doit être: ${allowed.join(", ")}` });
       }
@@ -539,6 +550,74 @@ export default async function adminRoutes(fastify, opts) {
     }
   });
 
+  // ------------------------------------------------------------------
+  // Géocodage — actions sur band_locations (nouveau pipeline)
+  // ------------------------------------------------------------------
+  fastify.post("/admin/api/locations/reset-errors", async (req, reply) => {
+    try {
+      const r = await pool.query(`
+        UPDATE band_locations
+        SET geocode_status='queued', geocode_tries_geo=0,
+            geocode_error=NULL, geocode_next_at=now(), updated_at=now()
+        WHERE geocode_status = 'error'
+      `);
+      const count = r.rowCount;
+      await pool.query(
+        `INSERT INTO crawl_log (level, message) VALUES ($1, $2)`,
+        ["info", `[admin:${req.adminUsername}] locations reset-errors: ${count} remise(s) en queue`]
+      ).catch(() => {});
+      return { ok: true, count };
+    } catch (err) {
+      fastify.log.error(err);
+      return reply.code(500).send({ ok: false, error: "Erreur locations reset-errors" });
+    }
+  });
+
+  fastify.post("/admin/api/locations/reset-llm", async (req, reply) => {
+    try {
+      const r = await pool.query(`
+        UPDATE band_locations
+        SET geocode_status='queued', geocode_tries_geo=0, geocode_tries_llm=0,
+            geocode_query=NULL, geocode_error=NULL, geocode_next_at=now(), updated_at=now()
+        WHERE geocode_status IN ('llm_needed', 'manual')
+      `);
+      const count = r.rowCount;
+      await pool.query(
+        `INSERT INTO crawl_log (level, message) VALUES ($1, $2)`,
+        ["info", `[admin:${req.adminUsername}] locations reset-llm: ${count} remise(s) en queue`]
+      ).catch(() => {});
+      return { ok: true, count };
+    } catch (err) {
+      fastify.log.error(err);
+      return reply.code(500).send({ ok: false, error: "Erreur locations reset-llm" });
+    }
+  });
+
+  fastify.post("/admin/api/locations/requeue-all", async (req, reply) => {
+    try {
+      const { include_done = false } = req.body || {};
+      const statuses = include_done
+        ? ["error", "llm_needed", "manual", "done"]
+        : ["error", "llm_needed"];
+      const r = await pool.query(`
+        UPDATE band_locations
+        SET geocode_status='queued', geocode_tries_geo=0, geocode_tries_llm=0,
+            geocode_query=NULL, geocode_error=NULL, geocode_next_at=now(), updated_at=now()
+        WHERE geocode_status = ANY($1::text[])
+          AND is_country_only = FALSE
+      `, [statuses]);
+      const count = r.rowCount;
+      await pool.query(
+        `INSERT INTO crawl_log (level, message) VALUES ($1, $2)`,
+        ["info", `[admin:${req.adminUsername}] locations requeue-all (include_done=${include_done}): ${count}`]
+      ).catch(() => {});
+      return { ok: true, count };
+    } catch (err) {
+      fastify.log.error(err);
+      return reply.code(500).send({ ok: false, error: "Erreur locations requeue-all" });
+    }
+  });
+
   // ------------------------------------------------------------------
   // Live status (agrégé pour le Monitor)
   // ------------------------------------------------------------------

apps/geocoder/src/enqueue.py

@@ -1,109 +1,174 @@
+"""
+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
 
-COUNTRY_FALLBACK = {
+from parser import (
-    "FR": "France",
+    parse_location_text,
-    "DE": "Germany",
+    country_centroid,
-    "IT": "Italy",
+    COUNTRY_NAME_TO_ISO2,
-    "GB": "United Kingdom",
+)
-    "ES": "Spain",
+
-    "SE": "Sweden",
+POLL_INTERVAL = int(os.environ.get("GEOCODE_ENQUEUE_POLL", "15"))
-    "NO": "Norway",
+BATCH         = int(os.environ.get("GEOCODE_ENQUEUE_BATCH", "100000"))
-    "FI": "Finland",
+
-    "PL": "Poland",
+
-    "NL": "Netherlands",
+def resolve_iso2(location_raw: str, band_country: str | None) -> str | None:
-    "BE": "Belgium",
+    t = location_raw.strip()
-    "CH": "Switzerland",
+    if len(t) == 2 and t.upper().isalpha():
-    "AT": "Austria",
+        return t.upper()
-    "PT": "Portugal",
+    iso = COUNTRY_NAME_TO_ISO2.get(t.lower())
-    "GR": "Greece",
+    if iso:
-    "CZ": "Czech Republic",
+        return iso
-    "SK": "Slovakia",
+    if band_country:
-    "SI": "Slovenia",
+        return band_country.strip().upper()
-    "HU": "Hungary",
+    return None
-    "UA": "Ukraine",
+
-    "RU": "Russia",
+
-    "DK": "Denmark",
+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 = 0
+
+    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:
+                print(f"[enqueue] WARN insert ma_id={ma_id}: {exc}")
+
+    # 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
 
-def make_query(location_text: str, country_code: str) -> str:
-    loc = (location_text or "").strip()
-    cc = (country_code or "").strip().upper()
-    ctry = COUNTRY_FALLBACK.get(cc, cc)
-    return f"{loc}, {ctry}" if loc else ctry
 
 def main():
     dsn  = os.environ["DATABASE_URL"]
-    batch = int(os.environ.get("GEOCODE_ENQUEUE_BATCH", "5000"))
-
-    # Retry policy
-    retry_after_hours = int(os.environ.get("GEOCODE_RETRY_AFTER_HOURS", "72"))  # 72h = 3 jours
-    force_retry = os.environ.get("GEOCODE_FORCE_RETRY", "0").strip() in ("1", "true", "yes", "on")
-
     conn = psycopg2.connect(dsn)
     conn.autocommit = True
 
+    print(f"[enqueue] daemon démarré, poll toutes les {POLL_INTERVAL}s")
+
     with conn.cursor() as cur:
-        # On sélectionne :
+        while True:
-        # - bands non géocodés avec location_text
+            # Chercher un job en attente
-        # - soit pas encore en queue, soit queue en error/queued et next_run_at <= now()
-        # - et si erreur récente: on respecte retry_after_hours sauf si force_retry
             cur.execute(
                 """
-            SELECT
+                SELECT id FROM job_triggers
-              b.ma_id, b.country, b.location_text
+                WHERE job_type = 'geocoder_enqueue' AND status = 'pending'
-            FROM bands b
+                ORDER BY created_at ASC
-            LEFT JOIN geocode_queue q ON q.ma_id = b.ma_id
+                LIMIT 1
-            WHERE b.geom IS NULL
+                FOR UPDATE SKIP LOCKED
-              AND b.location_text IS NOT NULL AND b.location_text <> ''
-              AND (
-                   q.ma_id IS NULL
-                   OR (q.status IN ('queued','error') AND q.next_run_at <= now())
-              )
-              AND (
-                   %s
-                   OR b.geocode_error IS NULL
-                   OR b.geocode_error_at IS NULL
-                   OR b.geocode_error_at < now() - (%s || ' hours')::interval
-              )
-            ORDER BY
-              COALESCE(q.next_run_at, now()) ASC,
-              b.ma_id ASC
-            LIMIT %s
-            """,
-            (force_retry, retry_after_hours, batch),
-        )
-
-        rows = cur.fetchall()
-        enq = 0
-
-        for ma_id, country, location_text in rows:
-            qtxt = make_query(location_text, country)
-
-            cur.execute(
                 """
-                INSERT INTO geocode_queue (ma_id, query, country, status, next_run_at)
-                VALUES (%s, %s, %s, 'queued', now())
-                ON CONFLICT (ma_id) DO UPDATE
-                  SET query = EXCLUDED.query,
-                      country = EXCLUDED.country,
-                      status = CASE
-                        WHEN geocode_queue.status IN ('done','processing') THEN geocode_queue.status
-                        ELSE 'queued'
-                      END,
-                      next_run_at = CASE
-                        WHEN geocode_queue.status IN ('done','processing') THEN geocode_queue.next_run_at
-                        ELSE now()
-                      END,
-                      updated_at = now()
-                """,
-                (ma_id, qtxt, country),
             )
-            enq += 1
+            row = cur.fetchone()
 
-        print(f"[enqueue] queued_or_updated={enq} (selected={len(rows)}) force_retry={force_retry} retry_after_hours={retry_after_hours}")
+            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 = run_enqueue(cur)
+                summary = (
+                    f"[geocoder_enqueue] job #{job_id} terminé — "
+                    f"inserted={inserted} country_fixed={country_fixed} skipped={skipped}"
+                )
+                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()

apps/geocoder/src/groq_worker.py

@@ -0,0 +1,315 @@
+"""
+groq_worker.py — Pipeline LLM Groq (free tier) pour désambiguïser les locations
+que Geoapify ne sait pas géocoder après MAX_GEO_TRIES tentatives.
+
+Flux :
+  1. Sélectionner band_locations WHERE geocode_status='llm_needed'
+  2. Vérifier llm_cache (sha256 de model+country+location_raw)
+  3. Sinon appeler Groq (JSON mode garanti) → extraire city + iso2
+  4. Si ville trouvée → geocode_query=clean_query, status='queued', tries_geo=0
+  5. Si aucune ville (is_null) → incrémenter tries_llm, backoff exponentiel
+  6. Après MAX_LLM_TRIES → status='manual'
+
+Rate limits Groq free tier :
+  llama-3.3-70b-versatile : 30 req/min, 1 000 req/jour
+  llama-3.1-8b-instant    : 30 req/min, 14 400 req/jour (fallback)
+"""
+
+import os
+import sys
+import time
+import json
+import hashlib
+import re
+import requests
+import psycopg2
+from datetime import datetime, timezone
+
+# sys.path[0] = src/ quand lancé comme "python src/groq_worker.py"
+from parser import COUNTRY_NAMES
+
+GROQ_API_KEY  = os.environ.get("GROQ_API_KEY", "").strip()
+GROQ_BASE     = "https://api.groq.com/openai/v1/chat/completions"
+MAX_LLM_TRIES = int(os.environ.get("GROQ_MAX_TRIES", "5"))
+CALL_DELAY    = float(os.environ.get("GROQ_CALL_DELAY", "2.5"))
+
+MODELS = [
+    # (model_id, req_per_min, req_per_day)
+    ("llama-3.3-70b-versatile", 30, 1000),
+    ("llama-3.1-8b-instant",    30, 14400),
+]
+
+_SYSTEM = (
+    "You are a precise location data extraction assistant. "
+    "Your only job is to extract a city name and ISO 3166-1 alpha-2 country code "
+    "from a location description. Always respond with valid JSON only."
+)
+
+_USER_TMPL = """\
+Extract the primary city (or town/village) and ISO 3166-1 alpha-2 country code \
+from this location.
+
+Band's registered country: {country}
+Location text: "{location}"
+
+Rules:
+- Use the band's country as context if the location is ambiguous
+- Return the most specific city/town name you can extract
+- Strip all administrative divisions: keep just the city name
+- If there are multiple cities separated by "/" or "and", return the first one
+- If no real place can be extracted, set city to null
+
+Respond ONLY with this JSON (no other text):
+{{"city": "city name or null", "country": "ISO2 code", "confidence": 0.0}}"""
+
+
+def _input_hash(model: str, location_raw: str, country: str) -> str:
+    key = f"{model}|{(country or '').upper()}|{location_raw.strip().lower()}"
+    return hashlib.sha256(key.encode()).hexdigest()
+
+
+def _call_groq(model: str, location_raw: str, country: str) -> tuple[str, int, int]:
+    prompt = _USER_TMPL.format(
+        country=country or "unknown (European metal band)",
+        location=location_raw,
+    )
+    r = requests.post(
+        GROQ_BASE,
+        headers={
+            "Authorization": f"Bearer {GROQ_API_KEY}",
+            "Content-Type":  "application/json",
+        },
+        json={
+            "model":           model,
+            "messages":        [
+                {"role": "system", "content": _SYSTEM},
+                {"role": "user",   "content": prompt},
+            ],
+            "max_tokens":      80,
+            "temperature":     0.0,
+            "response_format": {"type": "json_object"},
+        },
+        timeout=30,
+    )
+    if r.status_code == 429:
+        raise RuntimeError(f"rate_limit:{model}")
+    r.raise_for_status()
+    data  = r.json()
+    text  = data["choices"][0]["message"]["content"]
+    usage = data.get("usage", {})
+    return text, usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0)
+
+
+def _parse_json(text: str) -> dict | None:
+    try:
+        return json.loads(text.strip())
+    except json.JSONDecodeError:
+        m = re.search(r'\{[^}]+\}', text, re.DOTALL)
+        if m:
+            try:
+                return json.loads(m.group())
+            except json.JSONDecodeError:
+                pass
+    return None
+
+
+def _apply_clean_query(cur, loc_id: int, city: str, iso2: str | None, tries_llm: int):
+    country_name = COUNTRY_NAMES.get((iso2 or '').upper(), iso2 or '')
+    clean_query  = f"{city}, {country_name}" if country_name else city
+    cur.execute(
+        """
+        UPDATE band_locations
+        SET geocode_status='queued',
+            geocode_query=%s,
+            geocode_tries_geo=0,
+            geocode_tries_llm=%s,
+            geocode_next_at=now(),
+            geocode_error=NULL,
+            updated_at=now()
+        WHERE id=%s
+        """,
+        (clean_query, tries_llm + 1, loc_id),
+    )
+
+
+def main():
+    if not GROQ_API_KEY:
+        print("[groq] GROQ_API_KEY non configuré — arrêt")
+        sys.exit(0)
+
+    dsn  = os.environ["DATABASE_URL"]
+    conn = psycopg2.connect(dsn)
+    conn.autocommit = True
+
+    day_count = {m[0]: 0 for m in MODELS}
+    min_count = {m[0]: 0 for m in MODELS}
+    min_start = time.monotonic()
+    day_start = datetime.now(timezone.utc).date()
+
+    with conn.cursor() as cur:
+        while True:
+            # Reset compteurs si nouvelle minute / nouveau jour
+            if time.monotonic() - min_start >= 60:
+                min_count = {m[0]: 0 for m in MODELS}
+                min_start = time.monotonic()
+            today = datetime.now(timezone.utc).date()
+            if today != day_start:
+                day_count = {m[0]: 0 for m in MODELS}
+                day_start = today
+
+            cur.execute(
+                """
+                SELECT bl.id, bl.ma_id, bl.location_raw, bl.geocode_tries_llm,
+                       b.country
+                FROM band_locations bl
+                JOIN bands b ON b.ma_id = bl.ma_id
+                WHERE bl.geocode_status = 'llm_needed'
+                  AND bl.geocode_next_at <= now()
+                ORDER BY bl.geocode_tries_llm ASC, bl.id ASC
+                LIMIT 1
+                FOR UPDATE OF bl SKIP LOCKED
+                """
+            )
+            row = cur.fetchone()
+            if not row:
+                print("[groq] rien à traiter, attente 120s")
+                time.sleep(120)
+                continue
+
+            loc_id, ma_id, location_raw, tries_llm, country = row
+
+            # Choisir un modèle disponible
+            chosen_model = None
+            for model, rpm, rpd in MODELS:
+                if min_count[model] < rpm and day_count[model] < rpd:
+                    chosen_model = model
+                    break
+
+            if not chosen_model:
+                wait = max(5, 60 - (time.monotonic() - min_start))
+                print(f"[groq] quota atteint, attente {wait:.0f}s")
+                cur.execute(
+                    "UPDATE band_locations SET geocode_status='llm_needed', updated_at=now() WHERE id=%s",
+                    (loc_id,),
+                )
+                time.sleep(wait)
+                continue
+
+            # Vérifier llm_cache
+            cache_hash = _input_hash(chosen_model, location_raw, country or '')
+            cur.execute(
+                "SELECT parsed_city, parsed_country, is_null FROM llm_cache WHERE input_hash=%s",
+                (cache_hash,),
+            )
+            cached = cur.fetchone()
+
+            if cached:
+                parsed_city, parsed_country, is_null = cached
+                print(f"[groq] id={loc_id} CACHE HIT city={parsed_city}")
+                if is_null or not parsed_city:
+                    _handle_null_city(cur, loc_id, tries_llm)
+                else:
+                    _apply_clean_query(cur, loc_id, parsed_city, parsed_country, tries_llm)
+                continue
+
+            # Appel API Groq
+            try:
+                response_text, tok_in, tok_out = _call_groq(
+                    chosen_model, location_raw, country or ''
+                )
+                min_count[chosen_model] += 1
+                day_count[chosen_model] += 1
+
+                parsed      = _parse_json(response_text)
+                city        = (parsed or {}).get("city")
+                iso2        = (parsed or {}).get("country") or country
+                is_null     = not city
+                # Coût approximatif llama 70B
+                cost_usd    = (tok_in * 0.00000059 + tok_out * 0.00000079)
+
+                prompt_text = _USER_TMPL.format(
+                    country=country or "unknown (European metal band)",
+                    location=location_raw,
+                )
+                cur.execute(
+                    """
+                    INSERT INTO llm_cache
+                      (input_hash, model, prompt, response,
+                       parsed_city, parsed_country, is_null,
+                       tokens_in, tokens_out, cost_usd)
+                    VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
+                    ON CONFLICT (input_hash) DO NOTHING
+                    """,
+                    (cache_hash, chosen_model, prompt_text, response_text,
+                     city, iso2, is_null, tok_in, tok_out, cost_usd),
+                )
+
+                if is_null:
+                    print(f"[groq] id={loc_id} city=null (tries_llm={tries_llm + 1})")
+                    _handle_null_city(cur, loc_id, tries_llm)
+                else:
+                    _apply_clean_query(cur, loc_id, city, iso2, tries_llm)
+                    print(f"[groq] id={loc_id} → queued city='{city}' ({iso2})")
+
+            except RuntimeError as exc:
+                if "rate_limit" in str(exc):
+                    hit_model = str(exc).split(":", 1)[1] if ":" in str(exc) else chosen_model
+                    min_count[hit_model] = 9999
+                    print(f"[groq] rate limit {hit_model}")
+                    cur.execute(
+                        "UPDATE band_locations SET geocode_status='llm_needed', updated_at=now() WHERE id=%s",
+                        (loc_id,),
+                    )
+                    continue
+                # Autre erreur réseau/API → backoff
+                new_tries = tries_llm + 1
+                backoff   = 30 * new_tries
+                cur.execute(
+                    """
+                    UPDATE band_locations
+                    SET geocode_tries_llm=%s,
+                        geocode_error=%s,
+                        geocode_next_at=now() + (%s || ' minutes')::interval,
+                        updated_at=now()
+                    WHERE id=%s
+                    """,
+                    (new_tries, str(exc)[:300], backoff, loc_id),
+                )
+                print(f"[groq] id={loc_id} error: {exc}")
+
+            time.sleep(CALL_DELAY)
+
+    conn.close()
+
+
+def _handle_null_city(cur, loc_id: int, tries_llm: int):
+    new_tries = tries_llm + 1
+    if new_tries >= MAX_LLM_TRIES:
+        cur.execute(
+            """
+            UPDATE band_locations
+            SET geocode_status='manual',
+                geocode_error='llm: city=null après max tries',
+                geocode_tries_llm=%s,
+                updated_at=now()
+            WHERE id=%s
+            """,
+            (new_tries, loc_id),
+        )
+        print(f"[groq] id={loc_id} → manual (exhausted)")
+    else:
+        backoff = 60 * new_tries
+        cur.execute(
+            """
+            UPDATE band_locations
+            SET geocode_tries_llm=%s,
+                geocode_next_at=now() + (%s || ' minutes')::interval,
+                updated_at=now()
+            WHERE id=%s
+            """,
+            (new_tries, backoff, loc_id),
+        )
+
+
+if __name__ == "__main__":
+    main()

apps/geocoder/src/parser.py

@@ -0,0 +1,209 @@
+"""
+parser.py — Parse les location_text Metal Archives en steps structurés.
+
+Cas gérés :
+  "Oslo"                                     → 1 step
+  "Bergen (early); Oslo (later)"             → 2 steps
+  "Bergen / Oslo (early)"                    → 2 rows, même step_order
+  "Ski, Nordre Follo, Akershus (early)"      → 1 step, location_raw = texte complet
+  "N/A", "Unknown", "", "Various"            → liste vide
+  "FR", "France"                             → 1 step, is_country_only=True
+"""
+
+import re
+
+SKIP_VALUES = frozenset({
+    '', 'n/a', 'na', 'unknown', 'various', 'international',
+    'various locations', 'worldwide', '-', '—', 'tba', 'tbd',
+    'none', 'not available', 'disbanded', 'see comment', 'see above',
+    'multiple', 'many', 'different', 'different locations',
+})
+
+STANDALONE_COUNTRY_NAMES = frozenset({
+    'france', 'germany', 'italy', 'spain', 'sweden', 'norway', 'finland',
+    'poland', 'netherlands', 'belgium', 'switzerland', 'austria', 'portugal',
+    'greece', 'czech republic', 'slovakia', 'slovenia', 'hungary', 'ukraine',
+    'russia', 'denmark', 'united kingdom', 'england', 'scotland', 'ireland',
+    'northern ireland', 'wales', 'romania', 'bulgaria', 'serbia', 'croatia',
+    'bosnia', 'bosnia and herzegovina', 'albania', 'estonia', 'latvia',
+    'lithuania', 'iceland', 'luxembourg', 'malta', 'moldova', 'belarus',
+    'georgia', 'cyprus', 'turkey', 'north macedonia', 'macedonia',
+    'montenegro', 'kosovo', 'san marino', 'liechtenstein', 'monaco',
+    'andorra', 'vatican',
+})
+
+# ISO2 → (lat, lon) des capitales / centroïdes de référence
+COUNTRY_CENTROIDS = {
+    'AD': (42.5063,  1.5218), 'AL': (41.3317, 19.8172),
+    'AT': (48.2092, 16.3728), 'BA': (43.8519, 18.3866),
+    'BE': (50.8503,  4.3517), 'BG': (42.6977, 23.3219),
+    'BY': (53.9045, 27.5615), 'CH': (46.9480,  7.4474),
+    'CY': (35.1856, 33.3823), 'CZ': (50.0880, 14.4208),
+    'DE': (52.5200, 13.4050), 'DK': (55.6761, 12.5683),
+    'EE': (59.4370, 24.7536), 'ES': (40.4168, -3.7038),
+    'FI': (60.1699, 24.9384), 'FR': (48.8566,  2.3522),
+    'GB': (51.5074, -0.1278), 'GE': (41.6941, 44.8337),
+    'GR': (37.9838, 23.7275), 'HR': (45.8150, 15.9819),
+    'HU': (47.4979, 19.0402), 'IE': (53.3498, -6.2603),
+    'IS': (64.1355,-21.8954), 'IT': (41.9028, 12.4964),
+    'LI': (47.1416,  9.5215), 'LT': (54.6872, 25.2797),
+    'LU': (49.6116,  6.1319), 'LV': (56.9496, 24.1052),
+    'MC': (43.7333,  7.4000), 'MD': (47.0105, 28.8638),
+    'ME': (42.4304, 19.2594), 'MK': (42.0040, 21.4361),
+    'MT': (35.8997, 14.5147), 'NL': (52.3676,  4.9041),
+    'NO': (59.9139, 10.7522), 'PL': (52.2297, 21.0122),
+    'PT': (38.7167, -9.1333), 'RO': (44.4268, 26.1025),
+    'RS': (44.8176, 20.4633), 'RU': (55.7558, 37.6173),
+    'SE': (59.3293, 18.0686), 'SI': (46.0511, 14.5051),
+    'SK': (48.1486, 17.1077), 'SM': (43.9424, 12.4578),
+    'TR': (39.9334, 32.8597), 'UA': (50.4501, 30.5234),
+    'VA': (41.9029, 12.4534), 'XK': (42.6026, 20.9030),
+}
+
+COUNTRY_NAME_TO_ISO2 = {
+    'france': 'FR', 'germany': 'DE', 'italy': 'IT', 'spain': 'ES',
+    'sweden': 'SE', 'norway': 'NO', 'finland': 'FI', 'poland': 'PL',
+    'netherlands': 'NL', 'belgium': 'BE', 'switzerland': 'CH',
+    'austria': 'AT', 'portugal': 'PT', 'greece': 'GR',
+    'czech republic': 'CZ', 'slovakia': 'SK', 'slovenia': 'SI',
+    'hungary': 'HU', 'ukraine': 'UA', 'russia': 'RU', 'denmark': 'DK',
+    'united kingdom': 'GB', 'england': 'GB', 'scotland': 'GB',
+    'northern ireland': 'GB', 'wales': 'GB', 'ireland': 'IE',
+    'romania': 'RO', 'bulgaria': 'BG', 'serbia': 'RS', 'croatia': 'HR',
+    'bosnia': 'BA', 'bosnia and herzegovina': 'BA', 'albania': 'AL',
+    'estonia': 'EE', 'latvia': 'LV', 'lithuania': 'LT', 'iceland': 'IS',
+    'luxembourg': 'LU', 'malta': 'MT', 'moldova': 'MD', 'belarus': 'BY',
+    'georgia': 'GE', 'cyprus': 'CY', 'turkey': 'TR',
+    'north macedonia': 'MK', 'macedonia': 'MK', 'montenegro': 'ME',
+    'kosovo': 'XK', 'san marino': 'SM', 'liechtenstein': 'LI',
+    'monaco': 'MC', 'andorra': 'AD', 'vatican': 'VA',
+}
+
+COUNTRY_NAMES = {v: k.title() for k, v in COUNTRY_NAME_TO_ISO2.items() if len(v) == 2}
+# Fix doublons (en → GB wins "United Kingdom")
+COUNTRY_NAMES.update({
+    'GB': 'United Kingdom', 'CZ': 'Czech Republic', 'BA': 'Bosnia and Herzegovina',
+    'MK': 'North Macedonia', 'XK': 'Kosovo',
+})
+
+_ISO2_RE     = re.compile(r'^[A-Z]{2}$')
+_CITY_SPLIT  = re.compile(r'\s*/\s*|\s*\\\s*|\s+and\s+|\s*&\s*', re.IGNORECASE)
+_STEP_RE     = re.compile(r'([^;(]+?)(?:\s*\(([^)]+?)\))?\s*(?:;|$)')
+
+
+def _classify(text: str) -> str:
+    t = text.strip().lower().rstrip(';').strip()
+    if t in SKIP_VALUES:
+        return 'skip'
+    if _ISO2_RE.match(text.strip()):
+        return 'country_code'
+    if t in STANDALONE_COUNTRY_NAMES:
+        return 'country_name'
+    return 'parseable'
+
+
+def parse_location_text(location_text: str, band_country: str | None = None) -> list[dict]:
+    """
+    Retourne une liste de dicts, un par ligne band_locations :
+      step_order     int
+      step_label     str | None
+      location_raw   str
+      is_country_only bool
+    """
+    if not location_text:
+        return []
+
+    kind = _classify(location_text)
+    if kind == 'skip':
+        return []
+    if kind == 'country_code':
+        return [{'step_order': 0, 'step_label': None,
+                 'location_raw': location_text.strip().upper(), 'is_country_only': True}]
+    if kind == 'country_name':
+        return [{'step_order': 0, 'step_label': None,
+                 'location_raw': location_text.strip(), 'is_country_only': True}]
+
+    rows = []
+    step_order = 0
+
+    for m in _STEP_RE.finditer(location_text.strip()):
+        raw   = m.group(1).strip().rstrip(';').strip()
+        label = (m.group(2) or '').strip() or None
+
+        if not raw or _classify(raw) == 'skip':
+            step_order += 1
+            continue
+
+        cities    = [c.strip() for c in _CITY_SPLIT.split(raw) if c.strip()]
+        any_added = False
+
+        for city in cities:
+            ck = _classify(city)
+            if ck == 'skip':
+                continue
+            rows.append({
+                'step_order':    step_order,
+                'step_label':    label,
+                'location_raw':  city,
+                'is_country_only': ck in ('country_code', 'country_name'),
+            })
+            any_added = True
+
+        if any_added:
+            step_order += 1
+
+    return rows
+
+
+def country_centroid(iso2: str) -> tuple[float, float] | None:
+    code = iso2.strip().upper()
+    if code in COUNTRY_CENTROIDS:
+        return COUNTRY_CENTROIDS[code]
+    name_iso = COUNTRY_NAME_TO_ISO2.get(iso2.strip().lower())
+    if name_iso:
+        return COUNTRY_CENTROIDS.get(name_iso)
+    return None
+
+
+def build_fallback_queries(location_raw: str, country_code: str | None) -> list[str]:
+    """
+    Construit une liste ordonnée de requêtes Geoapify (plus spécifique → plus vague).
+    Le pays est injecté comme contexte à chaque niveau.
+    """
+    country_name = COUNTRY_NAMES.get((country_code or '').upper(), '')
+    parts = [p.strip() for p in location_raw.split(',')]
+
+    seen: set[str]   = set()
+    result: list[str] = []
+
+    def add(q: str):
+        q = q.strip()
+        if q and q not in seen:
+            seen.add(q)
+            result.append(q)
+
+    has_country = bool(country_name) and (
+        country_name.lower() in location_raw.lower()
+        or (country_code and country_code.upper() in location_raw.upper())
+    )
+
+    # Requête complète + pays (si pays pas déjà dans le texte)
+    if not has_country and country_name:
+        add(f"{location_raw}, {country_name}")
+    add(location_raw)
+
+    # Retire les divisions administratives une par une
+    for n in range(len(parts) - 1, 0, -1):
+        prefix = ', '.join(parts[:n])
+        if not has_country and country_name:
+            add(f"{prefix}, {country_name}")
+        add(prefix)
+
+    # Juste la ville + pays
+    if parts:
+        city = parts[0]
+        if country_name:
+            add(f"{city}, {country_name}")
+        add(city)
+
+    return result

apps/geocoder/src/worker.py

@@ -1,3 +1,14 @@
+"""
+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
@@ -5,30 +16,36 @@ 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"))   # 5 req/s free tier
+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 = 0.0
+_last_call = 0.0
 
-def polite_sleep():
+
-    global _last
+def _polite_sleep():
-    elapsed = time.monotonic() - _last
+    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 = time.monotonic()
+    _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()
+    _polite_sleep()
     r = requests.get(GEOAPIFY_BASE, params=params, timeout=30)
     if r.status_code == 429:
-        raise RuntimeError(f"geoapify_throttle status=429")
+        raise RuntimeError("geoapify_throttle:429")
     r.raise_for_status()
     data     = r.json()
     features = data.get("features") or []
@@ -38,12 +55,48 @@ def geoapify_search(query: str) -> dict | None:
     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)
@@ -54,127 +107,169 @@ def main():
         while processed < MAX_PER_RUN:
             cur.execute(
                 """
-                SELECT ma_id, query, country, tries
+                SELECT bl.id, bl.ma_id, bl.location_raw,
-                FROM geocode_queue
+                       bl.geocode_tries_geo, bl.geocode_query, b.country
-                WHERE status IN ('queued','error')
+                FROM band_locations bl
-                  AND next_run_at <= now()
+                JOIN bands b ON b.ma_id = bl.ma_id
-                ORDER BY next_run_at ASC, ma_id ASC
+                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 SKIP LOCKED
+                FOR UPDATE OF bl SKIP LOCKED
                 """
             )
             row = cur.fetchone()
             if not row:
-                print("[worker] nothing to do. sleeping 60s")
+                print("[worker] rien à traiter, attente 60s")
                 time.sleep(60)
                 continue
 
-            ma_id, query, country, tries = row
+            loc_id, ma_id, location_raw, tries, llm_query, country = row
 
             cur.execute(
-                "UPDATE geocode_queue SET status='processing', updated_at=now() WHERE ma_id=%s",
+                "UPDATE band_locations SET geocode_status='processing', updated_at=now() WHERE id=%s",
-                (ma_id,),
+                (loc_id,),
             )
 
-            # cache hit ?
+            # Liste de requêtes à essayer : requête LLM en premier si disponible
-            cur.execute("SELECT lat, lon, raw FROM geocode_cache WHERE query=%s", (query,))
+            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 and cached[1] is not None:
+                if cached and cached[0] is not None:
-                lat, lon, raw = cached
+                    lat, lon   = cached
-                cur.execute(
+                    used_query = query
-                    """
+                    provider   = 'geoapify-cache'
-                    UPDATE bands
+                    success    = True
-                    SET lat=%s, lon=%s,
+                    print(f"[worker] id={loc_id} cache HIT '{query}'")
-                        geom=ST_SetSRID(ST_MakePoint(%s,%s),4326)::geography,
+                    break
-                        geocoded_at=now(),
-                        geocode_provider='geoapify-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
 
-            # requête Geoapify
                 try:
                     res = geoapify_search(query)
-                if not res:
+                    if 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"])
-
+                        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)
+                            INSERT INTO geocode_cache
-                    VALUES (%s,'geoapify',%s,%s,ST_SetSRID(ST_MakePoint(%s,%s),4326)::geography,%s,now())
+                              (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,
+                              SET lat=EXCLUDED.lat, lon=EXCLUDED.lon,
-                          raw=EXCLUDED.raw, provider=EXCLUDED.provider, updated_at=now()
+                                  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 bands
+                            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,
-                        geom=ST_SetSRID(ST_MakePoint(%s,%s),4326)::geography,
+                        geocode_status='done',
-                        geocoded_at=now(),
-                        geocode_provider='geoapify',
                         geocode_query=%s,
-                        geocode_raw=%s,
+                        geocode_provider=%s,
+                        geocode_confidence=%s,
                         geocode_error=NULL,
-                        geocode_error_at=NULL
+                        updated_at=now()
-                    WHERE ma_id=%s
+                    WHERE id=%s
                     """,
-                    (lat, lon, lon, lat, query, json.dumps(res["raw"]), ma_id),
+                    (lat, lon, used_query, provider, confidence, loc_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_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),
                 )
+                sync_bands_primary(cur, ma_id)
+            else:
+                new_tries = tries + 1
+                if new_tries >= MAX_GEO_TRIES:
                     cur.execute(
                         """
-                    UPDATE geocode_queue
+                        UPDATE band_locations
-                    SET status='error', tries=tries+1, last_error=%s,
+                        SET geocode_status='llm_needed',
-                        next_run_at=now() + (%s || ' minutes')::interval, updated_at=now()
+                            geocode_tries_geo=%s,
-                    WHERE ma_id=%s
+                            geocode_error=%s,
+                            updated_at=now()
+                        WHERE id=%s
                         """,
-                    (str(e)[:400], backoff_minutes, ma_id),
+                        (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
-                print(f"[worker] ma_id={ma_id} ERROR {e} -> retry in {backoff_minutes}m")
 
     conn.close()
+    print(f"[worker] terminé, processed={processed}")
+
 
 if __name__ == "__main__":
     main()

docker-compose.dev.yml

@@ -35,6 +35,21 @@ services:
       - traefik.enable=false
     restart: unless-stopped
 
+  geocoder-enqueue:
+    build:
+      context: apps/geocoder
+    working_dir: /app
+    environment:
+      DATABASE_URL: ${DATABASE_URL}
+      GEOCODE_ENQUEUE_BATCH: "100000"
+      GEOCODE_ENQUEUE_POLL: "15"
+    command: ["python", "src/enqueue.py"]
+    networks:
+      - coolify
+    labels:
+      - traefik.enable=false
+    restart: unless-stopped
+
   geocoder-worker:
     build:
       context: apps/geocoder
@@ -44,13 +59,30 @@ services:
       GEOAPIFY_API_KEY: ${GEOAPIFY_API_KEY}
       GEOCODER_MIN_DELAY: "0.22"
       GEOCODER_JITTER: "0.10"
-      GEOCODE_MAX_PER_RUN: "5000"
+      GEOCODE_MAX_PER_RUN: "100000"
+      GEOCODE_MAX_GEO_TRIES: "3"
     command: ["python", "src/worker.py"]
     networks:
       - coolify
     labels:
       - traefik.enable=false
-    restart: "no"
+    restart: unless-stopped
+
+  groq-worker:
+    build:
+      context: apps/geocoder
+    working_dir: /app
+    environment:
+      DATABASE_URL: ${DATABASE_URL}
+      GROQ_API_KEY: ${GROQ_API_KEY}
+      GROQ_MAX_TRIES: "5"
+      GROQ_CALL_DELAY: "2.5"
+    command: ["python", "src/groq_worker.py"]
+    networks:
+      - coolify
+    labels:
+      - traefik.enable=false
+    restart: unless-stopped
 
   pgadmin:
     image: dpage/pgadmin4:8

docker-compose.yml

@@ -7,6 +7,21 @@ services:
     networks:
       - coolify
 
+  geocoder-enqueue:
+    build:
+      context: apps/geocoder
+    working_dir: /app
+    environment:
+      DATABASE_URL: ${DATABASE_URL}
+      GEOCODE_ENQUEUE_BATCH: "100000"
+      GEOCODE_ENQUEUE_POLL: "15"
+    command: ["python", "src/enqueue.py"]
+    networks:
+      - coolify
+    labels:
+      - traefik.enable=false
+    restart: unless-stopped
+
   geocoder-worker:
     build:
       context: apps/geocoder
@@ -16,7 +31,8 @@ services:
       GEOAPIFY_API_KEY: ${GEOAPIFY_API_KEY}
       GEOCODER_MIN_DELAY: "0.22"
       GEOCODER_JITTER: "0.10"
-      GEOCODE_MAX_PER_RUN: "5000"
+      GEOCODE_MAX_PER_RUN: "100000"
+      GEOCODE_MAX_GEO_TRIES: "3"
     command: ["python", "src/worker.py"]
     networks:
       - coolify
@@ -24,6 +40,22 @@ services:
       - traefik.enable=false
     restart: unless-stopped
 
+  groq-worker:
+    build:
+      context: apps/geocoder
+    working_dir: /app
+    environment:
+      DATABASE_URL: ${DATABASE_URL}
+      GROQ_API_KEY: ${GROQ_API_KEY}
+      GROQ_MAX_TRIES: "5"
+      GROQ_CALL_DELAY: "2.5"
+    command: ["python", "src/groq_worker.py"]
+    networks:
+      - coolify
+    labels:
+      - traefik.enable=false
+    restart: unless-stopped
+
   pgadmin:
     image: dpage/pgadmin4:8
     environment: