diff --git a/apps/admin/site/app.js b/apps/admin/site/app.js index 654a909..646682f 100644 --- a/apps/admin/site/app.js +++ b/apps/admin/site/app.js @@ -928,6 +928,7 @@ async function renderGeocoding() {

Nouveau pipeline — band_locations

+ @@ -982,6 +983,22 @@ async function renderGeocoding() { }); }; + 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) ?"); @@ -1127,10 +1144,11 @@ async function renderJobs() {

🗺 Géocodeur

-

Le géocodeur tourne en continu — ces actions modifient la queue directement.

+

Le géocodeur et le worker Groq tournent en continu. L'enqueue parse les bands en band_locations.

- - + + +
@@ -1177,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"); diff --git a/apps/api/src/adminRoutes.js b/apps/api/src/adminRoutes.js index 075390c..50f61b6 100644 --- a/apps/api/src/adminRoutes.js +++ b/apps/api/src/adminRoutes.js @@ -487,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(", ")}` }); } diff --git a/apps/geocoder/src/enqueue.py b/apps/geocoder/src/enqueue.py index fca5ad3..37ea0c7 100644 --- a/apps/geocoder/src/enqueue.py +++ b/apps/geocoder/src/enqueue.py @@ -1,29 +1,24 @@ """ -enqueue.py — Peuple band_locations depuis bands.location_text. +enqueue.py — Daemon qui attend des job_triggers de type 'geocoder_enqueue' +et peuple band_locations depuis bands.location_text. -Usage : - python src/enqueue.py - -Actions : - 1. Parse location_text de tous les bands via parser.parse_location_text - 2. INSERT INTO band_locations (ON CONFLICT DO NOTHING → idempotent) - 3. Résout immédiatement les is_country_only avec centroïdes hardcodés - 4. Continue à alimenter geocode_queue (ancien pipeline) pour les bands sans - location_raw géocodable (backward compat carte existante) +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 -import sys -# Quand lancé comme "python src/enqueue.py", sys.path[0] = src/ from parser import ( parse_location_text, country_centroid, COUNTRY_NAME_TO_ISO2, - COUNTRY_NAMES, ) +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() @@ -37,89 +32,141 @@ def resolve_iso2(location_raw: str, band_country: str | None) -> str | None: return None -def main(): - dsn = os.environ["DATABASE_URL"] - batch = int(os.environ.get("GEOCODE_ENQUEUE_BATCH", "50000")) +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 main(): + dsn = os.environ["DATABASE_URL"] conn = psycopg2.connect(dsn) conn.autocommit = True - inserted = 0 - country_fixed = 0 - skipped = 0 + print(f"[enqueue] daemon démarré, poll toutes les {POLL_INTERVAL}s") with conn.cursor() as cur: - 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") + 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() - for ma_id, country, location_text in bands: - steps = parse_location_text(location_text, country) - if not steps: - skipped += 1 + if not row: + time.sleep(POLL_INTERVAL) 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}") + 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é") - # 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 - """ - ) - country_rows = cur.fetchall() - - for loc_id, location_raw, band_country in country_rows: - 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), + 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}" ) - country_fixed += 1 - else: - print(f"[enqueue] no centroid id={loc_id} raw='{location_raw}'") + 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}") - print( - f"[enqueue] done — inserted={inserted} " - f"country_fixed={country_fixed} skipped={skipped}" - ) conn.close() diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 9b692e0..024e9f0 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -42,12 +42,13 @@ services: 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: "no" + restart: unless-stopped geocoder-worker: build: diff --git a/docker-compose.yml b/docker-compose.yml index a03d40e..5767230 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -14,12 +14,13 @@ services: 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: "no" + restart: unless-stopped geocoder-worker: build: