spb/ka4 Public

1# Author: Simon-Pierre Boucher2# Contact: contact@spboucher.ai3"""Client REST Firecrawl : search, map, scrape.45Le respect du robots.txt et le débit sont gérés en amont (voir src/robots.py et6src/scraper.py). Ce client reste une couche transport minimale.7"""89from __future__ import annotations1011import time12from typing import Any, Optional1314import requests1516from .util import RateLimiter171819class FirecrawlClient:20    # ka4 : fail-fast pour ne pas bloquer le crawl concurrent21    def __init__(self, api_key: str, base_url: str = "https://api.firecrawl.dev", delay: float = 1.5,22                 timeout: int = 45, retries: int = 2):23        self.base = base_url.rstrip("/")24        self.session = requests.Session()25        self.session.headers.update(26            {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}27        )28        self.limiter = RateLimiter(delay)29        self.timeout = timeout30        self.retries = retries3132    def _post(self, path: str, payload: dict[str, Any], retries: Optional[int] = None) -> dict[str, Any]:33        url = f"{self.base}{path}"34        last_err: Optional[Exception] = None35        for attempt in range(retries or self.retries):36            self.limiter.wait()37            try:38                r = self.session.post(url, json=payload, timeout=self.timeout)39                if r.status_code == 429:40                    time.sleep(1.5 * (attempt + 1))41                    continue42                r.raise_for_status()43                return r.json()44            except Exception as e:  # noqa: BLE00145                last_err = e46                time.sleep(1.0 * (attempt + 1))47        raise RuntimeError(f"Firecrawl {path} a échoué: {last_err}")4849    def search(self, query: str, limit: int = 10) -> list[dict[str, Any]]:50        data = self._post("/v1/search", {"query": query, "limit": limit})51        return data.get("data", []) or []5253    def map(self, url: str, search: Optional[str] = None, limit: int = 200) -> list[str]:54        payload: dict[str, Any] = {"url": url, "limit": limit}55        if search:56            payload["search"] = search57        data = self._post("/v1/map", payload)58        links = data.get("links", []) or []59        return [l["url"] if isinstance(l, dict) else l for l in links]6061    def scrape(self, url: str, formats: Optional[list[str]] = None) -> dict[str, Any]:62        # demande aussi les liens -> alimente la frontière agressive63        data = self._post("/v1/scrape", {"url": url, "formats": formats or ["markdown", "links"]})64        d = data.get("data", {}) or {}65        return {66            "markdown": d.get("markdown", "") or "",67            "metadata": d.get("metadata", {}) or {},68            "links": d.get("links", []) or [],69            "backend": "firecrawl",70        }71