Compare commits
2 commits
3d973c430b
...
dba6b9f730
| Author | SHA1 | Date | |
|---|---|---|---|
| dba6b9f730 | |||
| c39a8513f8 |
9 changed files with 1230 additions and 236 deletions
|
|
@ -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 style="display:flex;align-items:center;justify-content:space-between;flex-wrap:wrap;gap:8px;margin-bottom:12px">
|
||||
<div style="font-size:13px;font-weight:700;color:var(--bone)">Pipeline géocodage</div>
|
||||
<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:12px">
|
||||
<h2 style="margin:0">Progression</h2>
|
||||
<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 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>
|
||||
<div style="background:rgba(255,255,255,0.06);border-radius:8px;height:18px;overflow:hidden;margin-bottom:8px">
|
||||
<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,30 +1041,63 @@ 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");
|
||||
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;
|
||||
const errored = r.queue.find((x) => x.status === "error")?.n || 0;
|
||||
|
||||
// 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;
|
||||
const errored = r.queue.find((x) => x.status === "error")?.n || 0;
|
||||
const processing = r.queue.find((x) => x.status === "processing")?.n || 0;
|
||||
const pct = total ? Math.round((done / total) * 100) : 0;
|
||||
const pct = total ? Math.round((done / total) * 100) : 0;
|
||||
|
||||
const statsEl = document.getElementById("geo-stats");
|
||||
if (statsEl) statsEl.innerHTML = `
|
||||
${statCard("Total queue", total)}
|
||||
${statCard("Géocodés ✓", done, "ok")}
|
||||
${statCard("En attente", queued, "warn")}
|
||||
${statCard("En cours", processing || 0, processing ? "ok" : "")}
|
||||
${statCard("Géocodés", done, "ok")}
|
||||
${statCard("En attente", queued, queued ? "warn" : "")}
|
||||
${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-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>
|
||||
|
|
@ -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");
|
||||
|
|
|
|||
54
apps/api/migrations/008_band_locations.sql
Normal file
54
apps/api/migrations/008_band_locations.sql
Normal file
|
|
@ -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()
|
||||
);
|
||||
|
|
@ -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)
|
||||
// ------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -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 = {
|
||||
"FR": "France",
|
||||
"DE": "Germany",
|
||||
"IT": "Italy",
|
||||
"GB": "United Kingdom",
|
||||
"ES": "Spain",
|
||||
"SE": "Sweden",
|
||||
"NO": "Norway",
|
||||
"FI": "Finland",
|
||||
"PL": "Poland",
|
||||
"NL": "Netherlands",
|
||||
"BE": "Belgium",
|
||||
"CH": "Switzerland",
|
||||
"AT": "Austria",
|
||||
"PT": "Portugal",
|
||||
"GR": "Greece",
|
||||
"CZ": "Czech Republic",
|
||||
"SK": "Slovakia",
|
||||
"SI": "Slovenia",
|
||||
"HU": "Hungary",
|
||||
"UA": "Ukraine",
|
||||
"RU": "Russia",
|
||||
"DK": "Denmark",
|
||||
}
|
||||
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 = 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")
|
||||
|
||||
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:
|
||||
# On sélectionne :
|
||||
# - bands non géocodés avec location_text
|
||||
# - 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
|
||||
b.ma_id, b.country, b.location_text
|
||||
FROM bands b
|
||||
LEFT JOIN geocode_queue q ON q.ma_id = b.ma_id
|
||||
WHERE b.geom IS NULL
|
||||
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)
|
||||
|
||||
while True:
|
||||
# Chercher un job en attente
|
||||
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),
|
||||
SELECT id FROM job_triggers
|
||||
WHERE job_type = 'geocoder_enqueue' AND status = 'pending'
|
||||
ORDER BY created_at ASC
|
||||
LIMIT 1
|
||||
FOR UPDATE SKIP LOCKED
|
||||
"""
|
||||
)
|
||||
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()
|
||||
|
|
|
|||
315
apps/geocoder/src/groq_worker.py
Normal file
315
apps/geocoder/src/groq_worker.py
Normal file
|
|
@ -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()
|
||||
209
apps/geocoder/src/parser.py
Normal file
209
apps/geocoder/src/parser.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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,47 +16,89 @@ 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
|
||||
JITTER = float(os.environ.get("GEOCODER_JITTER", "0.10"))
|
||||
MAX_PER_RUN = int(os.environ.get("GEOCODE_MAX_PER_RUN", "100000"))
|
||||
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_call
|
||||
elapsed = time.monotonic() - _last_call
|
||||
wait = max(0.0, MIN_DELAY - elapsed) + random.uniform(0.0, JITTER)
|
||||
if wait > 0:
|
||||
time.sleep(wait)
|
||||
_last_call = time.monotonic()
|
||||
|
||||
def polite_sleep():
|
||||
global _last
|
||||
elapsed = time.monotonic() - _last
|
||||
wait = max(0.0, MIN_DELAY - elapsed) + random.uniform(0.0, JITTER)
|
||||
time.sleep(wait)
|
||||
_last = 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()
|
||||
data = r.json()
|
||||
features = data.get("features") or []
|
||||
if not features:
|
||||
return None
|
||||
props = features[0].get("properties", {})
|
||||
props = features[0].get("properties", {})
|
||||
coords = features[0].get("geometry", {}).get("coordinates", [])
|
||||
if len(coords) < 2:
|
||||
return None
|
||||
rank = props.get("rank") or {}
|
||||
confidence = rank.get("confidence", 0.5) if isinstance(rank, dict) else 0.5
|
||||
return {
|
||||
"lat": props.get("lat") or coords[1],
|
||||
"lon": props.get("lon") or coords[0],
|
||||
"raw": data,
|
||||
"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"]
|
||||
dsn = os.environ["DATABASE_URL"]
|
||||
conn = psycopg2.connect(dsn)
|
||||
conn.autocommit = True
|
||||
|
||||
|
|
@ -54,127 +107,169 @@ def main():
|
|||
while processed < MAX_PER_RUN:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT ma_id, query, country, tries
|
||||
FROM geocode_queue
|
||||
WHERE status IN ('queued','error')
|
||||
AND next_run_at <= now()
|
||||
ORDER BY next_run_at ASC, ma_id ASC
|
||||
SELECT bl.id, bl.ma_id, bl.location_raw,
|
||||
bl.geocode_tries_geo, bl.geocode_query, b.country
|
||||
FROM band_locations bl
|
||||
JOIN bands b ON b.ma_id = bl.ma_id
|
||||
WHERE bl.geocode_status = 'queued'
|
||||
AND bl.geocode_next_at <= now()
|
||||
ORDER BY bl.geocode_next_at ASC, bl.id ASC
|
||||
LIMIT 1
|
||||
FOR UPDATE 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",
|
||||
(ma_id,),
|
||||
"UPDATE band_locations SET geocode_status='processing', updated_at=now() WHERE id=%s",
|
||||
(loc_id,),
|
||||
)
|
||||
|
||||
# cache hit ?
|
||||
cur.execute("SELECT lat, lon, raw FROM geocode_cache WHERE query=%s", (query,))
|
||||
cached = cur.fetchone()
|
||||
if cached and cached[0] is not None and cached[1] is not None:
|
||||
lat, lon, raw = cached
|
||||
# Liste de requêtes à essayer : requête LLM en premier si disponible
|
||||
fallbacks = build_fallback_queries(location_raw, country)
|
||||
if llm_query and llm_query not in fallbacks:
|
||||
queries = [llm_query] + fallbacks
|
||||
else:
|
||||
queries = fallbacks
|
||||
|
||||
# Déduplique en gardant l'ordre
|
||||
seen: set[str] = set()
|
||||
unique: list[str] = []
|
||||
for q in queries:
|
||||
if q not in seen:
|
||||
seen.add(q)
|
||||
unique.append(q)
|
||||
|
||||
success = False
|
||||
rate_limit = False
|
||||
used_query = None
|
||||
lat = lon = None
|
||||
confidence = 0.5
|
||||
provider = 'geoapify'
|
||||
last_err = None
|
||||
|
||||
for query in unique:
|
||||
# Cache hit ?
|
||||
cur.execute(
|
||||
"""
|
||||
UPDATE bands
|
||||
SET lat=%s, lon=%s,
|
||||
geom=ST_SetSRID(ST_MakePoint(%s,%s),4326)::geography,
|
||||
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),
|
||||
"SELECT lat, lon FROM geocode_cache WHERE query=%s",
|
||||
(query,),
|
||||
)
|
||||
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")
|
||||
cached = cur.fetchone()
|
||||
if cached and cached[0] is not None:
|
||||
lat, lon = cached
|
||||
used_query = query
|
||||
provider = 'geoapify-cache'
|
||||
success = True
|
||||
print(f"[worker] id={loc_id} cache HIT '{query}'")
|
||||
break
|
||||
|
||||
try:
|
||||
res = geoapify_search(query)
|
||||
if res:
|
||||
lat = float(res["lat"])
|
||||
lon = float(res["lon"])
|
||||
confidence = float(res.get("confidence") or 0.5)
|
||||
used_query = query
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO geocode_cache
|
||||
(query, provider, lat, lon, geom, raw, updated_at)
|
||||
VALUES (%s,'geoapify',%s,%s,
|
||||
ST_SetSRID(ST_MakePoint(%s,%s),4326)::geography,
|
||||
%s, now())
|
||||
ON CONFLICT (query) DO UPDATE
|
||||
SET lat=EXCLUDED.lat, lon=EXCLUDED.lon,
|
||||
geom=EXCLUDED.geom, raw=EXCLUDED.raw,
|
||||
provider=EXCLUDED.provider, updated_at=now()
|
||||
""",
|
||||
(query, lat, lon, lon, lat, json.dumps(res["raw"])),
|
||||
)
|
||||
success = True
|
||||
print(f"[worker] id={loc_id} OK '{query}' lat={lat:.4f} lon={lon:.4f}")
|
||||
break
|
||||
else:
|
||||
last_err = f"no_result:'{query}'"
|
||||
except RuntimeError as exc:
|
||||
if "429" in str(exc):
|
||||
print("[worker] rate limit Geoapify, attente 60s")
|
||||
cur.execute(
|
||||
"""
|
||||
UPDATE band_locations
|
||||
SET geocode_status='queued',
|
||||
geocode_next_at=now() + interval '60 seconds',
|
||||
updated_at=now()
|
||||
WHERE id=%s
|
||||
""",
|
||||
(loc_id,),
|
||||
)
|
||||
rate_limit = True
|
||||
time.sleep(60)
|
||||
break
|
||||
last_err = str(exc)[:300]
|
||||
except Exception as exc:
|
||||
last_err = str(exc)[:300]
|
||||
|
||||
if rate_limit:
|
||||
continue
|
||||
|
||||
# requête Geoapify
|
||||
try:
|
||||
res = geoapify_search(query)
|
||||
if not res:
|
||||
cur.execute(
|
||||
"UPDATE bands SET geocode_error=%s, geocode_error_at=now() WHERE ma_id=%s",
|
||||
("no_result", ma_id),
|
||||
)
|
||||
if success:
|
||||
cur.execute(
|
||||
"""
|
||||
UPDATE band_locations
|
||||
SET lat=%s, lon=%s,
|
||||
geocode_status='done',
|
||||
geocode_query=%s,
|
||||
geocode_provider=%s,
|
||||
geocode_confidence=%s,
|
||||
geocode_error=NULL,
|
||||
updated_at=now()
|
||||
WHERE id=%s
|
||||
""",
|
||||
(lat, lon, used_query, provider, confidence, loc_id),
|
||||
)
|
||||
sync_bands_primary(cur, ma_id)
|
||||
else:
|
||||
new_tries = tries + 1
|
||||
if new_tries >= MAX_GEO_TRIES:
|
||||
cur.execute(
|
||||
"""
|
||||
UPDATE 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
|
||||
UPDATE band_locations
|
||||
SET geocode_status='llm_needed',
|
||||
geocode_tries_geo=%s,
|
||||
geocode_error=%s,
|
||||
updated_at=now()
|
||||
WHERE id=%s
|
||||
""",
|
||||
("no_result", ma_id),
|
||||
(new_tries, last_err or "all_fallbacks_failed", loc_id),
|
||||
)
|
||||
processed += 1
|
||||
print(f"[worker] ma_id={ma_id} no_result -> postpone")
|
||||
continue
|
||||
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}")
|
||||
|
||||
lat = float(res["lat"])
|
||||
lon = float(res["lon"])
|
||||
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO geocode_cache(query, provider, lat, lon, geom, raw, updated_at)
|
||||
VALUES (%s,'geoapify',%s,%s,ST_SetSRID(ST_MakePoint(%s,%s),4326)::geography,%s,now())
|
||||
ON CONFLICT (query) DO UPDATE
|
||||
SET lat=EXCLUDED.lat, lon=EXCLUDED.lon, geom=EXCLUDED.geom,
|
||||
raw=EXCLUDED.raw, provider=EXCLUDED.provider, updated_at=now()
|
||||
""",
|
||||
(query, lat, lon, lon, lat, json.dumps(res["raw"])),
|
||||
)
|
||||
|
||||
cur.execute(
|
||||
"""
|
||||
UPDATE bands
|
||||
SET lat=%s, lon=%s,
|
||||
geom=ST_SetSRID(ST_MakePoint(%s,%s),4326)::geography,
|
||||
geocoded_at=now(),
|
||||
geocode_provider='geoapify',
|
||||
geocode_query=%s,
|
||||
geocode_raw=%s,
|
||||
geocode_error=NULL,
|
||||
geocode_error_at=NULL
|
||||
WHERE ma_id=%s
|
||||
""",
|
||||
(lat, lon, lon, lat, query, json.dumps(res["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} 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),
|
||||
)
|
||||
cur.execute(
|
||||
"""
|
||||
UPDATE geocode_queue
|
||||
SET status='error', tries=tries+1, last_error=%s,
|
||||
next_run_at=now() + (%s || ' minutes')::interval, updated_at=now()
|
||||
WHERE ma_id=%s
|
||||
""",
|
||||
(str(e)[:400], backoff_minutes, ma_id),
|
||||
)
|
||||
processed += 1
|
||||
print(f"[worker] ma_id={ma_id} ERROR {e} -> retry in {backoff_minutes}m")
|
||||
processed += 1
|
||||
|
||||
conn.close()
|
||||
print(f"[worker] terminé, processed={processed}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
Loading…
Reference in a new issue