spb/food-ka Public

1# -----------------------------------------------------------------------------2# Food-Ka — connecteur IGA (Sobeys Québec — plateforme Voilà, voila.ca)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# iga.net délègue l'épicerie en ligne à voila.ca. Deux voies :5#   1. (principale) API REST publique des promotions — JSON complet avec prix,6#      prix promo, format et catégorie, pagination par curseur, sans anti-bot.7#   2. (complément, optionnelle) recherche de la SPA rendue via Scrapfly8#      (render_js) pour les produits hors promotion.9# -----------------------------------------------------------------------------10from __future__ import annotations1112import re1314from bs4 import BeautifulSoup1516from ..schema import Product, normalize_category, parse_price17from .base import BaseConnector1819PLP_URL = "https://voila.ca/api/product-listing-pages/v1/pages/promotions"20# regionId public observé (région de livraison Québec) — requis par l'API21REGION_ID = "dd7143ad-a16a-4c1d-8dee-e3c7e48dc511"2223# (terme de recherche, catégorie canonique) — voie complémentaire (rendu JS)24SEARCHES = [25    ("fruits", "Fruits et légumes"),26    ("viande", "Viandes et volailles"),27    ("fromage", "Charcuteries et fromages"),28    ("pain", "Boulangerie"),29    ("surgele", "Surgelés"),30]3132_ID_RE = re.compile(r"/([0-9]+[A-Z]{2})/?$")33_SIZE_IN_TITLE = re.compile(34    r"(\d+(?:[.,]\d+)?\s*(?:kg|g|ml|l|lb|oz|un))\b", re.IGNORECASE)353637class IgaConnector(BaseConnector):38    source_id = "iga"39    request_delay = 1.040    max_api_pages = 12           # 50 produits/page (promotions en cours)41    use_search = False           # recherches Scrapfly complémentaires (coûteux)4243    # -- voie 1 : API REST des promotions --------------------------------------44    def _fetch_promotions(self) -> list[Product]:45        products: list[Product] = []46        token: str | None = None47        for _ in range(self.max_api_pages):48            params = {"regionId": REGION_ID, "maxPageSize": 50,49                      "maxProductsToDecorate": 50}50            if token:51                params["pageToken"] = token52            data = self.get(PLP_URL, params=params,53                            headers={"Accept": "application/json",54                                     "Accept-Language": "fr-CA,fr;q=0.9"}).json()55            decorated = list(data.get("decoratedProducts") or [])56            for group in data.get("productGroups") or []:57                decorated.extend(group.get("decoratedProducts") or [])58            for p in decorated:59                rid = p.get("retailerProductId")60                name = p.get("name") or ""61                if not rid or not name:62                    continue63                price = (p.get("price") or {}).get("amount")64                promo = (p.get("promoPrice") or {}).get("amount")65                current, regular = (promo, price) if promo else (price, None)66                image = p.get("image") or {}67                srcset = image.get("bopSrcset") or ""68                img_url = srcset.split(" ")[0] if srcset else ""69                cat_path = p.get("categoryPath") or []70                unit = ((p.get("unitPrice") or {}).get("price") or {}).get("amount")71                products.append(Product(72                    source=self.source_id,73                    external_id=str(rid),74                    url=f"https://voila.ca/products/{rid}",75                    name=name,76                    brand=p.get("brand") or "",77                    category=normalize_category(cat_path[0] if cat_path else ""),78                    category_raw=" / ".join(cat_path),79                    size_label=p.get("packSizeDescription") or "",80                    price=float(current) if current else None,81                    regular_price=float(regular) if regular else None,82                    on_sale=promo is not None,83                    in_stock=p.get("available"),84                    keywords=[a.get("label") for a in p.get("iconAttributes") or []85                              if isinstance(a, dict) and a.get("label")],86                    images=[img_url] if img_url.startswith("http") else [],87                ))88            token = (data.get("metadata") or {}).get("nextPageToken")89            if not token or not decorated:90                break91        return products9293    # -- voie 2 : recherche SPA rendue (complément) -----------------------------94    def _parse_search(self, html: str, category: str, term: str) -> list[Product]:95        soup = BeautifulSoup(html, "html.parser")96        products: list[Product] = []97        for link in soup.select('[data-test="fop-product-link"][href]'):98            href = link["href"]99            m = _ID_RE.search(href)100            if not m:101                continue102            tile = link103            for _ in range(6):104                if tile.parent is None:105                    break106                tile = tile.parent107                if tile.select_one('[data-test="fop-price"]'):108                    break109            title_el = tile.select_one('[data-test="fop-title"]') or link110            price_el = tile.select_one('[data-test="fop-price"]')111            orig_el = tile.select_one('[data-test="fop-original-price"]')112            unit_el = tile.select_one('[data-test="fop-price-per-unit"]')113            img = tile.select_one("img[src]")114            name = title_el.get_text(" ", strip=True)115            if not name:116                continue117            size_m = _SIZE_IN_TITLE.search(name)118            regular = parse_price(orig_el.get_text(" ", strip=True)) if orig_el else None119            products.append(Product(120                source=self.source_id,121                external_id=m.group(1),122                url=f"https://voila.ca{href}" if href.startswith("/") else href,123                name=name,124                category=category,125                category_raw=term,126                size_label=size_m.group(1) if size_m else "",127                price=parse_price(price_el.get_text(" ", strip=True)) if price_el else None,128                regular_price=regular,129                on_sale=regular is not None,130                unit_price_label=(unit_el.get_text(" ", strip=True).strip("()")131                                  if unit_el else ""),132                images=([img["src"]] if img is not None133                        and str(img.get("src", "")).startswith("http") else []),134            ))135        return products136137    def fetch(self) -> list[Product]:138        products = self._fetch_promotions()139        seen = {p.uid for p in products}140        if self.use_search:141            for term, category in SEARCHES:142                url = f"https://voila.ca/products?q={term.replace(' ', '+')}"143                try:144                    html = self.get_scrapfly(145                        url, render_js=True,146                        headers={"Accept-Language": "fr-CA,fr;q=0.9"},147                        wait_for_selector='[data-test="fop-product-link"]')148                except Exception:149                    continue150                for prod in self._parse_search(html, category, term):151                    if prod.uid not in seen:152                        seen.add(prod.uid)153                        products.append(prod)154        return products155