feat(crawler): priority queue + délais réduits (1.5-2.5s) + limit 500

- 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 <noreply@anthropic.com>
This commit is contained in:
Nicolas Fryder 2026-06-30 21:05:17 +02:00
parent 94d94d926b
commit ca45697207
4 changed files with 32 additions and 25 deletions

View file

@ -6,8 +6,9 @@ FLARESOLVERR_URL = os.getenv("FLARESOLVERR_URL", "http://flaresolverr:8191")
# Delays between requests (seconds) # Delays between requests (seconds)
LIST_MIN_DELAY = float(os.getenv("CRAWLER_LIST_MIN_DELAY", "2.0")) LIST_MIN_DELAY = float(os.getenv("CRAWLER_LIST_MIN_DELAY", "2.0"))
LIST_MAX_DELAY = float(os.getenv("CRAWLER_LIST_MAX_DELAY", "4.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_MIN_DELAY = float(os.getenv("CRAWLER_BAND_MIN_DELAY", "1.5"))
BAND_MAX_DELAY = float(os.getenv("CRAWLER_BAND_MAX_DELAY", "5.0")) 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 pause every N band pages
COOLDOWN_EVERY = int(os.getenv("CRAWLER_COOLDOWN_EVERY", "60")) COOLDOWN_EVERY = int(os.getenv("CRAWLER_COOLDOWN_EVERY", "60"))
COOLDOWN_MIN = float(os.getenv("CRAWLER_COOLDOWN_MIN", "15")) COOLDOWN_MIN = float(os.getenv("CRAWLER_COOLDOWN_MIN", "15"))

View file

@ -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]: 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 = """ sql = """
SELECT ma_id, data->>'url' AS url, name, country SELECT ma_id, data->>'url' AS url, name, country
FROM bands FROM bands
WHERE data->'band_page' IS NULL WHERE data->>'url' IS NOT NULL
AND 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: if country:
sql += " AND country = %s" sql += " AND country = %s"
params.append(country) 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) params.append(limit)
with get_conn() as conn: 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()] 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 # Crawl run tracking
# ------------------------------------------------------------------ # ------------------------------------------------------------------

View file

@ -12,7 +12,7 @@ from .config import (
) )
from .db import ( from .db import (
upsert_bands, upsert_band_enriched, upsert_bands, upsert_band_enriched,
get_bands_to_enrich, get_bands_stale, get_bands_to_enrich,
start_crawl_run, finish_crawl_run, start_crawl_run, finish_crawl_run,
get_checkpoint, set_checkpoint, get_checkpoint, set_checkpoint,
) )

View file

@ -24,6 +24,7 @@ import schedule
from .config import ( from .config import (
FLARESOLVERR_URL, FS_TIMEOUT_MS, FLARESOLVERR_URL, FS_TIMEOUT_MS,
SCHED_INCREMENTAL_H, SCHED_ENRICH_H, SCHED_FULL_DAY, SCHED_INCREMENTAL_H, SCHED_ENRICH_H, SCHED_FULL_DAY,
ENRICH_LIMIT,
) )
from .flaresolverr import FlareSolverr, FlareSolverrError from .flaresolverr import FlareSolverr, FlareSolverrError
from .jobs import run_full_crawl, run_incremental, run_enrich from .jobs import run_full_crawl, run_incremental, run_enrich
@ -67,7 +68,7 @@ def main():
def job_enrich(): def job_enrich():
log.info("[scheduler] → enrich") log.info("[scheduler] → enrich")
run_enrich(ma, limit=200) run_enrich(ma, limit=ENRICH_LIMIT)
def job_full(): def job_full():
if SCHED_FULL_DAY and datetime.now(timezone.utc).day == SCHED_FULL_DAY: if SCHED_FULL_DAY and datetime.now(timezone.utc).day == SCHED_FULL_DAY: