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/_woocommerce.py : socle commun des épiceries sur WooCommerce5#   L'API publique "Store API" expose le catalogue en JSON sans clé :6#   /wp-json/wc/store/v1/products?per_page=100&page=N. Les prix y sont en7#   unités mineures (cents) — "1299" + currency_minor_unit=2 -> 12,99 $.8# -----------------------------------------------------------------------------9from __future__ import annotations1011import html12import re1314from ..schema import Product, normalize_category15from .base import BaseConnector1617_TAG_RE = re.compile(r"<[^>]+>")1819# format dans le nom du produit : "Yogourt nature (650 g)", "Huile 2 x 750 ml"20_SIZE_IN_NAME_RE = re.compile(21    r"(\d+\s*[x×]\s*)?\d+(?:[.,]\d+)?\s*(?:kg|g|mg|lb|lbs|oz|ml|cl|l|un|unités?)\b",22    re.IGNORECASE)232425def _strip_html(text: str | None, max_len: int = 500) -> str:26    """Retire balises et entités HTML d'un champ WooCommerce, tronque (~500)."""27    if not text:28        return ""29    clean = html.unescape(_TAG_RE.sub(" ", text))30    clean = re.sub(r"\s+", " ", clean).strip()31    if len(clean) > max_len:32        clean = clean[:max_len].rsplit(" ", 1)[0] + "…"33    return clean343536def _minor(value: str | int | None, minor_unit: int) -> float | None:37    """Convertit un prix Store API (unités mineures) en dollars : "1299" -> 12.99."""38    if value in (None, ""):39        return None40    try:41        return round(int(value) / (10 ** minor_unit), 2)42    except (TypeError, ValueError):43        return None444546class WooStoreConnector(BaseConnector):47    """Base des épiceries WooCommerce — sous-classes : source_id et domain."""4849    domain: str = ""          # ex. "bocoboco.ca"50    max_pages: int = 8        # 100 produits/page51    per_page: int = 1005253    def _page(self, page: int) -> list[dict]:54        url = (f"https://{self.domain}/wp-json/wc/store/v1/products"55               f"?per_page={self.per_page}&page={page}")56        resp = self.get(url, headers={"Accept": "application/json"})57        data = resp.json()58        return data if isinstance(data, list) else []5960    def _to_product(self, raw: dict) -> Product | None:61        pid = raw.get("id")62        name = html.unescape(_TAG_RE.sub("", raw.get("name") or "")).strip()63        if not pid or not name:64            return None6566        prices = raw.get("prices") or {}67        minor_unit = int(prices.get("currency_minor_unit") or 2)68        price = _minor(prices.get("price"), minor_unit)69        on_sale = bool(raw.get("on_sale"))70        regular = _minor(prices.get("regular_price"), minor_unit) if on_sale else None71        if regular is not None and price is not None and regular <= price:72            regular = None  # solde annoncé mais prix identiques : pas un rabais7374        # taxonomie : première catégorie = catégorie brute, les autres en mots-clés75        cats = [c.get("name") or "" for c in raw.get("categories") or []76                if isinstance(c, dict)]77        cats = [html.unescape(c).strip() for c in cats if c]78        category_raw = cats[0] if cats else ""79        keywords = cats[1:]8081        # format : extrait du nom s'il y figure (ex. "Tofu ferme 454 g")82        m = _SIZE_IN_NAME_RE.search(name)83        size_label = m.group(0).strip() if m else ""8485        return Product(86            source=self.source_id,87            external_id=str(pid),88            url=raw.get("permalink") or f"https://{self.domain}/?p={pid}",89            name=name,90            brand="",91            category=normalize_category(category_raw),92            category_raw=category_raw,93            size_label=size_label,94            price=price,95            regular_price=regular,96            on_sale=on_sale and regular is not None,97            in_stock=(bool(raw["is_in_stock"])98                      if raw.get("is_in_stock") is not None else None),99            description=_strip_html(raw.get("short_description")100                                    or raw.get("description")),101            keywords=keywords,102            images=[img["src"] for img in raw.get("images") or []103                    if isinstance(img, dict) and img.get("src")],104        )105106    # -- contrat ------------------------------------------------------------------107    def fetch(self) -> list[Product]:108        products: list[Product] = []109        seen: set[str] = set()110        for page in range(1, self.max_pages + 1):111            try:112                batch = self._page(page)113            except Exception:   # fin de pagination ou site en dérangement114                break115            if not batch:116                break117            for raw in batch:118                try:119                    prod = self._to_product(raw)120                except Exception:121                    continue122                if prod and prod.uid not in seen:123                    seen.add(prod.uid)124                    products.append(prod)125        return products126