# Author: Simon-Pierre Boucher # Contact: contact@spboucher.ai """Client REST Scrapfly (https://scrapfly.io) — backend de scrape alternatif. Atouts : rendu JavaScript, contournement anti-bot (ASP), proxies. Utile quand Firecrawl échoue ou renvoie une page vide (SPA, protection anti-scraping). """ from __future__ import annotations import time from typing import Any, Optional import requests from .util import RateLimiter class ScrapflyClient: def __init__(self, api_key: str, base_url: str = "https://api.scrapfly.io", delay: float = 1.5, render_js: bool = False): self.api_key = api_key self.base = base_url.rstrip("/") self.render_js = render_js self.session = requests.Session() self.limiter = RateLimiter(delay) def scrape(self, url: str, render_js: Optional[bool] = None, retries: int = 2) -> dict[str, Any]: params = { "key": self.api_key, "url": url, "format": "markdown", "asp": "true", # Anti Scraping Protection bypass "render_js": "true" if (self.render_js if render_js is None else render_js) else "false", } last_err: Optional[Exception] = None for attempt in range(retries): self.limiter.wait() try: r = self.session.get(f"{self.base}/scrape", params=params, timeout=70) if r.status_code == 429: time.sleep(1.5 * (attempt + 1)) continue r.raise_for_status() result = (r.json() or {}).get("result", {}) or {} content = result.get("content", "") or "" meta = { "title": (result.get("metadata") or {}).get("title", "") if isinstance(result.get("metadata"), dict) else "", "status_code": result.get("status_code"), "url": result.get("url", url), } return {"markdown": content, "metadata": meta, "links": [], "backend": "scrapfly"} except Exception as e: # noqa: BLE001 last_err = e time.sleep(1.5 * (attempt + 1)) raise RuntimeError(f"Scrapfly scrape a échoué: {last_err}")