# ============================================================================== # Author: Simon-Pierre Boucher # File: restoka/connectors/chownow.py # Desc: Connecteur templatisé ChowNow (Palier 1 — plateforme de commande en # ligne des restos eux-mêmes -> PRIX RÉELS, contexte `takeout`). # API JSON publique et non authentifiée : # GET api.chownow.com/api/restaurant/ (fiche : nom, # adresse, téléphone, GPS « lat,lng », cuisines, site web) # GET api.chownow.com/api/restaurant//menu (menu_categories[] # .items[] {name, description, price, size} + graphe complet # des modificateurs -> options §5.2) # Une intégration = un id de resto ChowNow, consigné dans # data/sources.json (source « chownow » -> integrations[]) et/ou # découvert automatiquement par le connecteur site-resto # (data/chownow-discovered.json). CONNECTEUR DORMANT tant qu'aucune # intégration n'est connue (SkipSource — pas un échec). # Seuls les restos au Québec (state == QC) sont émis. # # Mode d'extraction : API directe (2 requêtes/resto), politesse 0.6 s. # Base légale §15 : API ouverte de la plateforme du resto (source de # première partie), extraction minimale, lien source conservé. # ============================================================================== from __future__ import annotations import json import sys from pathlib import Path from ..schema import Restaurant from .base import BaseConnector, SkipSource API = "https://api.chownow.com/api/restaurant" SOURCES_PATH = Path(__file__).resolve().parents[2] / "data" / "sources.json" def parse_geo(geo: str) -> tuple[float | None, float | None]: """« 45.5017,-73.5673 » -> (lat, lng).""" try: lat_s, lng_s = (geo or "").split(",", 1) return float(lat_s), float(lng_s) except (ValueError, AttributeError): return None, None def build_options(item: dict, mod_cats: dict[str, dict], mods: dict[str, dict]) -> list[dict]: """Options §5.2 depuis le graphe de modificateurs ChowNow (item -> modifier_categories -> modifiers {name, price}).""" options: list[dict] = [] for cat_id in item.get("modifier_categories") or []: cat = mod_cats.get(str(cat_id)) if not cat: continue choices = [] for mod_id in cat.get("modifiers") or []: mod = mods.get(str(mod_id)) if not mod or not (mod.get("name") or "").strip(): continue price = mod.get("price") choices.append({ "name": mod["name"].strip(), "price_delta": round(float(price), 2) if isinstance(price, (int, float)) else 0.0, }) if choices: options.append({ "group": (cat.get("name") or "").strip() or "Options", "required": bool(cat.get("min_qty")), "choices": choices, }) return options def build_menu(menu_payload: dict, captured_at: str) -> dict | None: """Menu standard Resto·Ka (CLAUDE.md §5.2) depuis menu_categories.""" mod_cats = {str(c.get("id")): c for c in menu_payload.get("modifier_categories") or []} mods = {str(m.get("id")): m for m in menu_payload.get("modifiers") or []} sections_out: list[dict] = [] seen: set[str] = set() for cat in menu_payload.get("menu_categories") or []: name = (cat.get("name") or "").strip() if not name or name in seen: continue items = [] for it in cat.get("items") or []: title = (it.get("name") or "").strip() if not title: continue size = (it.get("size") or "").strip() if size and size.lower() not in ("regular", "régulier"): title = f"{title} ({size})" price = it.get("price") entry = { "name": title, "description": (it.get("description") or "").strip(), "price": round(float(price), 2) if isinstance(price, (int, float)) and price > 0 else None, } options = build_options(it, mod_cats, mods) if options: entry["options"] = options items.append(entry) if items: seen.add(name) sections_out.append({"name": name, "items": items}) if not sections_out: return None return { "price_context": "takeout", "price_source": "chownow", "currency": "CAD", "captured_at": captured_at, "sections": sections_out, } class ChowNowConnector(BaseConnector): source_id = "chownow" request_delay = 0.8 timeout = 30 use_detail_cache = False # 2 requêtes API légères par resto def _integrations(self) -> list[dict]: registry = json.loads(SOURCES_PATH.read_text(encoding="utf-8")) out: list[dict] = [] for src in registry.get("sources", []): if src.get("id") == self.source_id: out.extend(src.get("integrations") or []) # ids découverts automatiquement par site-resto (widget ChowNow # détecté sur le site d'un resto) — voir siteresto.py discovered = SOURCES_PATH.parent / "chownow-discovered.json" if discovered.exists(): try: extra = json.loads(discovered.read_text(encoding="utf-8")) known = {str(i.get("key")) for i in out} out.extend(i for i in extra.get("integrations", []) if str(i.get("key")) not in known) except ValueError: pass return out def _now(self) -> str: import datetime return datetime.datetime.now(datetime.timezone.utc) \ .strftime("%Y-%m-%dT%H:%M:%SZ") def _fetch_restaurant(self, cn_id: str) -> Restaurant | None: info = self.get(f"{API}/{cn_id}", headers={"Accept": "application/json"}).json() addr = info.get("address") or {} state = (addr.get("state") or "").strip().upper() if state and state != "QC": print(f"[resto-ka] chownow: {cn_id} hors Québec ({state}) — ignoré", file=sys.stderr) return None name = (info.get("name") or "").strip() if not name: return None lat, lng = parse_geo(addr.get("geo") or "") menu_payload = self.get(f"{API}/{cn_id}/menu", headers={"Accept": "application/json"}).json() menu = build_menu(menu_payload or {}, self._now()) services = ["takeout"] if (info.get("fulfillment") or {}).get("delivery") \ or info.get("has_low_cost_delivery"): services.append("delivery") return Restaurant( source=self.source_id, external_id=str(cn_id), name=name, url=f"https://direct.chownow.com/order/{cn_id}/locations", address=(addr.get("street_address1") or "").strip(), city=(addr.get("city") or "").strip(), postal_code=(addr.get("zip") or "").strip(), lat=lat, lng=lng, phone=(info.get("phone") or "").strip(), website=(info.get("website_url") or "").strip(), services=services, menu=menu, ) def fetch(self) -> list[Restaurant]: integrations = [i for i in self._integrations() if i.get("key") and i.get("status") != "inactif"] if not integrations: raise SkipSource( "aucune intégration ChowNow connue (data/sources.json et " "data/chownow-discovered.json vides) — connecteur dormant, " "en attente des découvertes du connecteur site-resto") out: list[Restaurant] = [] for integ in integrations: cn_id = str(integ["key"]) try: resto = self._fetch_restaurant(cn_id) except Exception as exc: # un resto ne bloque pas les autres print(f"[resto-ka] chownow: intégration {cn_id} ignorée : " f"{exc}", file=sys.stderr) continue if resto is not None: out.append(resto) print(f"[resto-ka] chownow: {len(integrations)} intégration(s) -> " f"{len(out)} resto(s) QC avec menu takeout") return out