Nicolas Fryder
222e71a608
L'admin mélangeait pipeline legacy et récent (onglet "Queue" pointait vers geocode_queue mort → erreur "Cannot read properties of undefined (reading 'reduce')"), dispersait les logs (Monitor/Historique/Jobs) et les actions (Géocodage/Jobs/Checkpoints) sur des onglets séparés sans lien entre eux. Nouvelle nav (5 onglets) : - Vue d'ensemble : stats bands + genre/pays (ex-Dashboard) + cartes live auto-refresh 15s (crawl en cours, géocodage, jobs en attente, file d'enrichissement) — fusion Dashboard+Monitor+Queue. - Groupes : liste bands inchangée + filtre "Conflits seulement" (nouveau has_conflict sur GET /admin/api/bands) ; le modal band affiche maintenant une section "Conflits en attente" resolvable inline (garder ma valeur / accepter Metal Archives) — remplace l'onglet Conflits séparé. - Activité : liste chronologique unique (GET /admin/api/activity, fusion crawl_run tous types + admin_audit_log), cliquable → dialog avec logs complets du run (GET /admin/api/logs?run_id=) ou diff avant/après pour une action admin. Remplace Crawl runs + Historique + Audit + la liste de jobs de l'ex-onglet Jobs. - Actions : toutes les commandes (crawl, géocodage, maintenance) sur une seule page, groupées, chaque bouton vérifié contre son endpoint réel + checkpoints en lecture seule. Remplace Jobs + les boutons de Géocodage. - LLM : inchangé. Backend - endpoints morts supprimés (audit-log, conflicts liste, crawl-runs liste — tous remplacés par /admin/api/activity ou le filtre has_conflict) ; vérifié 1:1 qu'aucun appel front ne pointe vers un endpoint disparu et qu'aucun endpoint restant n'est orphelin. - GET /admin/api/bands : nouveau filtre has_conflict. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
805 lines
33 KiB
JavaScript
805 lines
33 KiB
JavaScript
import { requireAdminSession, writeAuditLog } from "./adminAuth.js";
|
|
|
|
const BAND_SORT_COLUMNS = new Set([
|
|
"ma_id", "name", "country", "status", "genre",
|
|
"formed_year", "enriched", "crawled_at", "updated_at", "first_seen_at",
|
|
]);
|
|
|
|
const BAND_EDITABLE_FIELDS = [
|
|
"name", "country", "status", "genre", "formed_year",
|
|
"themes", "location_text", "lat", "lon",
|
|
];
|
|
|
|
function pagination(query) {
|
|
const page = Math.max(1, Number(query.page) || 1);
|
|
const pageSize = Math.min(200, Math.max(1, Number(query.pageSize) || 50));
|
|
return { page, pageSize, offset: (page - 1) * pageSize };
|
|
}
|
|
|
|
export default async function adminRoutes(fastify, opts) {
|
|
const { pool } = opts;
|
|
|
|
fastify.addHook("preHandler", async (req, reply) => {
|
|
const username = requireAdminSession(req, reply);
|
|
if (!username) return reply; // already sent 401
|
|
req.adminUsername = username;
|
|
});
|
|
|
|
// ------------------------------------------------------------------
|
|
// Stats globales
|
|
// ------------------------------------------------------------------
|
|
fastify.get("/admin/api/stats", async (req, reply) => {
|
|
try {
|
|
const [totals, byStatus, byCountry, byGenre] = await Promise.all([
|
|
pool.query(`
|
|
SELECT
|
|
count(*)::int AS total,
|
|
count(*) FILTER (WHERE enriched = true)::int AS enriched,
|
|
count(*) FILTER (WHERE enriched = false)::int AS not_enriched,
|
|
count(*) FILTER (WHERE geom IS NOT NULL)::int AS geocoded,
|
|
count(*) FILTER (WHERE geom IS NULL)::int AS not_geocoded,
|
|
count(*) FILTER (WHERE crawled_at IS NOT NULL)::int AS crawled_by_current_system,
|
|
count(*) FILTER (WHERE data->'band_page' IS NULL)::int AS never_enriched,
|
|
count(*) FILTER (WHERE crawled_at < now() - interval '30 days')::int AS stale
|
|
FROM bands
|
|
`),
|
|
pool.query(`
|
|
SELECT COALESCE(NULLIF(trim(status), ''), 'Unknown') AS status, count(*)::int AS total
|
|
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
|
|
`),
|
|
pool.query(`
|
|
SELECT genre, count(*)::int AS total
|
|
FROM bands WHERE genre IS NOT NULL AND genre != ''
|
|
GROUP BY 1 ORDER BY total DESC
|
|
`),
|
|
]);
|
|
return {
|
|
ok: true,
|
|
totals: totals.rows[0],
|
|
by_status: byStatus.rows,
|
|
by_country: byCountry.rows,
|
|
by_genre: byGenre.rows,
|
|
};
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur stats" });
|
|
}
|
|
});
|
|
|
|
// ------------------------------------------------------------------
|
|
// Queue d'enrichissement (même logique que le crawler Python)
|
|
// ------------------------------------------------------------------
|
|
fastify.get("/admin/api/queue", async (req, reply) => {
|
|
try {
|
|
const r = await pool.query(`
|
|
SELECT
|
|
count(*) FILTER (WHERE data->'band_page' IS NULL)::int AS new_bands,
|
|
count(*) FILTER (
|
|
WHERE crawled_at IS NOT NULL AND updated_at > crawled_at + interval '1 minute'
|
|
)::int AS modified_since_enrich,
|
|
count(*) FILTER (
|
|
WHERE crawled_at IS NULL AND data->'band_page' IS NOT NULL
|
|
)::int AS legacy_pending,
|
|
count(*) FILTER (
|
|
WHERE crawled_at IS NOT NULL AND crawled_at < now() - interval '30 days'
|
|
)::int AS stale
|
|
FROM bands
|
|
WHERE data->>'url' IS NOT NULL
|
|
`);
|
|
const recentRuns = await pool.query(`
|
|
SELECT id, run_type, status, started_at, finished_at,
|
|
bands_seen, bands_new, bands_updated, bands_enriched, error
|
|
FROM crawl_run
|
|
WHERE run_type = 'enrich'
|
|
ORDER BY started_at DESC
|
|
LIMIT 5
|
|
`);
|
|
return { ok: true, breakdown: r.rows[0], recent_enrich_runs: recentRuns.rows };
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur queue" });
|
|
}
|
|
});
|
|
|
|
// ------------------------------------------------------------------
|
|
// Bands - liste paginée / recherche / filtres
|
|
// ------------------------------------------------------------------
|
|
fastify.get("/admin/api/bands", async (req, reply) => {
|
|
try {
|
|
const { q, location_q, country, status, genre, themes_q, enriched, has_lat, has_location, has_conflict, sort, dir } = req.query || {};
|
|
const { page, pageSize, offset } = pagination(req.query || {});
|
|
|
|
const where = [];
|
|
const vals = [];
|
|
let i = 1;
|
|
|
|
if (q) {
|
|
const query = String(q).trim().slice(0, 100);
|
|
if (query.length >= 2) {
|
|
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());
|
|
i++;
|
|
}
|
|
if (status) {
|
|
where.push(`COALESCE(NULLIF(trim(status), ''), 'Unknown') = $${i}`);
|
|
vals.push(String(status));
|
|
i++;
|
|
}
|
|
if (genre) {
|
|
where.push(`genre ILIKE $${i}`);
|
|
vals.push(`%${String(genre)}%`);
|
|
i++;
|
|
}
|
|
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 = '')`);
|
|
if (has_conflict === "true" || has_conflict === "1") where.push(`crawler_pending IS NOT NULL AND crawler_pending <> '{}'::jsonb`);
|
|
if (themes_q) {
|
|
const tq = String(themes_q).trim().slice(0, 100);
|
|
if (tq.length >= 1) { where.push(`themes ILIKE $${i}`); vals.push(`%${tq}%`); i++; }
|
|
}
|
|
|
|
const whereSql = where.length ? `WHERE ${where.join(" AND ")}` : "";
|
|
const sortCol = BAND_SORT_COLUMNS.has(sort) ? sort : "ma_id";
|
|
const sortDir = dir === "desc" ? "DESC" : "ASC";
|
|
|
|
const countSql = `SELECT count(*)::int AS total FROM bands ${whereSql}`;
|
|
const dataSql = `
|
|
SELECT ma_id, name, country, status, genre, location_text, themes, formed_year,
|
|
enriched, lat, lon, crawled_at, updated_at, first_seen_at, locked_fields
|
|
FROM bands
|
|
${whereSql}
|
|
ORDER BY ${sortCol} ${sortDir} NULLS LAST
|
|
LIMIT $${i} OFFSET $${i + 1}
|
|
`;
|
|
vals.push(pageSize, offset);
|
|
|
|
const [countRes, dataRes] = await Promise.all([
|
|
pool.query(countSql, vals.slice(0, i - 1)),
|
|
pool.query(dataSql, vals),
|
|
]);
|
|
|
|
return {
|
|
ok: true,
|
|
items: dataRes.rows,
|
|
total: countRes.rows[0].total,
|
|
page,
|
|
pageSize,
|
|
};
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur recherche bands" });
|
|
}
|
|
});
|
|
|
|
fastify.get("/admin/api/bands/:ma_id", async (req, reply) => {
|
|
try {
|
|
const id = Number(req.params.ma_id);
|
|
if (!Number.isFinite(id) || id < 0) {
|
|
return reply.code(400).send({ ok: false, error: "bad ma_id" });
|
|
}
|
|
const [r, locs, llm] = await Promise.all([
|
|
pool.query(`SELECT * FROM bands WHERE ma_id = $1 LIMIT 1`, [id]),
|
|
pool.query(`
|
|
SELECT id, step_order, step_label, location_raw, is_country_only,
|
|
lat, lon, geocode_status, geocode_query, geocode_provider,
|
|
geocode_confidence, geocode_error, geocode_tries_geo,
|
|
geocode_tries_llm, geocode_next_at, updated_at
|
|
FROM band_locations
|
|
WHERE ma_id = $1
|
|
ORDER BY step_order ASC, id ASC
|
|
`, [id]).catch(() => ({ rows: [] })),
|
|
pool.query(`
|
|
SELECT id, model, location_raw, country, parsed_city, parsed_country,
|
|
is_null, tokens_in, tokens_out, cost_usd, prompt, response, created_at
|
|
FROM llm_cache
|
|
WHERE ma_id = $1
|
|
OR lower(location_raw) IN (
|
|
SELECT lower(location_raw) FROM band_locations WHERE ma_id = $1
|
|
)
|
|
ORDER BY created_at DESC
|
|
LIMIT 50
|
|
`, [id]).catch(() => ({ rows: [] })),
|
|
]);
|
|
if (!r.rows.length) return reply.code(404).send({ ok: false, error: "not found" });
|
|
return { ok: true, item: r.rows[0], locations: locs.rows, llm: llm.rows };
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur récupération band" });
|
|
}
|
|
});
|
|
|
|
fastify.patch("/admin/api/bands/:ma_id", async (req, reply) => {
|
|
try {
|
|
const id = Number(req.params.ma_id);
|
|
if (!Number.isFinite(id) || id < 0) {
|
|
return reply.code(400).send({ ok: false, error: "bad ma_id" });
|
|
}
|
|
const body = req.body || {};
|
|
const updates = {};
|
|
for (const field of BAND_EDITABLE_FIELDS) {
|
|
if (Object.prototype.hasOwnProperty.call(body, field)) {
|
|
updates[field] = body[field];
|
|
}
|
|
}
|
|
if (Object.keys(updates).length === 0) {
|
|
return reply.code(400).send({ ok: false, error: "Aucun champ à modifier" });
|
|
}
|
|
if ("formed_year" in updates) {
|
|
const y = updates.formed_year === null ? null : Number(updates.formed_year);
|
|
if (y !== null && (!Number.isFinite(y) || y < 1800 || y > 2100)) {
|
|
return reply.code(400).send({ ok: false, error: "formed_year invalide" });
|
|
}
|
|
updates.formed_year = y;
|
|
}
|
|
if ("lat" in updates) {
|
|
const v = updates.lat === null ? null : Number(updates.lat);
|
|
if (v !== null && (!Number.isFinite(v) || v < -90 || v > 90)) {
|
|
return reply.code(400).send({ ok: false, error: "lat invalide" });
|
|
}
|
|
updates.lat = v;
|
|
}
|
|
if ("lon" in updates) {
|
|
const v = updates.lon === null ? null : Number(updates.lon);
|
|
if (v !== null && (!Number.isFinite(v) || v < -180 || v > 180)) {
|
|
return reply.code(400).send({ ok: false, error: "lon invalide" });
|
|
}
|
|
updates.lon = v;
|
|
}
|
|
|
|
const before = await pool.query(`SELECT * FROM bands WHERE ma_id = $1 LIMIT 1`, [id]);
|
|
if (!before.rows.length) return reply.code(404).send({ ok: false, error: "not found" });
|
|
|
|
const setCols = Object.keys(updates);
|
|
// Marque les champs édités comme verrouillés (ne seront plus écrasés par le crawler)
|
|
const newLocked = Object.fromEntries(setCols.map((c) => [c, true]));
|
|
const setSql = setCols.map((c, idx) => `${c} = $${idx + 2}`).join(", ");
|
|
const vals = [id, ...setCols.map((c) => updates[c])];
|
|
|
|
const after = await pool.query(
|
|
`UPDATE bands SET ${setSql},
|
|
locked_fields = locked_fields || $${vals.length + 1}::jsonb,
|
|
crawler_pending = (
|
|
SELECT jsonb_strip_nulls(jsonb_build_object(${
|
|
setCols.map((c) => `'${c}', crawler_pending->'${c}'`).join(", ")
|
|
}))
|
|
FROM bands WHERE ma_id = $1
|
|
)
|
|
WHERE ma_id = $1 RETURNING *`,
|
|
[...vals, JSON.stringify(newLocked)]
|
|
);
|
|
|
|
await writeAuditLog(
|
|
pool, req.adminUsername, "update", "bands", id,
|
|
before.rows[0], after.rows[0]
|
|
);
|
|
|
|
return { ok: true, item: after.rows[0] };
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur mise à jour band" });
|
|
}
|
|
});
|
|
|
|
// ------------------------------------------------------------------
|
|
// Checkpoints
|
|
// (l'historique des crawl_run est exposé de façon unifiée par
|
|
// GET /admin/api/activity, avec le détail logs via GET /admin/api/logs?run_id=)
|
|
// ------------------------------------------------------------------
|
|
fastify.get("/admin/api/crawl-checkpoints", async (req, reply) => {
|
|
try {
|
|
const r = await pool.query(`SELECT key, value, updated_at FROM crawl_checkpoint ORDER BY key`);
|
|
return { ok: true, items: r.rows };
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur checkpoints" });
|
|
}
|
|
});
|
|
|
|
// ------------------------------------------------------------------
|
|
// Logs crawler
|
|
// ------------------------------------------------------------------
|
|
fastify.get("/admin/api/logs", async (req, reply) => {
|
|
try {
|
|
const { level, run_id, since, min_id } = req.query || {};
|
|
const { page, pageSize, offset } = pagination(req.query || {});
|
|
const where = [];
|
|
const vals = [];
|
|
let i = 1;
|
|
if (level) { where.push(`level = $${i}`); vals.push(level); i++; }
|
|
if (run_id) { where.push(`run_id = $${i}`); vals.push(Number(run_id)); i++; }
|
|
if (since) { where.push(`created_at > $${i}`); vals.push(since); i++; }
|
|
if (min_id) { where.push(`id > $${i}`); vals.push(Number(min_id)); i++; }
|
|
const whereSql = where.length ? `WHERE ${where.join(" AND ")}` : "";
|
|
|
|
vals.push(pageSize, offset);
|
|
const r = await pool.query(`
|
|
SELECT id, run_id, level, message, ma_id, created_at
|
|
FROM crawl_log ${whereSql}
|
|
ORDER BY created_at DESC
|
|
LIMIT $${i} OFFSET $${i + 1}
|
|
`, vals);
|
|
|
|
return { ok: true, items: r.rows };
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur logs" });
|
|
}
|
|
});
|
|
|
|
// ------------------------------------------------------------------
|
|
// 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" });
|
|
}
|
|
});
|
|
|
|
// ------------------------------------------------------------------
|
|
// Conflits : champs verrouillés avec valeur différente côté crawler.
|
|
// La liste bulk est remplacée par le filtre has_conflict=1 sur
|
|
// GET /admin/api/bands (visible dans l'onglet Groupes) ; la résolution
|
|
// champ par champ reste ici et est utilisée depuis le modal band.
|
|
// ------------------------------------------------------------------
|
|
fastify.post("/admin/api/bands/:ma_id/resolve-conflict", async (req, reply) => {
|
|
try {
|
|
const id = Number(req.params.ma_id);
|
|
if (!Number.isFinite(id)) return reply.code(400).send({ ok: false, error: "bad ma_id" });
|
|
const { field, action } = req.body || {}; // action: 'keep_mine' | 'accept_crawler'
|
|
if (!field || !["keep_mine", "accept_crawler"].includes(action)) {
|
|
return reply.code(400).send({ ok: false, error: "field et action requis" });
|
|
}
|
|
const before = await pool.query(`SELECT * FROM bands WHERE ma_id = $1`, [id]);
|
|
if (!before.rows.length) return reply.code(404).send({ ok: false, error: "not found" });
|
|
|
|
let sql, vals;
|
|
if (action === "keep_mine") {
|
|
// Garde la valeur actuelle, efface juste le pending pour ce champ
|
|
sql = `UPDATE bands SET crawler_pending = crawler_pending - $2 WHERE ma_id = $1`;
|
|
vals = [id, field];
|
|
} else {
|
|
// Applique la valeur du crawler, déverrouille le champ
|
|
const crawlerVal = before.rows[0].crawler_pending?.[field];
|
|
sql = `UPDATE bands SET ${field} = $2::text, locked_fields = locked_fields - $3, crawler_pending = crawler_pending - $3 WHERE ma_id = $1`;
|
|
vals = [id, crawlerVal, field];
|
|
}
|
|
const after = await pool.query(sql + " RETURNING *", vals);
|
|
await writeAuditLog(pool, req.adminUsername, `resolve_conflict:${action}`, "bands", id, before.rows[0], after.rows[0]);
|
|
return { ok: true, item: after.rows[0] };
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur résolution conflit" });
|
|
}
|
|
});
|
|
|
|
// ------------------------------------------------------------------
|
|
// Géocodage : stats + déclenchement
|
|
// ------------------------------------------------------------------
|
|
fastify.get("/admin/api/geocoding", async (req, reply) => {
|
|
try {
|
|
const [cache, recent, locations, llm, providers, llmModels] = await Promise.all([
|
|
pool.query(`SELECT count(*)::int AS n FROM geocode_cache`),
|
|
pool.query(`
|
|
SELECT b.ma_id, b.name, b.country, b.geocode_provider, b.geocoded_at,
|
|
b.geocode_query, b.geocode_error
|
|
FROM bands b
|
|
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,
|
|
count(*) FILTER (WHERE is_null)::int AS n_null
|
|
FROM llm_cache
|
|
`).catch(() => ({ rows: [{ n: 0, total_cost_usd: 0, n_null: 0 }] })),
|
|
pool.query(`
|
|
SELECT COALESCE(geocode_provider, '(en attente)') AS provider,
|
|
count(*)::int AS n,
|
|
round(avg(geocode_confidence)::numeric, 2) AS avg_conf
|
|
FROM band_locations
|
|
GROUP BY geocode_provider ORDER BY n DESC
|
|
`).catch(() => ({ rows: [] })),
|
|
pool.query(`
|
|
SELECT model,
|
|
count(*)::int AS n,
|
|
count(*) FILTER (WHERE is_null)::int AS n_null,
|
|
sum(cost_usd)::numeric(10,6) AS cost
|
|
FROM llm_cache GROUP BY model ORDER BY n DESC
|
|
`).catch(() => ({ rows: [] })),
|
|
]);
|
|
return {
|
|
ok: true,
|
|
cache_size: cache.rows[0].n,
|
|
recent: recent.rows,
|
|
locations: locations.rows,
|
|
llm_cache: llm.rows[0],
|
|
providers: providers.rows,
|
|
llm_models: llmModels.rows,
|
|
};
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur geocoding stats" });
|
|
}
|
|
});
|
|
|
|
// Appels LLM récents (debug). Filtres: model, only_null=1, q (location).
|
|
fastify.get("/admin/api/llm", async (req, reply) => {
|
|
try {
|
|
const { model, only_null, q } = req.query || {};
|
|
const { pageSize, offset } = pagination(req.query || {});
|
|
const where = [];
|
|
const vals = [];
|
|
let i = 1;
|
|
if (model) { where.push(`model = $${i}`); vals.push(String(model)); i++; }
|
|
if (only_null === "1" || only_null === "true") where.push(`is_null = true`);
|
|
if (q) {
|
|
const qq = String(q).trim().slice(0, 100);
|
|
if (qq.length >= 1) {
|
|
where.push(`(location_raw ILIKE $${i} OR parsed_city ILIKE $${i})`);
|
|
vals.push(`%${qq}%`); i++;
|
|
}
|
|
}
|
|
const whereSql = where.length ? `WHERE ${where.join(" AND ")}` : "";
|
|
const [rows, cnt] = await Promise.all([
|
|
pool.query(`
|
|
SELECT lc.id, lc.ma_id, b.name AS band_name, lc.model, lc.location_raw,
|
|
lc.country, lc.parsed_city, lc.parsed_country, lc.is_null,
|
|
lc.tokens_in, lc.tokens_out, lc.cost_usd, lc.prompt, lc.response,
|
|
lc.created_at
|
|
FROM llm_cache lc
|
|
LEFT JOIN bands b ON b.ma_id = lc.ma_id
|
|
${whereSql}
|
|
ORDER BY lc.created_at DESC
|
|
LIMIT $${i} OFFSET $${i + 1}
|
|
`, [...vals, pageSize, offset]),
|
|
pool.query(`SELECT count(*)::int AS total FROM llm_cache lc ${whereSql}`, vals),
|
|
]);
|
|
return { ok: true, items: rows.rows, total: cnt.rows[0].total };
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur LLM debug" });
|
|
}
|
|
});
|
|
|
|
// ------------------------------------------------------------------
|
|
// Job triggers (déclenche un job dans le crawler sans redéployer)
|
|
// ------------------------------------------------------------------
|
|
fastify.get("/admin/api/job-triggers", async (req, reply) => {
|
|
try {
|
|
const r = await pool.query(`
|
|
SELECT id, job_type, status, requested_by, created_at, started_at, finished_at, error
|
|
FROM job_triggers ORDER BY created_at DESC LIMIT 30
|
|
`);
|
|
return { ok: true, items: r.rows };
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur job triggers" });
|
|
}
|
|
});
|
|
|
|
fastify.post("/admin/api/job-triggers", async (req, reply) => {
|
|
try {
|
|
const { job_type } = req.body || {};
|
|
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(", ")}` });
|
|
}
|
|
const r = await pool.query(
|
|
`INSERT INTO job_triggers (job_type, requested_by) VALUES ($1, $2) RETURNING id`,
|
|
[job_type, req.adminUsername]
|
|
);
|
|
const triggerId = r.rows[0].id;
|
|
await pool.query(
|
|
`INSERT INTO crawl_log (level, message) VALUES ($1, $2)`,
|
|
["info", `[admin:${req.adminUsername}] job trigger créé: ${job_type} (#${triggerId})`]
|
|
).catch(() => {});
|
|
return { ok: true, id: triggerId };
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur création job" });
|
|
}
|
|
});
|
|
|
|
// ------------------------------------------------------------------
|
|
// Géocodage — RESET complet & purge cache (pour tout relancer proprement)
|
|
// ------------------------------------------------------------------
|
|
// Remet TOUTES les band_locations (hors pays-seul) en 'queued' et efface les
|
|
// résultats : lat/lon, provider, confiance, granularité, query, erreurs, essais.
|
|
fastify.post("/admin/api/locations/reset-all", async (req, reply) => {
|
|
try {
|
|
const r = await pool.query(`
|
|
UPDATE band_locations
|
|
SET geocode_status='queued',
|
|
lat=NULL, lon=NULL,
|
|
geocode_provider=NULL, geocode_confidence=NULL, geocode_granularity=NULL,
|
|
geocode_query=NULL, geocode_error=NULL,
|
|
geocode_tries_geo=0, geocode_tries_llm=0,
|
|
geocode_next_at=now(), updated_at=now()
|
|
WHERE is_country_only = FALSE
|
|
`);
|
|
const count = r.rowCount;
|
|
await pool.query(
|
|
`INSERT INTO crawl_log (level, message) VALUES ($1, $2)`,
|
|
["warning", `[admin:${req.adminUsername}] RESET géocodage complet: ${count} band_locations remises à zéro`]
|
|
).catch(() => {});
|
|
return { ok: true, count };
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur reset-all" });
|
|
}
|
|
});
|
|
|
|
// Purge les entrées de cache issues de l'ancien pipeline Nominatim (raw avec
|
|
// place_rank). Force le worker à re-géocoder ces lieux via Geoapify.
|
|
fastify.post("/admin/api/geocode-cache/purge-nominatim", async (req, reply) => {
|
|
try {
|
|
const r = await pool.query(`DELETE FROM geocode_cache WHERE raw ? 'place_rank'`);
|
|
const count = r.rowCount;
|
|
await pool.query(
|
|
`INSERT INTO crawl_log (level, message) VALUES ($1, $2)`,
|
|
["warning", `[admin:${req.adminUsername}] purge cache Nominatim: ${count} entrée(s) supprimée(s)`]
|
|
).catch(() => {});
|
|
return { ok: true, count };
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur purge-nominatim" });
|
|
}
|
|
});
|
|
|
|
// ------------------------------------------------------------------
|
|
// 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)
|
|
// ------------------------------------------------------------------
|
|
fastify.get("/admin/api/live", async (req, reply) => {
|
|
try {
|
|
const [active_runs, pending_jobs, geo_queue, geo_processing, checkpoints] = await Promise.all([
|
|
pool.query(`
|
|
SELECT id, run_type, countries, status, started_at,
|
|
bands_seen, bands_new, bands_updated, bands_enriched, error
|
|
FROM crawl_run WHERE status = 'running' ORDER BY started_at DESC
|
|
`),
|
|
pool.query(`
|
|
SELECT id, job_type, status, requested_by, created_at
|
|
FROM job_triggers WHERE status = 'pending' ORDER BY created_at ASC
|
|
`),
|
|
pool.query(`SELECT geocode_status AS status, count(*)::int AS n FROM band_locations GROUP BY geocode_status ORDER BY n DESC`),
|
|
pool.query(`
|
|
SELECT bl.ma_id, b.name, b.country, bl.location_raw, bl.geocode_tries_geo AS tries,
|
|
bl.geocode_error AS last_error, bl.updated_at
|
|
FROM band_locations bl JOIN bands b ON b.ma_id = bl.ma_id
|
|
WHERE bl.geocode_status = 'processing' LIMIT 3
|
|
`),
|
|
pool.query(`SELECT key, value, updated_at FROM crawl_checkpoint ORDER BY key`),
|
|
]);
|
|
return {
|
|
ok: true,
|
|
active_runs: active_runs.rows,
|
|
pending_jobs: pending_jobs.rows,
|
|
geo_queue: geo_queue.rows,
|
|
geo_processing: geo_processing.rows,
|
|
checkpoints: checkpoints.rows,
|
|
};
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur live status" });
|
|
}
|
|
});
|
|
|
|
// ------------------------------------------------------------------
|
|
// Annuler un crawl run spécifique
|
|
// ------------------------------------------------------------------
|
|
fastify.post("/admin/api/crawl-runs/:id/cancel", async (req, reply) => {
|
|
try {
|
|
const id = Number(req.params.id);
|
|
if (!Number.isFinite(id)) return reply.code(400).send({ ok: false, error: "bad id" });
|
|
const r = await pool.query(`
|
|
UPDATE crawl_run SET status='error', finished_at=now(), error='annulé manuellement par admin'
|
|
WHERE id=$1 AND status='running' RETURNING id, run_type
|
|
`, [id]);
|
|
if (!r.rows.length) return reply.code(404).send({ ok: false, error: "Run non trouvé ou déjà terminé" });
|
|
await pool.query(
|
|
`INSERT INTO crawl_log (level, message) VALUES ($1, $2)`,
|
|
["warning", `[admin:${req.adminUsername}] run #${id} (${r.rows[0].run_type}) annulé manuellement`]
|
|
).catch(() => {});
|
|
return { ok: true, id };
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur annulation run" });
|
|
}
|
|
});
|
|
|
|
// ------------------------------------------------------------------
|
|
// Annuler un job trigger en attente
|
|
// ------------------------------------------------------------------
|
|
fastify.post("/admin/api/job-triggers/:id/cancel", async (req, reply) => {
|
|
try {
|
|
const id = Number(req.params.id);
|
|
if (!Number.isFinite(id)) return reply.code(400).send({ ok: false, error: "bad id" });
|
|
const r = await pool.query(`
|
|
UPDATE job_triggers SET status='cancelled', finished_at=now()
|
|
WHERE id=$1 AND status='pending' RETURNING id, job_type
|
|
`, [id]);
|
|
if (!r.rows.length) return reply.code(404).send({ ok: false, error: "Job non trouvé ou déjà traité" });
|
|
await pool.query(
|
|
`INSERT INTO crawl_log (level, message) VALUES ($1, $2)`,
|
|
["warning", `[admin:${req.adminUsername}] job trigger #${id} (${r.rows[0].job_type}) annulé`]
|
|
).catch(() => {});
|
|
return { ok: true, id };
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur annulation job" });
|
|
}
|
|
});
|
|
|
|
// Le journal d'audit (actions admin) est désormais exposé de façon unifiée
|
|
// par GET /admin/api/activity (type=admin_action), avec crawl_run côté
|
|
// system. writeAuditLog() continue d'alimenter admin_audit_log ci-dessous.
|
|
|
|
// ------------------------------------------------------------------
|
|
// Activité — liste unifiée : crawl_run (crawl/enrich/geocoder_enqueue,
|
|
// manuel ou planifié) + admin_audit_log (actions admin). Une seule liste
|
|
// chronologique cliquable pour l'admin ; le détail d'un 'run' se récupère
|
|
// via GET /admin/api/logs?run_id=, celui d'un 'admin_action' est déjà
|
|
// inclus (before/after).
|
|
// ------------------------------------------------------------------
|
|
fastify.get("/admin/api/activity", async (req, reply) => {
|
|
try {
|
|
const { type, status } = req.query || {};
|
|
const { page, pageSize, offset } = pagination(req.query || {});
|
|
|
|
const base = `
|
|
SELECT
|
|
'run'::text AS type, id, run_type AS subtype, status,
|
|
NULL::text AS actor, started_at AS ts, finished_at,
|
|
jsonb_build_object(
|
|
'bands_seen', bands_seen, 'bands_new', bands_new,
|
|
'bands_updated', bands_updated, 'bands_enriched', bands_enriched,
|
|
'error', error, 'countries', countries
|
|
) AS summary
|
|
FROM crawl_run
|
|
UNION ALL
|
|
SELECT
|
|
'admin_action'::text AS type, id, action AS subtype, 'done'::text AS status,
|
|
admin_username AS actor, created_at AS ts, created_at AS finished_at,
|
|
jsonb_build_object(
|
|
'target_table', target_table, 'target_id', target_id,
|
|
'before', before_data, 'after', after_data
|
|
) AS summary
|
|
FROM admin_audit_log
|
|
`;
|
|
|
|
const where = [];
|
|
const vals = [];
|
|
let i = 1;
|
|
if (type) { where.push(`type = $${i}`); vals.push(String(type)); i++; }
|
|
if (status) { where.push(`status = $${i}`); vals.push(String(status)); i++; }
|
|
const whereSql = where.length ? `WHERE ${where.join(" AND ")}` : "";
|
|
|
|
const countSql = `SELECT count(*)::int AS total FROM (${base}) t ${whereSql}`;
|
|
const dataSql = `
|
|
SELECT * FROM (${base}) t
|
|
${whereSql}
|
|
ORDER BY ts DESC
|
|
LIMIT $${i} OFFSET $${i + 1}
|
|
`;
|
|
|
|
const [countRes, dataRes] = await Promise.all([
|
|
pool.query(countSql, vals),
|
|
pool.query(dataSql, [...vals, pageSize, offset]),
|
|
]);
|
|
|
|
return { ok: true, items: dataRes.rows, total: countRes.rows[0].total, page, pageSize };
|
|
} catch (err) {
|
|
fastify.log.error(err);
|
|
return reply.code(500).send({ ok: false, error: "Erreur activité" });
|
|
}
|
|
});
|
|
}
|