# Author: Simon-Pierre Boucher # Contact: contact@spboucher.ai """Dispatcher de scrape multi-backend + garde robots.txt. - search / map : toujours via Firecrawl. - scrape : selon le backend choisi (auto | firecrawl | scrapfly). En mode `auto`, Firecrawl est tenté d'abord, puis Scrapfly en repli si le résultat est vide ou en erreur. """ from __future__ import annotations from typing import Any, Optional from .config import Config from .firecrawl_client import FirecrawlClient from .robots import RobotsCache from .scrapfly_client import ScrapflyClient class Scraper: def __init__(self, cfg: Config): self.cfg = cfg self.firecrawl = FirecrawlClient(cfg.firecrawl_api_key, cfg.firecrawl_base, cfg.request_delay) self.scrapfly = ( ScrapflyClient(cfg.scrapfly_api_key, cfg.scrapfly_base, cfg.request_delay, cfg.scrapfly_render_js) if cfg.scrapfly_api_key else None ) self.robots = RobotsCache(cfg.user_agent, enabled=cfg.respect_robots) self.backend = cfg.scraper_backend # -- découverte (Firecrawl) ------------------------------------------- def search(self, query: str, limit: int = 10) -> list[dict[str, Any]]: return self.firecrawl.search(query, limit) def map(self, url: str, search: Optional[str] = None) -> list[str]: return self.firecrawl.map(url, search) # -- scrape ------------------------------------------------------------ def scrape(self, url: str) -> dict[str, Any]: if not self.robots.allowed(url): return {"blocked": True, "markdown": "", "metadata": {}, "links": [], "backend": None} order = self._order() last_err: Optional[Exception] = None for backend in order: try: data = self._scrape_with(backend, url) if data.get("markdown", "").strip(): return data except Exception as e: # noqa: BLE001 last_err = e if last_err: return {"error": str(last_err), "markdown": "", "metadata": {}, "links": [], "backend": None} return {"markdown": "", "metadata": {}, "links": [], "backend": order[-1] if order else None} def _order(self) -> list[str]: if self.backend == "firecrawl": return ["firecrawl"] if self.backend == "scrapfly": return ["scrapfly"] if self.scrapfly else ["firecrawl"] # auto : Firecrawl puis Scrapfly en repli return ["firecrawl", "scrapfly"] if self.scrapfly else ["firecrawl"] def _scrape_with(self, backend: str, url: str) -> dict[str, Any]: if backend == "scrapfly": if not self.scrapfly: raise RuntimeError("Scrapfly non configuré (SCRAPFLY_API_KEY manquant)") return self.scrapfly.scrape(url) return self.firecrawl.scrape(url)