nico/metalfrom.eu
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

feat(geocoder): seuil de confiance strict, dedup, purge ancien pipeline

Browse source 
Étape 0 de la refonte pipeline crawl/géocodage.

Auto-requeue du géocodage
- migration 013: trigger trg_bands_geocode_dirty (AFTER UPDATE OF location_text
  ON bands) supprime les band_locations d'un band dès que sa localisation
  change réellement (IS DISTINCT FROM) — plus de points fantômes après un
  déménagement détecté par le crawler incrémental.
- enqueue.py: en plus du déclenchement manuel, un scan automatique tourne
  toutes les ENQUEUE_AUTO_INTERVAL_MIN minutes (défaut 60) et rattrape les
  bands rendus "dirty" par le trigger (idempotent, ON CONFLICT DO NOTHING —
  pas de canal de notification supplémentaire nécessaire). Chaque exécution
  (manuelle ou auto) crée une ligne crawl_run + logs rattachés, en miroir de
  crawler/src/db.py, pour apparaître dans l'activité admin à venir.

Crawl complet en calcul glissant
- remplace CRAWLER_SCHED_FULL_DAY (jour fixe du mois) par
  CRAWLER_FULL_CRAWL_INTERVAL_DAYS (défaut 60) basé sur le checkpoint
  last_full_crawl_at — robuste au calendrier et aux redémarrages (vérifié
  aussi au démarrage, pas seulement le tick quotidien 03:00 UTC).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Nicolas Fryder 2026-07-02 21:07:16 +02:00
parent 88f6f2bff9
commit 1ae5aa4b57
6 changed files with 160 additions and 60 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
apps/api/migrations/013_geocode_dirty_trigger.sql Normal file
View file

@ -0,0 +1,25 @@
-- 013_geocode_dirty_trigger.sql
-- Auto-requeue du géocodage quand location_text change en DB.
--
-- Avant ce trigger, un changement de localisation (déménagement détecté par le
-- crawler incrémental) laissait les anciennes band_locations en place — rien ne
-- les invalidait. On supprime les band_locations du band concerné dès que son
-- location_text change réellement ; le daemon enqueue (déclenchement manuel +
-- scan périodique de rattrapage, voir enqueue.py) reparse le texte et recrée
-- les steps. Resync complet (pas de diff fin) : simple et sûr, le cache
-- (geocode_cache + le fast-path dedup par (lieu,pays) dans band_locations)
-- absorbe le coût pour les lieux déjà vus ailleurs.
CREATE OR REPLACE FUNCTION bands_geocode_dirty() RETURNS trigger AS $$
BEGIN
IF TG_OP = 'UPDATE' AND OLD.location_text IS DISTINCT FROM NEW.location_text THEN
DELETE FROM band_locations WHERE ma_id = NEW.ma_id;
END IF;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
DROP TRIGGER IF EXISTS trg_bands_geocode_dirty ON bands;
CREATE TRIGGER trg_bands_geocode_dirty
AFTER UPDATE OF location_text ON bands
FOR EACH ROW EXECUTE FUNCTION bands_geocode_dirty();

5
apps/crawler/src/config.py
View file

@ -23,7 +23,8 @@ AJAX_PAGE_SIZE = int(os.getenv("CRAWLER_AJAX_PAGE_SIZE", "500"))
# Schedules (hours between runs, 0 = disabled)
SCHED_INCREMENTAL_H = int(os.getenv("CRAWLER_SCHED_INCREMENTAL_H", "4"))
SCHED_ENRICH_H = int(os.getenv("CRAWLER_SCHED_ENRICH_H", "1"))
# Full crawl: day of month (1-28), 0 = disabled
SCHED_FULL_DAY = int(os.getenv("CRAWLER_SCHED_FULL_DAY", "1"))
# Full crawl : filet de sécurité, calcul glissant depuis le dernier crawl complet
# (checkpoint last_full_crawl_at), vérifié chaque jour à 03:00 UTC. 0 = désactivé.
FULL_CRAWL_INTERVAL_DAYS = int(os.getenv("CRAWLER_FULL_CRAWL_INTERVAL_DAYS", "60"))
MA_BASE = "https://www.metal-archives.com"

33
apps/crawler/src/main.py
View file

@ -4,14 +4,16 @@ Point d'entrée du crawler. Lance le scheduler et exécute les jobs périodiques
Jobs :
- Toutes les SCHED_INCREMENTAL_H heures : crawl incrémental (additions + modifications)
- Toutes les SCHED_ENRICH_H heures : enrichissement des bands non enrichies
- Le SCHED_FULL_DAY du mois : crawl complet Europe
- Filet de sécurité (calcul glissant) : crawl complet Europe si plus de
CRAWLER_FULL_CRAWL_INTERVAL_DAYS jours depuis le dernier (checkpoint
last_full_crawl_at), vérifié chaque jour à 03:00 UTC + au démarrage
Variables d'env :
DATABASE_URL (obligatoire)
FLARESOLVERR_URL (défaut: http://flaresolverr:8191)
CRAWLER_SCHED_INCREMENTAL_H (défaut: 4)
CRAWLER_SCHED_ENRICH_H (défaut: 1)
CRAWLER_SCHED_FULL_DAY (défaut: 1, mettre 0 pour désactiver)
CRAWLER_FULL_CRAWL_INTERVAL_DAYS (défaut: 60, mettre 0 pour désactiver)
"""
import logging
import os
@ -23,11 +25,11 @@ import schedule
from .config import (
FLARESOLVERR_URL, FS_TIMEOUT_MS,
SCHED_INCREMENTAL_H, SCHED_ENRICH_H, SCHED_FULL_DAY,
SCHED_INCREMENTAL_H, SCHED_ENRICH_H, FULL_CRAWL_INTERVAL_DAYS,
ENRICH_LIMIT,
)
from .flaresolverr import FlareSolverr, FlareSolverrError
from .db import claim_job_trigger, finish_job_trigger
from .db import claim_job_trigger, finish_job_trigger, get_checkpoint
from .jobs import run_full_crawl, run_incremental, run_enrich
from .ma_http import MASession
@ -72,8 +74,18 @@ def main():
run_enrich(ma, limit=ENRICH_LIMIT)
def job_full():
if SCHED_FULL_DAY and datetime.now(timezone.utc).day == SCHED_FULL_DAY:
log.info("[scheduler] → full Europe crawl")
if not FULL_CRAWL_INTERVAL_DAYS:
return
last = get_checkpoint("last_full_crawl_at")
if last:
try:
last_dt = datetime.strptime(last, "%Y-%m-%d").replace(tzinfo=timezone.utc)
elapsed_days = (datetime.now(timezone.utc) - last_dt).days
if elapsed_days < FULL_CRAWL_INTERVAL_DAYS:
return
except ValueError:
pass # checkpoint mal formé, on relance par sécurité
log.info(f"[scheduler] → full Europe crawl (filet {FULL_CRAWL_INTERVAL_DAYS}j écoulé)")
run_full_crawl(ma)
# ------------------------------------------------------------------
@ -88,15 +100,16 @@ def main():
schedule.every(SCHED_ENRICH_H).hours.do(job_enrich)
log.info(f"[main] enrich scheduled every {SCHED_ENRICH_H}h")
if SCHED_FULL_DAY > 0:
# Vérifier chaque jour à 03:00 UTC si c'est le bon jour
if FULL_CRAWL_INTERVAL_DAYS > 0:
# Vérifié chaque jour à 03:00 UTC ; ne se déclenche que si l'intervalle est écoulé
schedule.every().day.at("03:00").do(job_full)
log.info(f"[main] full crawl scheduled day={SCHED_FULL_DAY} of each month at 03:00 UTC")
log.info(f"[main] full crawl: filet de {FULL_CRAWL_INTERVAL_DAYS}j, vérifié chaque jour à 03:00 UTC")
# Premier run au démarrage
# Premier run au démarrage (job_full rattrape un redémarrage qui aurait loupé la fenêtre)
log.info("[main] running initial jobs at startup")
job_incremental()
job_enrich()
job_full()
log.info("[main] entering scheduler loop")
while True:

139
apps/geocoder/src/enqueue.py
View file

@ -1,8 +1,18 @@
"""
enqueue.py — Daemon qui attend des job_triggers de type 'geocoder_enqueue'
et peuple band_locations depuis bands.location_text.
enqueue.py — Daemon qui peuple band_locations depuis bands.location_text.
Deux déclencheurs :
- manuel : job_triggers de type 'geocoder_enqueue'
(POST /admin/api/job-triggers {job_type: "geocoder_enqueue"})
- automatique : toutes les ENQUEUE_AUTO_INTERVAL_MIN minutes, en rattrapage.
Le trigger DB trg_bands_geocode_dirty (migration 013) supprime les
band_locations d'un band dès que son location_text change ; ce scan
périodique idempotent (ON CONFLICT DO NOTHING) les retrouve et les
reparse sans canal de notification supplémentaire.
Chaque exécution (manuelle ou auto) crée une ligne crawl_run pour apparaître
dans l'activité admin avec ses logs, en miroir de crawler/src/db.py.
Déclenché depuis l'UI admin → POST /admin/api/job-triggers {job_type: "geocoder_enqueue"}
Tourne en continu (restart: unless-stopped), poll toutes les 15s.
"""
@ -18,6 +28,7 @@ from parser import (
POLL_INTERVAL = int(os.environ.get("GEOCODE_ENQUEUE_POLL", "15"))
BATCH = int(os.environ.get("GEOCODE_ENQUEUE_BATCH", "100000"))
AUTO_INTERVAL_MIN = int(os.environ.get("ENQUEUE_AUTO_INTERVAL_MIN", "60"))
def resolve_iso2(location_raw: str, band_country: str | None) -> str | None:
@ -32,8 +43,9 @@ def resolve_iso2(location_raw: str, band_country: str | None) -> str | None:
return None
def run_enqueue(cur) -> tuple[int, int, int]:
"""Parse tous les bands et peuple band_locations. Retourne (inserted, country_fixed, skipped)."""
def run_enqueue(cur) -> tuple[int, int, int, int]:
"""Parse tous les bands et peuple band_locations.
Retourne (inserted, country_fixed, skipped, failed)."""
cur.execute(
"SELECT ma_id, country, location_text FROM bands "
"WHERE location_text IS NOT NULL AND location_text <> '' "
@ -111,16 +123,95 @@ def run_enqueue(cur) -> tuple[int, int, int]:
return inserted, country_fixed, skipped, failed
# ------------------------------------------------------------------
# crawl_run wrapping — même table/forme que apps/crawler/src/db.py, pour que
# chaque exécution de l'enqueue apparaisse dans l'activité admin avec ses logs.
# ------------------------------------------------------------------
def _start_crawl_run(cur) -> int:
cur.execute(
"INSERT INTO crawl_run (run_type) VALUES ('geocoder_enqueue') RETURNING id"
)
return cur.fetchone()[0]
def _finish_crawl_run(cur, run_id: int, stats: dict, error: str | None):
status = "error" if error else "done"
cur.execute(
"""
UPDATE crawl_run SET
status=%s, finished_at=now(),
bands_seen=%s, bands_new=%s, bands_updated=%s, error=%s
WHERE id=%s
""",
(status, stats.get("seen", 0), stats.get("new", 0),
stats.get("skipped", 0), error, run_id),
)
def _log(cur, run_id: int, level: str, message: str):
cur.execute(
"INSERT INTO crawl_log (run_id, level, message) VALUES (%s, %s, %s)",
(run_id, level, message[:2000]),
)
def _execute_run(cur, trigger_label: str, job_id: int | None = None) -> None:
"""Exécute run_enqueue() une fois, entourée d'un crawl_run + logs.
trigger_label : 'manuel' ou 'auto' (juste pour le message de log)."""
run_id = _start_crawl_run(cur)
_log(cur, run_id, "info", f"[geocoder_enqueue] démarré ({trigger_label})")
print(f"[enqueue] run #{run_id} démarré ({trigger_label})")
error = None
try:
inserted, country_fixed, skipped, failed = run_enqueue(cur)
summary = (
f"[geocoder_enqueue] terminé ({trigger_label}) — "
f"inserted={inserted} country_fixed={country_fixed} "
f"skipped={skipped} failed={failed}"
)
_log(cur, run_id, "info", summary)
print(f"[enqueue] {summary}")
_finish_crawl_run(cur, run_id, {
"seen": inserted + country_fixed, "new": inserted, "skipped": skipped,
}, None)
except Exception as exc:
error = str(exc)[:400]
_log(cur, run_id, "error", f"[geocoder_enqueue] erreur ({trigger_label}): {error}")
print(f"[enqueue] run #{run_id} erreur: {error}")
_finish_crawl_run(cur, run_id, {}, error)
if job_id is not None:
if error:
cur.execute(
"UPDATE job_triggers SET status='error', finished_at=now(), error=%s WHERE id=%s",
(error, job_id),
)
else:
cur.execute(
"UPDATE job_triggers SET status='done', finished_at=now() WHERE id=%s",
(job_id,),
)
def main():
dsn = os.environ["DATABASE_URL"]
conn = psycopg2.connect(dsn)
conn.autocommit = True
print(f"[enqueue] daemon démarré, poll toutes les {POLL_INTERVAL}s")
print(f"[enqueue] daemon démarré, poll {POLL_INTERVAL}s, auto toutes les {AUTO_INTERVAL_MIN}min")
last_auto = time.monotonic()
with conn.cursor() as cur:
while True:
# Chercher un job en attente
# Rattrapage automatique (bands rendus 'dirty' par le trigger DB)
if time.monotonic() - last_auto >= AUTO_INTERVAL_MIN * 60:
_execute_run(cur, "auto")
last_auto = time.monotonic()
# Déclenchement manuel via l'admin
cur.execute(
"""
SELECT id FROM job_triggers
@ -141,39 +232,7 @@ def main():
"UPDATE job_triggers SET status='running', started_at=now() WHERE id=%s",
(job_id,),
)
cur.execute(
"INSERT INTO crawl_log (level, message) VALUES ('info', %s)",
(f"[geocoder_enqueue] job #{job_id} démarré",),
)
print(f"[enqueue] job #{job_id} démarré")
try:
inserted, country_fixed, skipped, failed = run_enqueue(cur)
summary = (
f"[geocoder_enqueue] job #{job_id} terminé — "
f"inserted={inserted} country_fixed={country_fixed} "
f"skipped={skipped} failed={failed}"
)
cur.execute(
"UPDATE job_triggers SET status='done', finished_at=now() WHERE id=%s",
(job_id,),
)
cur.execute(
"INSERT INTO crawl_log (level, message) VALUES ('info', %s)",
(summary,),
)
print(f"[enqueue] {summary}")
except Exception as exc:
err = str(exc)[:400]
cur.execute(
"UPDATE job_triggers SET status='error', finished_at=now(), error=%s WHERE id=%s",
(err, job_id),
)
cur.execute(
"INSERT INTO crawl_log (level, message) VALUES ('error', %s)",
(f"[geocoder_enqueue] job #{job_id} erreur: {err}",),
)
print(f"[enqueue] job #{job_id} erreur: {err}")
_execute_run(cur, "manuel", job_id=job_id)
conn.close()

3
docker-compose.dev.yml
View file

@ -26,7 +26,7 @@ services:
FLARESOLVERR_URL: http://flaresolverr:8191
CRAWLER_SCHED_INCREMENTAL_H: "4"
CRAWLER_SCHED_ENRICH_H: "2"
CRAWLER_SCHED_FULL_DAY: "0"
CRAWLER_FULL_CRAWL_INTERVAL_DAYS: "0"
depends_on:
- flaresolverr
networks:
@ -43,6 +43,7 @@ services:
DATABASE_URL: ${DATABASE_URL}
GEOCODE_ENQUEUE_BATCH: "100000"
GEOCODE_ENQUEUE_POLL: "15"
ENQUEUE_AUTO_INTERVAL_MIN: "60"
command: ["python", "src/enqueue.py"]
networks:
- coolify

1
docker-compose.yml
View file

@ -15,6 +15,7 @@ services:
DATABASE_URL: ${DATABASE_URL}
GEOCODE_ENQUEUE_BATCH: "100000"
GEOCODE_ENQUEUE_POLL: "15"
ENQUEUE_AUTO_INTERVAL_MIN: "60"
command: ["python", "src/enqueue.py"]
networks:
- coolify