From ca45697207032d0c8d246d4362bbabe21018445e Mon Sep 17 00:00:00 2001 From: Nicolas Fryder Date: Tue, 30 Jun 2026 21:05:17 +0200 Subject: [PATCH] =?UTF-8?q?feat(crawler):=20priority=20queue=20+=20d=C3=A9?= =?UTF-8?q?lais=20r=C3=A9duits=20(1.5-2.5s)=20+=20limit=20500?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - BAND_MIN_DELAY: 2.5→1.5s, BAND_MAX_DELAY: 5.0→2.5s (avg 2s/band) - ENRICH_LIMIT: 200→500 (configurable via CRAWLER_ENRICH_LIMIT) - get_bands_to_enrich: remplace la queue FIFO simple par une file de priorité à 4 niveaux : 1. Nouveaux bands (band_page absent) 2. Modifiés depuis dernier enrichissement (updated_at > crawled_at + 1min) 3. Héritage ancien scraper (crawled_at IS NULL, band_page présent) 4. Stale (crawled_at < now() - 30 days) - Suppression de get_bands_stale() (logique absorbée par la queue) Objectif : ~17 jours pour réenrichir les 103k bands (6000 bands/jour à raison de 500/run × 12 runs/24h) Co-Authored-By: Claude Sonnet 4.6 --- apps/crawler/src/config.py | 5 ++-- apps/crawler/src/db.py | 47 +++++++++++++++++++++----------------- apps/crawler/src/jobs.py | 2 +- apps/crawler/src/main.py | 3 ++- 4 files changed, 32 insertions(+), 25 deletions(-) diff --git a/apps/crawler/src/config.py b/apps/crawler/src/config.py index b063018..8efc2e1 100644 --- a/apps/crawler/src/config.py +++ b/apps/crawler/src/config.py @@ -6,8 +6,9 @@ FLARESOLVERR_URL = os.getenv("FLARESOLVERR_URL", "http://flaresolverr:8191") # Delays between requests (seconds) LIST_MIN_DELAY = float(os.getenv("CRAWLER_LIST_MIN_DELAY", "2.0")) LIST_MAX_DELAY = float(os.getenv("CRAWLER_LIST_MAX_DELAY", "4.0")) -BAND_MIN_DELAY = float(os.getenv("CRAWLER_BAND_MIN_DELAY", "2.5")) -BAND_MAX_DELAY = float(os.getenv("CRAWLER_BAND_MAX_DELAY", "5.0")) +BAND_MIN_DELAY = float(os.getenv("CRAWLER_BAND_MIN_DELAY", "1.5")) +BAND_MAX_DELAY = float(os.getenv("CRAWLER_BAND_MAX_DELAY", "2.5")) +ENRICH_LIMIT = int(os.getenv("CRAWLER_ENRICH_LIMIT", "500")) # Cooldown pause every N band pages COOLDOWN_EVERY = int(os.getenv("CRAWLER_COOLDOWN_EVERY", "60")) COOLDOWN_MIN = float(os.getenv("CRAWLER_COOLDOWN_MIN", "15")) diff --git a/apps/crawler/src/db.py b/apps/crawler/src/db.py index 8036552..7952d00 100644 --- a/apps/crawler/src/db.py +++ b/apps/crawler/src/db.py @@ -135,18 +135,39 @@ def upsert_band_enriched(ma_id: int, data: Dict[str, Any], html_hash: str) -> bo def get_bands_to_enrich(country: Optional[str] = None, limit: int = 50) -> List[Dict]: - """Bands qui n'ont jamais été enrichies (data->'band_page' IS NULL).""" + """ + File de priorité pour l'enrichissement : + 1. Nouveaux bands (band_page absent) — jamais enrichis + 2. Modifiés depuis le dernier enrichissement (updated_at > crawled_at + 1 min) + 3. Héritage ancien scraper (crawled_at IS NULL, band_page présent) + 4. Stale (enrichis il y a > 30 jours par le système actuel) + """ sql = """ SELECT ma_id, data->>'url' AS url, name, country FROM bands - WHERE data->'band_page' IS NULL - AND data->>'url' IS NOT NULL + WHERE data->>'url' IS NOT NULL + AND ( + data->'band_page' IS NULL + OR crawled_at IS NULL + OR (crawled_at IS NOT NULL AND updated_at > crawled_at + interval '1 minute') + OR crawled_at < now() - interval '30 days' + ) """ - params = [] + params: list = [] if country: sql += " AND country = %s" params.append(country) - sql += " ORDER BY first_seen_at ASC NULLS LAST LIMIT %s" + sql += """ + ORDER BY + CASE + WHEN data->'band_page' IS NULL THEN 1 + WHEN crawled_at IS NOT NULL AND updated_at > crawled_at + interval '1 minute' THEN 2 + WHEN crawled_at IS NULL THEN 3 + ELSE 4 + END ASC, + updated_at DESC NULLS LAST + LIMIT %s + """ params.append(limit) with get_conn() as conn: @@ -155,22 +176,6 @@ def get_bands_to_enrich(country: Optional[str] = None, limit: int = 50) -> List[ return [dict(r) for r in cur.fetchall()] -def get_bands_stale(hours: int = 24 * 30, limit: int = 100) -> List[Dict]: - """Bands enrichies il y a plus de N heures (pour re-crawl périodique).""" - sql = """ - SELECT ma_id, data->>'url' AS url, name, country - FROM bands - WHERE enriched = true - AND (crawled_at IS NULL OR crawled_at < now() - make_interval(hours => %s)) - ORDER BY crawled_at ASC NULLS FIRST - LIMIT %s - """ - with get_conn() as conn: - with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: - cur.execute(sql, (hours, limit)) - return [dict(r) for r in cur.fetchall()] - - # ------------------------------------------------------------------ # Crawl run tracking # ------------------------------------------------------------------ diff --git a/apps/crawler/src/jobs.py b/apps/crawler/src/jobs.py index b573353..79cee0f 100644 --- a/apps/crawler/src/jobs.py +++ b/apps/crawler/src/jobs.py @@ -12,7 +12,7 @@ from .config import ( ) from .db import ( upsert_bands, upsert_band_enriched, - get_bands_to_enrich, get_bands_stale, + get_bands_to_enrich, start_crawl_run, finish_crawl_run, get_checkpoint, set_checkpoint, ) diff --git a/apps/crawler/src/main.py b/apps/crawler/src/main.py index ba0a962..de04a48 100644 --- a/apps/crawler/src/main.py +++ b/apps/crawler/src/main.py @@ -24,6 +24,7 @@ import schedule from .config import ( FLARESOLVERR_URL, FS_TIMEOUT_MS, SCHED_INCREMENTAL_H, SCHED_ENRICH_H, SCHED_FULL_DAY, + ENRICH_LIMIT, ) from .flaresolverr import FlareSolverr, FlareSolverrError from .jobs import run_full_crawl, run_incremental, run_enrich @@ -67,7 +68,7 @@ def main(): def job_enrich(): log.info("[scheduler] → enrich") - run_enrich(ma, limit=200) + run_enrich(ma, limit=ENRICH_LIMIT) def job_full(): if SCHED_FULL_DAY and datetime.now(timezone.utc).day == SCHED_FULL_DAY: