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/_shopify.py : socle commun des épiceries sur Shopify5#   Toute boutique Shopify expose son catalogue public en JSON :6#   /products.json?limit=250&page=N (ou /collections/{handle}/products.json7#   pour cibler des rayons précis). Aucun anti-bot, requests direct suffit.8# -----------------------------------------------------------------------------9from __future__ import annotations1011import html12import re1314from ..schema import Product, normalize_category15from .base import BaseConnector1617_TAG_RE = re.compile(r"<[^>]+>")181920def _strip_html(text: str | None, max_len: int = 500) -> str:21    """Retire les balises HTML d'un body_html Shopify et tronque (~500 car.)."""22    if not text:23        return ""24    clean = html.unescape(_TAG_RE.sub(" ", text))25    clean = re.sub(r"\s+", " ", clean).strip()26    if len(clean) > max_len:27        clean = clean[:max_len].rsplit(" ", 1)[0] + "…"28    return clean293031class ShopifyConnector(BaseConnector):32    """Base des épiceries Shopify — sous-classes : définir source_id et domain.3334    Par défaut on pagine /products.json (catalogue complet). Si le champ35    `product_type` de la boutique est vide ou inutilisable, une sous-classe36    peut définir `collections` : liste de (handle, catégorie canonique) —37    on pagine alors /collections/{handle}/products.json à la place.38    Une catégorie canonique vide ("") signifie : déduire du product_type.39    """4041    domain: str = ""                              # ex. "www.supermarchepa.com"42    max_pages: int = 8                            # 250 produits/page43    collections: list[tuple[str, str]] = []       # [(handle, catégorie canonique)]4445    # -- récupération -----------------------------------------------------------46    def _page(self, url: str) -> list[dict]:47        """Une page de produits JSON ; liste vide si la page n'existe pas."""48        resp = self.get(url, headers={"Accept": "application/json"})49        data = resp.json()50        return data.get("products") or []5152    # -- mapping Shopify -> Product ----------------------------------------------53    def _pick_variant(self, product: dict) -> dict:54        """Première variante disponible, sinon la première tout court."""55        variants = product.get("variants") or []56        for v in variants:57            if v.get("available"):58                return v59        return variants[0] if variants else {}6061    def _to_product(self, product: dict, category: str = "") -> Product | None:62        pid = product.get("id")63        handle = product.get("handle") or ""64        if not pid or not handle:65            return None66        variant = self._pick_variant(product)6768        # prix : la variante porte le prix courant ; compare_at_price est le69        # prix régulier quand le produit est en solde.70        try:71            price = float(variant.get("price")) if variant.get("price") else None72        except (TypeError, ValueError):73            price = None74        regular = None75        try:76            cap = variant.get("compare_at_price")77            if cap and price is not None and float(cap) > price:78                regular = float(cap)79        except (TypeError, ValueError):80            regular = None8182        # format : titre de variante s'il est réel, sinon le poids en grammes83        size_label = ""84        vtitle = variant.get("title") or ""85        if vtitle and vtitle != "Default Title":86            size_label = vtitle87        elif variant.get("grams"):88            size_label = f"{variant['grams']} g"8990        tags = [str(t) for t in product.get("tags") or [] if t]91        category_raw = product.get("product_type") or " ".join(tags)9293        return Product(94            source=self.source_id,95            external_id=str(pid),96            url=f"https://{self.domain}/products/{handle}",97            name=html.unescape(product.get("title") or ""),98            brand=product.get("vendor") or "",99            category=category or normalize_category(category_raw),100            category_raw=category_raw,101            size_label=size_label,102            price=price,103            regular_price=regular,104            on_sale=bool(regular is not None),105            in_stock=(bool(variant["available"])106                      if variant.get("available") is not None else None),107            description=_strip_html(product.get("body_html")),108            keywords=tags,109            images=[img["src"] for img in product.get("images") or []110                    if isinstance(img, dict) and img.get("src")],111        )112113    # -- contrat ------------------------------------------------------------------114    def fetch(self) -> list[Product]:115        products: list[Product] = []116        seen: set[str] = set()117        # sources à paginer : catalogue complet, ou collections ciblées118        sources: list[tuple[str, str]] = (119            [(f"https://{self.domain}/collections/{handle}/products.json", cat)120             for handle, cat in self.collections]121            or [(f"https://{self.domain}/products.json", "")])122        for base_url, category in sources:123            for page in range(1, self.max_pages + 1):124                try:125                    batch = self._page(f"{base_url}?limit=250&page={page}")126                except Exception:   # une collection qui casse ne bloque pas le reste127                    break128                if not batch:129                    break130                for raw in batch:131                    try:132                        prod = self._to_product(raw, category)133                    except Exception:134                        continue135                    if prod and prod.name and prod.uid not in seen:136                        seen.add(prod.uid)137                        products.append(prod)138        return products139