metalfrom.eu/apps/crawler/src/jobs.py

"""
Jobs de crawl. Chaque fonction est un job indépendant appelé par le scheduler.
"""
import logging
from datetime import UTC, datetime

from .config import (
    BAND_MAX_DELAY,
    BAND_MIN_DELAY,
    COOLDOWN_EVERY,
    COOLDOWN_MAX,
    COOLDOWN_MIN,
    MA_BASE,
)
from .db import (
    RunCancelled,
    finish_crawl_run,
    get_bands_to_enrich,
    get_checkpoint,
    log_event,
    raise_if_cancelled,
    set_checkpoint,
    start_crawl_run,
    update_crawl_run_progress,
    upsert_band_enriched,
    upsert_bands,
)
from .europe_codes import EUROPE_COUNTRY_CODES
from .ma_http import MASession
from .polite import cooldown, sleep_range
from .scraper_band import page_hash, parse_band_page

_EU_SET = set(EUROPE_COUNTRY_CODES)

log = logging.getLogger(__name__)

CHUNK = 300  # taille des batches d'upsert


# ------------------------------------------------------------------
# Crawl complet (tous les pays d'un coup)
# ------------------------------------------------------------------

def run_full_crawl(session: MASession, countries: list[str] = None):
    countries = countries or EUROPE_COUNTRY_CODES
    run_id = start_crawl_run("full_europe", countries)
    stats = {"seen": 0, "new": 0, "updated": 0, "enriched": 0}
    error = None
    cancelled = False
    log_event("info", f"full crawl started ({len(countries)} countries)", run_id=run_id)

    try:
        for cc in countries:
            raise_if_cancelled(run_id)
            log.info(f"[full] country={cc}")
            buf = []
            for band in session.fetch_country_bands(cc):
                band["data"] = {"url": band.pop("url", None)}
                buf.append(band)
                if len(buf) >= CHUNK:
                    # Un crawl complet Europe dure des heures : le contrôle doit
                    # tomber à chaque lot, pas seulement entre deux pays.
                    raise_if_cancelled(run_id)
                    r = upsert_bands(buf)
                    stats["seen"]    += len(buf)
                    stats["new"]     += r["inserted"]
                    stats["updated"] += r["updated"]
                    buf = []
            if buf:
                r = upsert_bands(buf)
                stats["seen"]    += len(buf)
                stats["new"]     += r["inserted"]
                stats["updated"] += r["updated"]

        set_checkpoint("last_full_crawl_at", _now_iso())
        log.info(f"[full] done: {stats}")
        log_event("info", f"full crawl done: {stats}", run_id=run_id)
    except RunCancelled:
        cancelled = True
        log.info(f"[full] annulé sur demande admin: {stats}")
        log_event("warning", f"full crawl annulé par un admin: {stats}", run_id=run_id)
    except Exception as e:
        error = str(e)
        log.error(f"[full] error: {e}", exc_info=True)
        log_event("error", f"full crawl error: {e}", run_id=run_id)
    finally:
        finish_crawl_run(run_id, stats, error, cancelled=cancelled)


# ------------------------------------------------------------------
# Crawl incrémental (dernières additions / modifications)
# ------------------------------------------------------------------

