spb/food-ka Public

1# -----------------------------------------------------------------------------2# Food-Ka — Agrégateur de produits d'épicerie (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/_loblaw.py : socle commun des bannières Loblaw (Maxi, Provigo)5#   Les sites Loblaw (Next.js) embarquent la grille produits complète dans6#   <script id="__NEXT_DATA__"> ; on la lit derrière l'anti-bot via Scrapfly.7# -----------------------------------------------------------------------------8from __future__ import annotations910import json11import re1213from ..schema import Product, parse_price14from .base import BaseConnector1516_NEXT_RE = re.compile(17    r'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>', re.S)1819# sous-catégories standard du réseau Loblaw : (slug, code, catégorie canonique)20CATEGORIES = [21    ("fruits-et-legumes", 28000, "Fruits et légumes"),22    ("viandes", 27998, "Viandes et volailles"),23    ("poissons-et-fruits-de-mer", 28004, "Poissons et fruits de mer"),24    ("charcuteries-et-fromages", 28131, "Charcuteries et fromages"),25    ("produits-laitiers-et-oeufs", 28008, "Produits laitiers et œufs"),26    ("boulangerie", 28020, "Boulangerie"),27    ("surgeles", 28195, "Surgelés"),28    ("garde-manger", 28016, "Garde-manger"),29    ("boissons", 28051, "Boissons"),30]313233class LoblawConnector(BaseConnector):34    """Base des bannières Loblaw — sous-classes : définir source_id et domain."""3536    domain: str = ""          # ex. "www.maxi.ca"37    pages_per_category: int = 2   # 48 produits/page ; ~860 produits/sync3839    def _next_data(self, url: str) -> dict | None:40        html = self.get_scrapfly(url)41        m = _NEXT_RE.search(html)42        return json.loads(m.group(1)) if m else None4344    def _tiles(self, data: dict) -> tuple[list[dict], bool]:45        """(tuiles produit, hasMore) — collecte récursive : selon la page46        (racine ou sous-catégorie), Loblaw embarque une ou plusieurs grilles47        `productTiles`, avec ou sans bloc `pagination`."""48        tiles: list[dict] = []49        has_more = False5051        def walk(o):52            nonlocal has_more53            if isinstance(o, dict):54                if "productTiles" in o:55                    tiles.extend(o.get("productTiles") or [])56                    if (o.get("pagination") or {}).get("hasMore"):57                        has_more = True58                    return59                for v in o.values():60                    walk(v)61            elif isinstance(o, list):62                for v in o:63                    walk(v)6465        try:66            walk(data["props"]["pageProps"]["initialData"]["layout"]["sections"])67        except (KeyError, TypeError):68            pass69        return tiles, has_more7071    def _tile_to_product(self, tile: dict, category: str,72                         category_raw: str) -> Product | None:73        pid = tile.get("productId")74        if not pid:75            return None76        pricing = tile.get("pricing") or {}77        price = parse_price(pricing.get("price"))78        was = parse_price(pricing.get("wasPrice"))79        images = [img.get("largeUrl") or img.get("imageUrl")80                  for img in tile.get("productImage") or [] if isinstance(img, dict)]81        link = tile.get("link") or ""82        badges = []83        deal = tile.get("deal") or {}84        if isinstance(deal, dict) and deal.get("name"):85            badges.append(str(deal["name"]))86        return Product(87            source=self.source_id,88            external_id=str(pid),89            url=f"https://{self.domain}{link}" if link.startswith("/") else link,90            name=tile.get("title") or "",91            brand=tile.get("brand") or "",92            category=category,93            category_raw=category_raw,94            size_label=tile.get("packageSizing") or "",95            price=price,96            regular_price=was,97            price_label=pricing.get("displayPrice") or "",98            on_sale=bool(was and price and was > price),99            in_stock=None,100            keywords=badges,101            details={"article": tile.get("articleNumber"),102                     "uom": tile.get("uom")},103            images=[i for i in images if i],104        )105106    def fetch(self) -> list[Product]:107        products: list[Product] = []108        seen: set[str] = set()109        for slug, code, category in CATEGORIES:110            for page in range(1, self.pages_per_category + 1):111                url = f"https://{self.domain}/fr/{slug}/c/{code}?page={page}"112                try:113                    data = self._next_data(url)114                except Exception:   # une catégorie qui casse ne bloque pas le reste115                    break116                if not data:117                    break118                tiles, has_more = self._tiles(data)119                new = 0120                for tile in tiles:121                    try:122                        prod = self._tile_to_product(tile, category,123                                                     slug.replace("-", " "))124                    except Exception:125                        continue126                    if prod and prod.uid not in seen:127                        seen.add(prod.uid)128                        products.append(prod)129                        new += 1130                # sans bloc pagination : on s'arrête dès qu'une page ne131                # rapporte plus de nouveaux produits132                if not has_more and new == 0:133                    break134        return products135