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>
This commit is contained in:
parent
c39a8513f8
commit
dba6b9f730
5 changed files with 173 additions and 90 deletions
|
|
@ -928,6 +928,7 @@ async function renderGeocoding() {
|
|||
<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>
|
||||
|
|
@ -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() {
|
|||
|
||||
<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-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-enqueue" style="border-color:rgba(63,174,90,0.5);color:var(--ok)">▶ Lancer l'enqueue band_locations</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>
|
||||
|
|
@ -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");
|
||||
|
|
|
|||
|
|
@ -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(", ")}` });
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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,27 +32,19 @@ 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"))
|
||||
|
||||
conn = psycopg2.connect(dsn)
|
||||
conn.autocommit = True
|
||||
|
||||
inserted = 0
|
||||
country_fixed = 0
|
||||
skipped = 0
|
||||
|
||||
with conn.cursor() as cur:
|
||||
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,),
|
||||
(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:
|
||||
|
|
@ -92,9 +79,7 @@ def main():
|
|||
AND bl.is_country_only = TRUE
|
||||
"""
|
||||
)
|
||||
country_rows = cur.fetchall()
|
||||
|
||||
for loc_id, location_raw, band_country in country_rows:
|
||||
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:
|
||||
|
|
@ -116,10 +101,72 @@ def main():
|
|||
else:
|
||||
print(f"[enqueue] no centroid id={loc_id} raw='{location_raw}'")
|
||||
|
||||
print(
|
||||
f"[enqueue] done — inserted={inserted} "
|
||||
f"country_fixed={country_fixed} skipped={skipped}"
|
||||
return inserted, country_fixed, skipped
|
||||
|
||||
|
||||
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 = 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()
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
Loading…
Reference in a new issue