-
-
-
Auto-refresh 30s
+
+
Nouveau pipeline — band_locations
+
+
+
+
-
+
+
+
+
Ancien pipeline — geocode_queue
+
+
+
+
+
+
+
—
+
-
20 derniers géocodages
+
20 derniers géocodages (bands)
`;
+ // 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; }
+ });
+ };
+
+ 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 +994,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 +1003,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 +1024,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");
diff --git a/apps/api/migrations/008_band_locations.sql b/apps/api/migrations/008_band_locations.sql
new file mode 100644
index 0000000..8de89e4
--- /dev/null
+++ b/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()
+);
diff --git a/apps/api/src/adminRoutes.js b/apps/api/src/adminRoutes.js
index ab38a7d..075390c 100644
--- a/apps/api/src/adminRoutes.js
+++ b/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);
@@ -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)
// ------------------------------------------------------------------
diff --git a/apps/geocoder/src/enqueue.py b/apps/geocoder/src/enqueue.py
index 18fcef4..fca5ad3 100644
--- a/apps/geocoder/src/enqueue.py
+++ b/apps/geocoder/src/enqueue.py
@@ -1,109 +1,127 @@
+"""
+enqueue.py — 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)
+"""
+
import os
import psycopg2
+import sys
-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",
-}
+# 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,
+)
+
+
+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 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"]
+ 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:
- # 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 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")
+
+ 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
- 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),
+ 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()
- 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
-
- print(f"[enqueue] queued_or_updated={enq} (selected={len(rows)}) force_retry={force_retry} retry_after_hours={retry_after_hours}")
+ 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),
+ )
+ country_fixed += 1
+ 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}"
+ )
conn.close()
+
if __name__ == "__main__":
main()
diff --git a/apps/geocoder/src/groq_worker.py b/apps/geocoder/src/groq_worker.py
new file mode 100644
index 0000000..02d3ca0
--- /dev/null
+++ b/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()
diff --git a/apps/geocoder/src/parser.py b/apps/geocoder/src/parser.py
new file mode 100644
index 0000000..d87a2a1
--- /dev/null
+++ b/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
diff --git a/apps/geocoder/src/worker.py b/apps/geocoder/src/worker.py
index 9af8a04..a71bf4a 100644
--- a/apps/geocoder/src/worker.py
+++ b/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,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()
diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml
index 7ebf760..9b692e0 100644
--- a/docker-compose.dev.yml
+++ b/docker-compose.dev.yml
@@ -35,6 +35,20 @@ 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"
+ command: ["python", "src/enqueue.py"]
+ networks:
+ - coolify
+ labels:
+ - traefik.enable=false
+ restart: "no"
+
geocoder-worker:
build:
context: apps/geocoder
@@ -44,13 +58,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
diff --git a/docker-compose.yml b/docker-compose.yml
index 236ea70..a03d40e 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -7,6 +7,20 @@ services:
networks:
- coolify
+ geocoder-enqueue:
+ build:
+ context: apps/geocoder
+ working_dir: /app
+ environment:
+ DATABASE_URL: ${DATABASE_URL}
+ GEOCODE_ENQUEUE_BATCH: "100000"
+ command: ["python", "src/enqueue.py"]
+ networks:
+ - coolify
+ labels:
+ - traefik.enable=false
+ restart: "no"
+
geocoder-worker:
build:
context: apps/geocoder
@@ -16,7 +30,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 +39,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: