diff --git a/apps/api/migrations/013_geocode_dirty_trigger.sql b/apps/api/migrations/013_geocode_dirty_trigger.sql new file mode 100644 index 0000000..7c83e7d --- /dev/null +++ b/apps/api/migrations/013_geocode_dirty_trigger.sql @@ -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(); diff --git a/apps/crawler/src/config.py b/apps/crawler/src/config.py index 8efc2e1..467c387 100644 --- a/apps/crawler/src/config.py +++ b/apps/crawler/src/config.py @@ -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" diff --git a/apps/crawler/src/main.py b/apps/crawler/src/main.py index 4d2efbe..dadc431 100644 --- a/apps/crawler/src/main.py +++ b/apps/crawler/src/main.py @@ -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) + 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_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,9 +74,19 @@ 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") - run_full_crawl(ma) + 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) # ------------------------------------------------------------------ # Planification @@ -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: diff --git a/apps/geocoder/src/enqueue.py b/apps/geocoder/src/enqueue.py index d0437f0..9f5c1b2 100644 --- a/apps/geocoder/src/enqueue.py +++ b/apps/geocoder/src/enqueue.py @@ -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. """ @@ -16,8 +26,9 @@ from parser import ( COUNTRY_NAME_TO_ISO2, ) -POLL_INTERVAL = int(os.environ.get("GEOCODE_ENQUEUE_POLL", "15")) -BATCH = int(os.environ.get("GEOCODE_ENQUEUE_BATCH", "100000")) +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() diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index b97cf34..b1d2cb8 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -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 diff --git a/docker-compose.yml b/docker-compose.yml index 76237c8..ea31132 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -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