def run_incremental(session: MASession, order_by: str):
    """
    order_by : 'created' | 'modified'
    S'arrête dès qu'on voit des entrées déjà connues (via le checkpoint).
    """
    checkpoint_key = f"last_{order_by}_check"
    since = get_checkpoint(checkpoint_key)
    run_id = start_crawl_run(f"incremental_{order_by}")
    stats = {"seen": 0, "new": 0, "updated": 0, "enriched": 0}
    error = None
    cancelled = False
    latest_date = since

    log.info(f"[incr/{order_by}] since={since}")
    log_event("info", f"incremental/{order_by} started (since={since})", run_id=run_id)
    try:
        buf = []
        for band in session.fetch_archive_bands(order_by, since_date=since):
            date_str = band.pop("date_str", None)
            if date_str and (latest_date is None or date_str > latest_date):
                latest_date = date_str

            if band.get("country") not in _EU_SET:
                continue  # on reste europe uniquement

            band["data"] = {"url": band.pop("url", None)}
            buf.append(band)
            if len(buf) >= CHUNK:
                raise_if_cancelled(run_id)
                r = upsert_bands(buf)
                stats["seen"]    += len(buf)
                stats["new"]     += r["inserted"]
                stats["updated"] += r["updated"]
                buf = []

        if buf:
            r = upsert_bands(buf)
            stats["seen"]    += len(buf)
            stats["new"]     += r["inserted"]
            stats["updated"] += r["updated"]

        if latest_date:
            set_checkpoint(checkpoint_key, latest_date)
        log.info(f"[incr/{order_by}] done: {stats}")
        log_event("info", f"incremental/{order_by} done: {stats}", run_id=run_id)
    except RunCancelled:
        cancelled = True
        log.info(f"[incr/{order_by}] annulé sur demande admin: {stats}")
        log_event("warning", f"incremental/{order_by} annulé par un admin: {stats}", run_id=run_id)
    except Exception as e:
        error = str(e)
        log.error(f"[incr/{order_by}] error: {e}", exc_info=True)
        log_event("error", f"incremental/{order_by} error: {e}", run_id=run_id)
    finally:
        finish_crawl_run(run_id, stats, error, cancelled=cancelled)


# ------------------------------------------------------------------
# Enrichissement des pages individuelles de bands
# ------------------------------------------------------------------

def run_enrich(session: MASession, limit: int = 100, country: str | None = None):
    """
    Visite les pages individuelles des bands non encore enrichies.
    Stocke themes, membres, label, dates MA, hash HTML.
    """
    run_id = start_crawl_run("enrich", [country] if country else None)
    stats = {"seen": 0, "new": 0, "updated": 0, "enriched": 0}
    error = None
    cancelled = False

    try:
        bands = get_bands_to_enrich(country=country, limit=limit)
        log.info(f"[enrich] {len(bands)} bands to enrich")
        log_event("info", f"enrich started: {len(bands)} bands queued", run_id=run_id)

        for i, band in enumerate(bands):
            # Volontairement AVANT le try/except interne : celui-ci fait
            # `except Exception: continue` et avalerait RunCancelled.
            raise_if_cancelled(run_id)

            url = band.get("url")
            if not url:
                continue
            if not url.startswith("http"):
                url = MA_BASE + url

            try:
                html = session.get_html(url)
                data = parse_band_page(html)
                h    = page_hash(html)
                ok   = upsert_band_enriched(band["ma_id"], data, h)
                if ok:
                    stats["enriched"] += 1
                stats["seen"] += 1
            except Exception as e:
                log.warning(f"[enrich] failed ma_id={band['ma_id']}: {e}")
                log_event("warning", f"enrich failed: {e}", run_id=run_id, ma_id=band["ma_id"])
                continue

            sleep_range(BAND_MIN_DELAY, BAND_MAX_DELAY)
            if COOLDOWN_EVERY and (i + 1) % COOLDOWN_EVERY == 0:
                cooldown(COOLDOWN_MIN, COOLDOWN_MAX)

            if (i + 1) % 10 == 0:
                update_crawl_run_progress(run_id, stats)

        log.info(f"[enrich] done: {stats}")
        log_event("info", f"enrich done: {stats}", run_id=run_id)
    except RunCancelled:
        cancelled = True
        log.info(f"[enrich] annulé sur demande admin: {stats}")
        log_event("warning", f"enrich annulé par un admin: {stats}", run_id=run_id)
    except Exception as e:
        error = str(e)
        log.error(f"[enrich] error: {e}", exc_info=True)
        log_event("error", f"enrich error: {e}", run_id=run_id)
    finally:
        finish_crawl_run(run_id, stats, error, cancelled=cancelled)


# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------

def _now_iso() -> str:
    return datetime.now(UTC).strftime("%Y-%m-%d")