diff --git a/apps/admin/site/app.js b/apps/admin/site/app.js index 0965d4b..4a4f928 100644 --- a/apps/admin/site/app.js +++ b/apps/admin/site/app.js @@ -3,11 +3,12 @@ const state = { username: null, view: "dashboard", - bands: { page: 1, pageSize: 50, q: "", country: "", status: "", genre: "", enriched: "", sort: "ma_id", dir: "asc", total: 0 }, + bands: { page: 1, pageSize: 50, q: "", location_q: "", country: "", status: "", genre: "", enriched: "", has_lat: "", has_location: "", sort: "ma_id", dir: "asc", total: 0 }, runs: { page: 1, pageSize: 50, run_type: "", status: "", total: 0 }, logs: { page: 1, pageSize: 100, level: "", autoRefresh: true }, audit: { page: 1, pageSize: 50 }, logsTimer: null, + queueTimer: null, }; // ------------------------------------------------------------------ @@ -37,6 +38,7 @@ function showLogin() { document.getElementById("login-screen").classList.remove("hidden"); document.getElementById("app").classList.add("hidden"); if (state.logsTimer) { clearInterval(state.logsTimer); state.logsTimer = null; } + if (state.queueTimer) { clearInterval(state.queueTimer); state.queueTimer = null; } } function showApp() { @@ -102,6 +104,7 @@ function router() { a.classList.toggle("active", a.dataset.view === view); }); if (state.logsTimer) { clearInterval(state.logsTimer); state.logsTimer = null; } + if (state.queueTimer) { clearInterval(state.queueTimer); state.queueTimer = null; } const renderers = { dashboard: renderDashboard, bands: renderBands, @@ -149,16 +152,16 @@ async function renderDashboard() { ${statCard("Stale (>30j)", t.stale, "warn")}
-
+

Par statut

${r.by_status.map((x) => barRow(x.status, x.total, maxStatus)).join("") || emptyRow()}
-
-

Par pays (top 20)

+
+

Par pays (${r.by_country.length})

${r.by_country.map((x) => barRow(x.country, x.total, maxCountry)).join("") || emptyRow()}
-
-

Par genre (top 20)

+
+

Par genre (${r.by_genre.length})

${r.by_genre.map((x) => barRow(x.genre, x.total, maxGenre)).join("") || emptyRow()}
@@ -181,39 +184,72 @@ function emptyRow() { return `
Aucune donnée
`; } // Queue // ------------------------------------------------------------------ async function renderQueue() { - content().innerHTML = `
Chargement…
`; - try { - const r = await api("/admin/api/queue"); - const b = r.breakdown; + if (!document.getElementById("q-body")) { content().innerHTML = ` -
- ${statCard("Nouveaux (priorité 1)", b.new_bands, "err")} - ${statCard("Modifiés depuis enrich (priorité 2)", b.modified_since_enrich, "warn")} - ${statCard("Legacy à ré-enrichir (priorité 3)", b.legacy_pending, "warn")} - ${statCard("Stale >30j (priorité 4)", b.stale, "ok")} -
+
-

Derniers runs d'enrichissement

+
+

Derniers runs d'enrichissement

+
+ Auto-refresh 15s + +
+
- - - ${r.recent_enrich_runs.map((x) => ` - - - - - - - - - - `).join("") || ``} - + +
IDStatutDébutFinVusEnrichisErreur
${x.id}${statusBadge(x.status)}${fmtDate(x.started_at)}${fmtDate(x.finished_at)}${x.bands_seen}${x.bands_enriched}${esc(x.error || "")}
Aucun run
IDStatutDébutFinVusEnrichisProgressionErreur
`; + document.getElementById("q-cleanup-btn").addEventListener("click", async () => { + const btn = document.getElementById("q-cleanup-btn"); + btn.disabled = true; + try { + const r = await api("/admin/api/crawl-runs/cleanup", { method: "POST", body: JSON.stringify({ older_than_minutes: 30 }) }); + alert(`${r.cleaned} run(s) annulé(s).`); + await loadQueue(); + } catch (e) { + alert(`Erreur: ${e.message}`); + } finally { btn.disabled = false; } + }); + } + await loadQueue(); + if (state.queueTimer) clearInterval(state.queueTimer); + state.queueTimer = setInterval(loadQueue, 15000); +} + +async function loadQueue() { + try { + const r = await api("/admin/api/queue"); + const b = r.breakdown; + const statsEl = document.getElementById("q-stats"); + if (statsEl) statsEl.innerHTML = ` + ${statCard("Nouveaux (priorité 1)", b.new_bands, "err")} + ${statCard("Modifiés depuis enrich (p.2)", b.modified_since_enrich, "warn")} + ${statCard("Legacy à ré-enrichir (p.3)", b.legacy_pending, "warn")} + ${statCard("Stale >30j (p.4)", b.stale, "ok")} + `; + const tbody = document.getElementById("q-body"); + if (tbody) tbody.innerHTML = r.recent_enrich_runs.map((x) => { + const isRunning = x.status === "running"; + const elapsed = isRunning ? Math.round((Date.now() - new Date(x.started_at)) / 60000) : null; + const prog = isRunning && x.bands_seen > 0 + ? `${x.bands_seen} vus, ${x.bands_enriched} enrichis${elapsed ? `, ${elapsed}min` : ""}` + : isRunning ? `en attente… (${elapsed}min)` : "—"; + return ` + ${x.id} + ${statusBadge(x.status)} + ${fmtDate(x.started_at)} + ${fmtDate(x.finished_at)} + ${x.bands_seen} + ${x.bands_enriched} + ${prog} + ${esc(x.error || "")} + `; + }).join("") || `Aucun run`; } catch (e) { - content().innerHTML = `
Erreur : ${esc(e.message)}
`; + const tbody = document.getElementById("q-body"); + if (tbody) tbody.innerHTML = `Erreur : ${esc(e.message)}`; } } @@ -229,14 +265,25 @@ async function renderBands() { const s = state.bands; content().innerHTML = `
- - - - + + + + + + +
@@ -245,10 +292,13 @@ async function renderBands() { `; document.getElementById("b-apply").addEventListener("click", () => { s.q = document.getElementById("b-q").value.trim(); + s.location_q = document.getElementById("b-location_q").value.trim(); s.country = document.getElementById("b-country").value.trim(); s.status = document.getElementById("b-status").value.trim(); s.genre = document.getElementById("b-genre").value.trim(); s.enriched = document.getElementById("b-enriched").value; + s.has_lat = document.getElementById("b-has_lat").value; + s.has_location = document.getElementById("b-has_location").value; s.page = 1; loadBands(); }); @@ -261,6 +311,7 @@ const BAND_COLUMNS = [ { key: "country", label: "Pays" }, { key: "status", label: "Statut" }, { key: "genre", label: "Genre" }, + { key: "location_text", label: "Lieu" }, { key: "formed_year", label: "Année" }, { key: "enriched", label: "Enrichi" }, { key: "crawled_at", label: "Crawlé le" }, @@ -273,10 +324,13 @@ async function loadBands() { page: s.page, pageSize: s.pageSize, sort: s.sort, dir: s.dir, }); if (s.q) params.set("q", s.q); + if (s.location_q) params.set("location_q", s.location_q); if (s.country) params.set("country", s.country); if (s.status) params.set("status", s.status); if (s.genre) params.set("genre", s.genre); if (s.enriched) params.set("enriched", s.enriched); + if (s.has_lat) params.set("has_lat", s.has_lat); + if (s.has_location) params.set("has_location", s.has_location); try { const r = await api(`/admin/api/bands?${params}`); @@ -295,6 +349,7 @@ async function loadBands() { ${esc(b.country || "")} ${esc(b.status || "")} ${esc(b.genre || "")} + ${esc(b.location_text || "")} ${b.formed_year || ""} ${b.enriched ? 'oui' : 'non'} ${fmtDate(b.crawled_at)} diff --git a/apps/api/src/adminRoutes.js b/apps/api/src/adminRoutes.js index b18bc54..1b16c86 100644 --- a/apps/api/src/adminRoutes.js +++ b/apps/api/src/adminRoutes.js @@ -45,16 +45,16 @@ export default async function adminRoutes(fastify, opts) { `), pool.query(` SELECT COALESCE(NULLIF(trim(status), ''), 'Unknown') AS status, count(*)::int AS total - FROM bands GROUP BY 1 ORDER BY total DESC LIMIT 20 + FROM bands GROUP BY 1 ORDER BY total DESC `), pool.query(` SELECT COALESCE(country, '??') AS country, count(*)::int AS total - FROM bands GROUP BY 1 ORDER BY total DESC LIMIT 20 + FROM bands GROUP BY 1 ORDER BY total DESC `), pool.query(` SELECT genre, count(*)::int AS total FROM bands WHERE genre IS NOT NULL AND genre != '' - GROUP BY 1 ORDER BY total DESC LIMIT 20 + GROUP BY 1 ORDER BY total DESC `), ]); return { @@ -110,7 +110,7 @@ export default async function adminRoutes(fastify, opts) { // ------------------------------------------------------------------ fastify.get("/admin/api/bands", async (req, reply) => { try { - const { q, country, status, genre, enriched, sort, dir } = req.query || {}; + const { q, location_q, country, status, genre, enriched, has_lat, has_location, sort, dir } = req.query || {}; const { page, pageSize, offset } = pagination(req.query || {}); const where = []; @@ -120,11 +120,19 @@ export default async function adminRoutes(fastify, opts) { if (q) { const query = String(q).trim().slice(0, 100); if (query.length >= 2) { - where.push(`(name ILIKE $${i} OR genre ILIKE $${i} OR location_text ILIKE $${i})`); + where.push(`(name ILIKE $${i} OR genre ILIKE $${i})`); vals.push(`%${query}%`); i++; } } + if (location_q) { + const lq = String(location_q).trim().slice(0, 100); + if (lq.length >= 1) { + where.push(`location_text ILIKE $${i}`); + vals.push(`%${lq}%`); + i++; + } + } if (country) { where.push(`country = $${i}`); vals.push(String(country).toUpperCase()); @@ -142,6 +150,10 @@ export default async function adminRoutes(fastify, opts) { } if (enriched === "true" || enriched === "1") where.push(`enriched = true`); if (enriched === "false" || enriched === "0") where.push(`enriched = false`); + if (has_lat === "true" || has_lat === "1") where.push(`lat IS NOT NULL`); + if (has_lat === "false" || has_lat === "0") where.push(`lat IS NULL`); + if (has_location === "true" || has_location === "1") where.push(`location_text IS NOT NULL AND location_text != ''`); + if (has_location === "false" || has_location === "0") where.push(`(location_text IS NULL OR location_text = '')`); const whereSql = where.length ? `WHERE ${where.join(" AND ")}` : ""; const sortCol = BAND_SORT_COLUMNS.has(sort) ? sort : "ma_id"; @@ -324,6 +336,29 @@ export default async function adminRoutes(fastify, opts) { } }); + // ------------------------------------------------------------------ + // Cleanup des runs bloqués (status=running depuis trop longtemps) + // ------------------------------------------------------------------ + fastify.post("/admin/api/crawl-runs/cleanup", async (req, reply) => { + try { + const olderThanMinutes = Math.max(1, Number(req.body?.older_than_minutes) || 30); + const r = await pool.query(` + UPDATE crawl_run + SET status = 'error', + finished_at = now(), + error = 'annulé manuellement (run bloqué)' + WHERE status = 'running' + AND started_at < now() - make_interval(mins => $1) + RETURNING id, run_type, started_at + `, [olderThanMinutes]); + await writeAuditLog(pool, req.adminUsername, "cleanup_stuck_runs", "crawl_run", null, null, { cleaned: r.rows }); + return { ok: true, cleaned: r.rows.length, runs: r.rows }; + } catch (err) { + fastify.log.error(err); + return reply.code(500).send({ ok: false, error: "Erreur cleanup" }); + } + }); + // ------------------------------------------------------------------ // Journal d'audit (actions admin) // ------------------------------------------------------------------ diff --git a/apps/crawler/src/db.py b/apps/crawler/src/db.py index 3e00f83..099b237 100644 --- a/apps/crawler/src/db.py +++ b/apps/crawler/src/db.py @@ -196,6 +196,22 @@ def log_event(level: str, message: str, run_id: Optional[int] = None, ma_id: Opt # Crawl run tracking # ------------------------------------------------------------------ +def update_crawl_run_progress(run_id: int, stats: Dict[str, int]): + """Mise à jour des compteurs d'un run en cours (progression live).""" + try: + with get_conn() as conn: + with conn.cursor() as cur: + cur.execute( + """UPDATE crawl_run SET + bands_seen=%s, bands_new=%s, bands_updated=%s, bands_enriched=%s + WHERE id=%s AND status='running'""", + (stats.get("seen", 0), stats.get("new", 0), + stats.get("updated", 0), stats.get("enriched", 0), run_id), + ) + except Exception as e: + log.warning(f"[db] update_crawl_run_progress failed: {e}") + + def start_crawl_run(run_type: str, countries: Optional[List[str]] = None) -> int: sql = "INSERT INTO crawl_run (run_type, countries) VALUES (%s, %s) RETURNING id" with get_conn() as conn: diff --git a/apps/crawler/src/jobs.py b/apps/crawler/src/jobs.py index ed161c1..ce2512d 100644 --- a/apps/crawler/src/jobs.py +++ b/apps/crawler/src/jobs.py @@ -14,6 +14,7 @@ from .db import ( upsert_bands, upsert_band_enriched, get_bands_to_enrich, start_crawl_run, finish_crawl_run, + update_crawl_run_progress, get_checkpoint, set_checkpoint, log_event, ) @@ -162,6 +163,9 @@ def run_enrich(session: MASession, limit: int = 100, country: Optional[str] = No if COOLDOWN_EVERY and (i + 1) % COOLDOWN_EVERY == 0: cooldown(COOLDOWN_MIN, COOLDOWN_MAX) + if (i + 1) % 50 == 0: + update_crawl_run_progress(run_id, stats) + log.info(f"[enrich] done: {stats}") log_event("info", f"enrich done: {stats}", run_id=run_id) except Exception as e: diff --git a/apps/crawler/src/ma_http.py b/apps/crawler/src/ma_http.py index ad9188b..1319469 100644 --- a/apps/crawler/src/ma_http.py +++ b/apps/crawler/src/ma_http.py @@ -62,9 +62,13 @@ class MASession: raise RuntimeError(f"HTTP {status} on {full_url}") try: return _extract_json(body) - except (json.JSONDecodeError, ValueError): - log.warning(f"[ma_http] non-JSON response from {url}") - raise + except (json.JSONDecodeError, ValueError) as e: + if attempt < retries: + log.warning(f"[ma_http] JSON invalide/vide depuis {url} (tentative {attempt+1}), refresh session…") + self.ensure_session(force_refresh=True) + sleep_range(LIST_MIN_DELAY * 2, LIST_MAX_DELAY * 2) + continue + raise RuntimeError(f"JSON parse error après {retries} tentatives sur {url}: {e}") raise RuntimeError(f"Failed after {retries} retries: {url}") def get_html(self, url: str, retries: int = 2) -> str: diff --git a/apps/geocoder/src/worker.py b/apps/geocoder/src/worker.py index cb28414..9af8a04 100644 --- a/apps/geocoder/src/worker.py +++ b/apps/geocoder/src/worker.py @@ -5,13 +5,12 @@ import json import requests import psycopg2 -NOMINATIM_BASE = os.environ.get("NOMINATIM_BASE", "https://nominatim.openstreetmap.org").rstrip("/") -NOMINATIM_EMAIL = os.environ.get("NOMINATIM_EMAIL", "").strip() -USER_AGENT = os.environ.get("NOMINATIM_USER_AGENT", "bm-geocoder/0.1 (contact: you@example.com)").strip() +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("NOMINATIM_MIN_DELAY", "1.05")) # >= 1s -JITTER = float(os.environ.get("NOMINATIM_JITTER", "0.35")) -MAX_PER_RUN = int(os.environ.get("GEOCODE_MAX_PER_RUN", "100000")) +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")) _last = 0.0 @@ -22,28 +21,28 @@ def polite_sleep(): time.sleep(wait) _last = time.monotonic() -def nominatim_search(query: str) -> dict | None: - # doc: /search + email conseillé pour gros volume :contentReference[oaicite:1]{index=1} - params = { - "q": query, - "format": "jsonv2", - "limit": 1, - "addressdetails": 1, - } - if NOMINATIM_EMAIL: - params["email"] = NOMINATIM_EMAIL - - headers = {"User-Agent": USER_AGENT} # requis par policy :contentReference[oaicite:2]{index=2} - +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() - r = requests.get(f"{NOMINATIM_BASE}/search", params=params, headers=headers, timeout=30) - if r.status_code in (403, 429, 503): - raise RuntimeError(f"nominatim_throttle status={r.status_code} body={r.text[:200]}") + r = requests.get(GEOAPIFY_BASE, params=params, timeout=30) + if r.status_code == 429: + raise RuntimeError(f"geoapify_throttle status=429") r.raise_for_status() - arr = r.json() - if not arr: + data = r.json() + features = data.get("features") or [] + if not features: return None - return arr[0] + props = features[0].get("properties", {}) + coords = features[0].get("geometry", {}).get("coordinates", []) + if len(coords) < 2: + return None + return { + "lat": props.get("lat") or coords[1], + "lon": props.get("lon") or coords[0], + "raw": data, + } def main(): dsn = os.environ["DATABASE_URL"] @@ -53,7 +52,6 @@ def main(): processed = 0 with conn.cursor() as cur: while processed < MAX_PER_RUN: - # prend 1 job à faire cur.execute( """ SELECT ma_id, query, country, tries @@ -73,17 +71,12 @@ def main(): ma_id, query, country, tries = row - # marque processing cur.execute( - """ - UPDATE geocode_queue - SET status='processing', updated_at=now() - WHERE ma_id=%s - """, + "UPDATE geocode_queue SET status='processing', updated_at=now() WHERE ma_id=%s", (ma_id,), ) - # 1) cache ? + # 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: @@ -94,7 +87,7 @@ def main(): SET lat=%s, lon=%s, geom=ST_SetSRID(ST_MakePoint(%s,%s),4326)::geography, geocoded_at=now(), - geocode_provider='nominatim-cache', + geocode_provider='geoapify-cache', geocode_query=%s, geocode_raw=%s, geocode_error=NULL, @@ -108,24 +101,19 @@ def main(): print(f"[worker] ma_id={ma_id} cache HIT -> done") continue - # 2) requête nominatim + # requête Geoapify try: - res = nominatim_search(query) + res = geoapify_search(query) if not res: cur.execute( - """ - UPDATE bands - SET geocode_error=%s, geocode_error_at=now() - WHERE ma_id=%s - """, + "UPDATE bands SET geocode_error=%s, geocode_error_at=now() WHERE ma_id=%s", ("no_result", ma_id), ) cur.execute( """ UPDATE geocode_queue SET status='error', tries=tries+1, last_error=%s, - next_run_at=now() + interval '7 days', - updated_at=now() + next_run_at=now() + interval '7 days', updated_at=now() WHERE ma_id=%s """, ("no_result", ma_id), @@ -137,33 +125,31 @@ def main(): lat = float(res["lat"]) lon = float(res["lon"]) - # écrit cache cur.execute( """ INSERT INTO geocode_cache(query, provider, lat, lon, geom, raw, updated_at) - VALUES (%s,'nominatim',%s,%s,ST_SetSRID(ST_MakePoint(%s,%s),4326)::geography,%s,now()) + 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, updated_at=now() + raw=EXCLUDED.raw, provider=EXCLUDED.provider, updated_at=now() """, - (query, lat, lon, lon, lat, json.dumps(res)), + (query, lat, lon, lon, lat, json.dumps(res["raw"])), ) - # écrit bands cur.execute( """ UPDATE bands SET lat=%s, lon=%s, geom=ST_SetSRID(ST_MakePoint(%s,%s),4326)::geography, geocoded_at=now(), - geocode_provider='nominatim', + 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), ma_id), + (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,)) @@ -171,22 +157,16 @@ def main(): print(f"[worker] ma_id={ma_id} OK lat={lat} lon={lon}") except Exception as e: - # backoff progressif backoff_minutes = min(60, 5 * (tries + 1)) cur.execute( - """ - UPDATE bands - SET geocode_error=%s, geocode_error_at=now() - WHERE ma_id=%s - """, + "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() + next_run_at=now() + (%s || ' minutes')::interval, updated_at=now() WHERE ma_id=%s """, (str(e)[:400], backoff_minutes, ma_id), diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 9af2eec..7ebf760 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -41,11 +41,9 @@ services: working_dir: /app environment: DATABASE_URL: ${DATABASE_URL} - NOMINATIM_BASE: "https://nominatim.openstreetmap.org" - NOMINATIM_EMAIL: ${NOMINATIM_EMAIL} - NOMINATIM_USER_AGENT: ${NOMINATIM_USER_AGENT} - NOMINATIM_MIN_DELAY: "1.05" - NOMINATIM_JITTER: "0.35" + GEOAPIFY_API_KEY: ${GEOAPIFY_API_KEY} + GEOCODER_MIN_DELAY: "0.22" + GEOCODER_JITTER: "0.10" GEOCODE_MAX_PER_RUN: "5000" command: ["python", "src/worker.py"] networks: diff --git a/docker-compose.yml b/docker-compose.yml index 758bcdb..236ea70 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -13,11 +13,9 @@ services: working_dir: /app environment: DATABASE_URL: ${DATABASE_URL} - NOMINATIM_BASE: "https://nominatim.openstreetmap.org" - NOMINATIM_EMAIL: ${NOMINATIM_EMAIL} - NOMINATIM_USER_AGENT: ${NOMINATIM_USER_AGENT} - NOMINATIM_MIN_DELAY: "1.05" - NOMINATIM_JITTER: "0.35" + GEOAPIFY_API_KEY: ${GEOAPIFY_API_KEY} + GEOCODER_MIN_DELAY: "0.22" + GEOCODER_JITTER: "0.10" GEOCODE_MAX_PER_RUN: "5000" command: ["python", "src/worker.py"] networks: