# ============================================================================== # Author: Simon-Pierre Boucher # File: restoka/connectors/base.py # Desc: Classe de base des connecteurs + backends de fetch (requests direct, # Firecrawl ou Scrapfly pour les sites JavaScript / anti-bot). # Calquée sur louka/connectors/base.py — mêmes conventions. # ============================================================================== from __future__ import annotations import json import os import time import requests from ..schema import Restaurant USER_AGENT = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126 Safari/537.36 " "RestoKaBot/1.0 (+https://www.resto-ka.com/bot; contact@spboucher.ai)") FIRECRAWL_API = "https://api.firecrawl.dev/v1/scrape" SCRAPFLY_API = "https://api.scrapfly.io/scrape" class SkipSource(RuntimeError): """Levée par un connecteur qui ne peut pas tourner (clé API manquante…). Le pipeline (ingest.run) consigne le motif et passe à la source suivante SANS marquer d'échec dans sync_log — ce n'est pas une erreur, la source est simplement en attente (ex. Yelp Fusion sans YELP_API_KEY).""" class BaseConnector: """Un connecteur = un adaptateur propre à une source (ou une plateforme). Sous-classes : définir `source_id` et implémenter `fetch()` qui retourne la liste complète des restaurants (avec menu) actuellement publiés par la source. Le pipeline (ingest.py) s'occupe du diff avec la base de données. """ source_id: str = "" request_delay: float = 0.6 # politesse entre requêtes timeout: int = 30 use_detail_cache: bool = True # cache BD des payloads détail def __init__(self) -> None: self.session = requests.Session() self.session.headers["User-Agent"] = USER_AGENT self._last_request = 0.0 self._detail_con = None # -- backends ------------------------------------------------------------- def _throttle(self) -> None: wait = self.request_delay - (time.time() - self._last_request) if wait > 0: time.sleep(wait) def get(self, url: str, **kw) -> requests.Response: """GET direct avec throttling poli.""" self._throttle() resp = self.session.get(url, timeout=self.timeout, **kw) self._last_request = time.time() resp.raise_for_status() return resp def post(self, url: str, **kw) -> requests.Response: """POST direct avec throttling poli (APIs internes des plateformes).""" self._throttle() resp = self.session.post(url, timeout=self.timeout, **kw) self._last_request = time.time() resp.raise_for_status() return resp def get_rendered(self, url: str) -> str: """HTML rendu (JavaScript exécuté) via Firecrawl. Nécessite FIRECRAWL_API_KEY dans l'environnement (.env). À utiliser pour les sites de restos SPA relativement ouverts. """ key = os.environ.get("FIRECRAWL_API_KEY") if not key: raise RuntimeError("FIRECRAWL_API_KEY manquant (voir .env)") resp = requests.post( FIRECRAWL_API, json={"url": url, "formats": ["html"]}, headers={"Authorization": f"Bearer {key}"}, timeout=90, ) resp.raise_for_status() data = resp.json() return (data.get("data") or {}).get("html", "") def scrapfly(self, url: str, render_js: bool = True, asp: bool = True, rendering_wait: int = 0, country: str = "ca", wait_for_selector: str | None = None, js_scenario: list | str | None = None, proxy_pool: str | None = None, headers: dict | None = None, method: str = "GET", body: str | None = None) -> dict: """Appel Scrapfly complet — retourne le dict `result` (content, status…). Requis pour les plateformes de livraison (anti-bot agressif) ; voir CLAUDE.md §10 et §15 avant d'industrialiser sur ces sources. """ import base64 key = os.environ.get("SCRAPFLY_KEY") if not key: raise RuntimeError("SCRAPFLY_KEY manquant (voir .env)") params: dict = {"key": key, "url": url, "country": country} if asp: params["asp"] = "true" if render_js: params["render_js"] = "true" if rendering_wait: params["rendering_wait"] = rendering_wait if wait_for_selector: params["wait_for_selector"] = wait_for_selector if proxy_pool: params["proxy_pool"] = proxy_pool if js_scenario is not None: js = js_scenario if isinstance(js_scenario, str) else json.dumps(js_scenario) params["js_scenario"] = base64.urlsafe_b64encode(js.encode()).decode() if headers: for k, v in headers.items(): params[f"headers[{k}]"] = v self._throttle() if method.upper() == "POST": resp = requests.post(SCRAPFLY_API, params=params, data=(body or ""), timeout=180) else: resp = requests.get(SCRAPFLY_API, params=params, timeout=180) self._last_request = time.time() try: return resp.json().get("result") or {} except ValueError: return {} def get_scrapfly(self, url: str, **kw) -> str: """HTML rendu via Scrapfly (ASP = bypass anti-bot + rendu JS).""" return self.scrapfly(url, **kw).get("content") or "" def detail(self, external_id: str, key: str, fetch_fn) -> dict: """Payload « détail » avec cache : `fetch_fn` n'est appelé que si le resto est nouveau ou si sa clé (hash du contenu liste) a changé. Évite de recapturer un menu complet (dizaines de requêtes) quand rien n'a bougé. `fetch_fn` doit retourner un dict JSON-sérialisable. """ if not self.use_detail_cache: return fetch_fn() or {} from .. import db if self._detail_con is None: self._detail_con = db.connect() cached = db.get_cached_detail(self._detail_con, self.source_id, str(external_id), key) if cached is not None: return cached payload = fetch_fn() or {} db.put_cached_detail(self._detail_con, self.source_id, str(external_id), key, payload) return payload # -- contrat -------------------------------------------------------------- def fetch(self) -> list[Restaurant]: raise NotImplementedError # ============================================================================= # Résilience anti-bot (Groupe KA) — auto-escalade de get() sans toucher au corps. # Ajouté par l'orchestrateur KA : enrobe BaseConnector.get pour qu'un blocage # anti-bot (403/429/503/challenge) ou une coupure réseau déclenche la chaîne # de secours (Oxylabs résidentiel -> Scrapfly ASP -> Bright Data). Voir # connectors/_resilient.py. Idempotent (marqueur _KA_RESILIENT_WRAPPED). # ============================================================================= if not getattr(BaseConnector, "_KA_RESILIENT_WRAPPED", False): import requests as _ka_requests # noqa: E402 from . import _resilient as _kar # noqa: E402 _ka_orig_get = BaseConnector.get def _ka_full_url(url, kw): try: return _ka_requests.Request("GET", url, params=kw.get("params")).prepare().url except Exception: # noqa: BLE001 return url def _ka_resilient_get(self, url, **kw): timeout = getattr(self, "timeout", 30) headers = kw.get("headers") try: return _ka_orig_get(self, url, **kw) except _ka_requests.HTTPError as exc: r = getattr(exc, "response", None) if r is not None and _kar.is_blocked(r): target = getattr(r, "url", None) or _ka_full_url(url, kw) better = _kar.escalate_if_blocked( r, target, timeout=timeout, headers=headers) if better is not None and getattr(better, "status_code", 0) == 200: return better raise except (_ka_requests.ConnectionError, _ka_requests.Timeout): better = _kar.escalate(_ka_full_url(url, kw), timeout=timeout, headers=headers) if better is not None and getattr(better, "status_code", 0) == 200: return better raise def _ka_get_resilient(self, url, *, render_js=False, country="ca", **kw): """Fetch anti-bot explicite : force la chaîne de secours au besoin. Comme get() mais tente d'abord le direct puis escalade même sur 200- challenge, avec rendu JS optionnel. Renvoie une réponse compatible requests (.text/.content/.status_code/.json()...). """ timeout = getattr(self, "timeout", 30) headers = kw.get("headers") try: resp = _ka_orig_get(self, url, **kw) except _ka_requests.HTTPError as exc: resp = getattr(exc, "response", None) except (_ka_requests.ConnectionError, _ka_requests.Timeout): resp = None target = _ka_full_url(url, kw) if resp is not None and getattr(resp, "url", None): target = resp.url return _kar.escalate_if_blocked(resp, target, timeout=timeout, country=country, render_js=render_js, headers=headers) BaseConnector.get = _ka_resilient_get BaseConnector.get_resilient = _ka_get_resilient BaseConnector._KA_RESILIENT_WRAPPED = True