spb/fabri-ka

1# -----------------------------------------------------------------------------2# Fabri-Ka — Agrégateur de produits québécois3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/immunotec.py : storefront sur mesure Immunotec (Vaudreuil-Dorion) —5#   Next.js + backend Exigo (scus-back2.immunotec.com). Le site et son API sont6#   derrière un challenge Cloudflare : tout passe par Scrapfly ASP. Catalogue7#   public = « mur produits » de la catégorie « Tous les produits »8#   (POST categories/product-wall, header api-key "/"). Les prix n'existent que9#   dans cette API (rendu client), jamais dans le HTML SSR ni en JSON-LD.10# -----------------------------------------------------------------------------11from __future__ import annotations1213import json14import re15import unicodedata1617from ..schema import Product, parse_price18from .base import BaseConnector19from .scrapfly import scrapfly_get, scrapfly_post2021API_BASE = "https://scus-back2.immunotec.com/"22WALL_PATH = "categories/product-wall"23API_HEADERS = {"api-key": "/"}   # valeur littérale codée dans le bundle du site24# webCategoryId de « Tous les produits » (pageProps.productCategory de25# /fr-CA/products) — redécouvert dynamiquement si le mur revient vide26ALL_CATEGORY_ID = 213927LOC_RE = re.compile(r"<loc>\s*(.*?)\s*</loc>", re.I | re.S)282930def _slugify(s: str) -> str:31    s = unicodedata.normalize("NFKD", s or "").encode("ascii", "ignore").decode()32    return re.sub(r"-+", "-", re.sub(r"[^a-z0-9]+", "-", s.lower())).strip("-")333435def _compact(s: str) -> str:36    return re.sub(r"[^a-z0-9]", "", (s or "").lower())373839class ImmunotecConnector(BaseConnector):40    platform = "immunotec"4142    def _wall(self, category_id: int, culture: str) -> list[dict]:43        body = {"countryCode": "CA", "cultureCode": culture,44                "categoryId": category_id, "isConsultant": False,45                "downlineCustomerId": 0, "isShareCart": False,46                "isEnrollment": False}47        status, content = scrapfly_post(API_BASE + WALL_PATH, body,48                                        headers=API_HEADERS)49        if status != 200:50            raise RuntimeError(f"immunotec product-wall {status}")51        data = json.loads(content)52        return ((data.get("records") or {}).get("items") or []) if data.get("status") else []5354    def _all_category_id(self) -> int:55        """Relit le webCategoryId « Tous les produits » depuis la page du mur."""56        status, html = scrapfly_get(f"{self.base}/fr-CA/products")57        if status == 200:58            m = re.search(r'"slug":\s*"all".{0,400}?"customerWebCategoryId":\s*"?(\d+)', html) \59                or re.search(r'"customerWebCategoryId":\s*"?(\d+)"?.{0,400}?"slug":\s*"all"', html)60            if m:61                return int(m.group(1))62        return ALL_CATEGORY_ID6364    def _product_slugs(self) -> dict[str, str]:65        """Slugs fr-CA du sitemap produits (slug -> URL de fiche)."""66        url = f"{self.base}/products-sitemap.xml"67        xml = ""68        try:69            r = self.session.get(url, timeout=self.timeout)70            if r.status_code == 200:71                xml = r.text72        except Exception:  # noqa: BLE001 — Cloudflare : on escalade73            pass74        if "<loc>" not in xml:75            try:76                status, xml = scrapfly_get(url)77                if status != 200:78                    xml = ""79            except Exception:  # noqa: BLE001 — les fiches sont optionnelles80                xml = ""81        return {loc.rsplit("/", 1)[-1]: loc82                for loc in LOC_RE.findall(xml) if "/fr-CA/products/" in loc}8384    @staticmethod85    def _match_slug(item: dict, titles: list[str], slugs: dict[str, str]) -> str | None:86        sku_digits = (item.get("sku") or "").lstrip("0")87        cands = [_slugify(t) for t in titles if t]88        for c in cands:89            if c in slugs:90                return slugs[c]91        if len(sku_digits) >= 4:92            for slug, loc in slugs.items():93                if sku_digits in _compact(slug):94                    return loc95        for c in cands:96            cc = _compact(c)97            if len(cc) < 6:98                continue99            for slug, loc in slugs.items():100                sc = _compact(slug)101                if len(sc) >= 6 and (cc.startswith(sc) or sc.startswith(cc)):102                    return loc103        return None104105    def fetch(self) -> list[Product]:106        items = self._wall(ALL_CATEGORY_ID, "fr-CA")107        if not items:108            items = self._wall(self._all_category_id(), "fr-CA")109        if not items:110            return []111        # titres anglais : les slugs de fiches sont dérivés des noms EN112        try:113            en_titles = {i.get("sku"): (i.get("titleDescription") or i.get("name"))114                         for i in self._wall(ALL_CATEGORY_ID, "en-CA")}115        except Exception:  # noqa: BLE001 — enrichissement d'URL seulement116            en_titles = {}117        slugs = self._product_slugs()118        wall_url = f"{self.base}/fr-CA/products"119        out: list[Product] = []120        for it in items:121            sku = str(it.get("sku") or "").strip()122            if not sku:123                continue124            title = it.get("titleDescription") or it.get("name") or ""125            price = parse_price(it.get("priceRetail") or it.get("price"))126            url = self._match_slug(it, [title, en_titles.get(sku, "")], slugs) \127                if slugs else None128            out.append(Product(129                store_id=self.store_id,130                external_id=sku,131                url=url or wall_url,132                title=title,133                description=it.get("description") or "",134                price=price,135                price_max=price,136                images=[it["imageUrl"]] if it.get("imageUrl") else [],137                product_type=it.get("itemType") or "",138                available=True,139            ))140        return out141