# ----------------------------------------------------------------------------- # Food-Ka — connecteur IGA (Sobeys Québec — plateforme Voilà, voila.ca) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # iga.net délègue l'épicerie en ligne à voila.ca. Deux voies : # 1. (principale) API REST publique des promotions — JSON complet avec prix, # prix promo, format et catégorie, pagination par curseur, sans anti-bot. # 2. (complément, optionnelle) recherche de la SPA rendue via Scrapfly # (render_js) pour les produits hors promotion. # ----------------------------------------------------------------------------- from __future__ import annotations import re from bs4 import BeautifulSoup from ..schema import Product, normalize_category, parse_price from .base import BaseConnector PLP_URL = "https://voila.ca/api/product-listing-pages/v1/pages/promotions" # regionId public observé (région de livraison Québec) — requis par l'API REGION_ID = "dd7143ad-a16a-4c1d-8dee-e3c7e48dc511" # (terme de recherche, catégorie canonique) — voie complémentaire (rendu JS) SEARCHES = [ ("fruits", "Fruits et légumes"), ("viande", "Viandes et volailles"), ("fromage", "Charcuteries et fromages"), ("pain", "Boulangerie"), ("surgele", "Surgelés"), ] _ID_RE = re.compile(r"/([0-9]+[A-Z]{2})/?$") _SIZE_IN_TITLE = re.compile( r"(\d+(?:[.,]\d+)?\s*(?:kg|g|ml|l|lb|oz|un))\b", re.IGNORECASE) class IgaConnector(BaseConnector): source_id = "iga" request_delay = 1.0 max_api_pages = 12 # 50 produits/page (promotions en cours) use_search = False # recherches Scrapfly complémentaires (coûteux) # -- voie 1 : API REST des promotions -------------------------------------- def _fetch_promotions(self) -> list[Product]: products: list[Product] = [] token: str | None = None for _ in range(self.max_api_pages): params = {"regionId": REGION_ID, "maxPageSize": 50, "maxProductsToDecorate": 50} if token: params["pageToken"] = token data = self.get(PLP_URL, params=params, headers={"Accept": "application/json", "Accept-Language": "fr-CA,fr;q=0.9"}).json() decorated = list(data.get("decoratedProducts") or []) for group in data.get("productGroups") or []: decorated.extend(group.get("decoratedProducts") or []) for p in decorated: rid = p.get("retailerProductId") name = p.get("name") or "" if not rid or not name: continue price = (p.get("price") or {}).get("amount") promo = (p.get("promoPrice") or {}).get("amount") current, regular = (promo, price) if promo else (price, None) image = p.get("image") or {} srcset = image.get("bopSrcset") or "" img_url = srcset.split(" ")[0] if srcset else "" cat_path = p.get("categoryPath") or [] unit = ((p.get("unitPrice") or {}).get("price") or {}).get("amount") products.append(Product( source=self.source_id, external_id=str(rid), url=f"https://voila.ca/products/{rid}", name=name, brand=p.get("brand") or "", category=normalize_category(cat_path[0] if cat_path else ""), category_raw=" / ".join(cat_path), size_label=p.get("packSizeDescription") or "", price=float(current) if current else None, regular_price=float(regular) if regular else None, on_sale=promo is not None, in_stock=p.get("available"), keywords=[a.get("label") for a in p.get("iconAttributes") or [] if isinstance(a, dict) and a.get("label")], images=[img_url] if img_url.startswith("http") else [], )) token = (data.get("metadata") or {}).get("nextPageToken") if not token or not decorated: break return products # -- voie 2 : recherche SPA rendue (complément) ----------------------------- def _parse_search(self, html: str, category: str, term: str) -> list[Product]: soup = BeautifulSoup(html, "html.parser") products: list[Product] = [] for link in soup.select('[data-test="fop-product-link"][href]'): href = link["href"] m = _ID_RE.search(href) if not m: continue tile = link for _ in range(6): if tile.parent is None: break tile = tile.parent if tile.select_one('[data-test="fop-price"]'): break title_el = tile.select_one('[data-test="fop-title"]') or link price_el = tile.select_one('[data-test="fop-price"]') orig_el = tile.select_one('[data-test="fop-original-price"]') unit_el = tile.select_one('[data-test="fop-price-per-unit"]') img = tile.select_one("img[src]") name = title_el.get_text(" ", strip=True) if not name: continue size_m = _SIZE_IN_TITLE.search(name) regular = parse_price(orig_el.get_text(" ", strip=True)) if orig_el else None products.append(Product( source=self.source_id, external_id=m.group(1), url=f"https://voila.ca{href}" if href.startswith("/") else href, name=name, category=category, category_raw=term, size_label=size_m.group(1) if size_m else "", price=parse_price(price_el.get_text(" ", strip=True)) if price_el else None, regular_price=regular, on_sale=regular is not None, unit_price_label=(unit_el.get_text(" ", strip=True).strip("()") if unit_el else ""), images=([img["src"]] if img is not None and str(img.get("src", "")).startswith("http") else []), )) return products def fetch(self) -> list[Product]: products = self._fetch_promotions() seen = {p.uid for p in products} if self.use_search: for term, category in SEARCHES: url = f"https://voila.ca/products?q={term.replace(' ', '+')}" try: html = self.get_scrapfly( url, render_js=True, headers={"Accept-Language": "fr-CA,fr;q=0.9"}, wait_for_selector='[data-test="fop-product-link"]') except Exception: continue for prod in self._parse_search(html, category, term): if prod.uid not in seen: seen.add(prod.uid) products.append(prod) return products