spb/fabri-ka

1# -----------------------------------------------------------------------------2# Fabri-Ka — Agrégateur de produits québécois3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# connectors/votresite.py : connecteur Votresite.ca (plateforme québécoise).5#   Les sites Votresite.ca sont des Drupal 8 gérés (thème owebo-votresite) avec6#   une boutique OpenCart montée dans un sous-répertoire (/boutique/ ou7#   /produits/, parfois autre). Rendu 100 % serveur, aucune API JSON :8#     - inventaire : <mount>/sitemap.xml (extension OpenCart, pas toujours9#       activée) sinon crawl des catégories <mount>/fr + pagination10#       ?limit=100&page=N ;11#     - fiche : URL …/<slug>-p<ID>[c<ID>…]/ — prix courant dans <h2>…$</h2>,12#       prix barré (spécial) dans <span style="text-decoration: line-through">,13#       « Disponibilité: » / « Modèle: » en <li>, description div#tab-description,14#       galerie <a class="thumbnail">. Les meta twitter:data1/2 servent de15#       secours mais affichent le prix RÉGULIER même en solde.16#   Dédup par ID produit (le même p<ID> apparaît sous plusieurs chemins de17#   catégorie, ex. -p361c37c45c44).18# -----------------------------------------------------------------------------19from __future__ import annotations2021import concurrent.futures as cf22import re23from html import unescape2425from ..schema import Product, parse_price26from .base import BaseConnector2728# …/slug-p361/ ou …/slug-p361c37c45/ (produit) ; …/slug-c116c139/ (catégorie)29PROD_URL_RE = re.compile(r"/([^/]+)-p(\d+)(?:c\d+)*/?(?:\?.*)?$")30CAT_URL_RE = re.compile(r"-c\d+(?:c\d+)*/?$")31LOC_RE = re.compile(r"<loc>\s*(?:<!\[CDATA\[)?\s*(.*?)\s*(?:\]\]>)?\s*</loc>",32                    re.I | re.S)33HREF_RE = re.compile(r'href="([^"]+)"')34SKIP_RE = re.compile(r"(/account|/cart|/checkout|/wishlist|/compare|/login|"35                     r"/register|route=|mailto:|tel:|javascript:)", re.I)3637# marqueurs OpenCart d'une boutique Votresite valide38_OC_MARKERS = ("image/cache", "route=product", 'class="thumbnails"',39               "list-unstyled")4041CANDIDATE_MOUNTS = ["/boutique", "/produits"]424344def _clean(s: str | None) -> str:45    return unescape(re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", s or ""))).strip()464748def _enc(url: str) -> str:49    """Les chemins d'images OpenCart contiennent espaces et accents bruts."""50    return url.replace(" ", "%20")515253class VotresiteConnector(BaseConnector):54    platform = "votresite"55    request_delay = 0.256    max_products = 200057    max_listing_pages = 250    # garde-fou du crawl catégories5859    # ------------------------------------------------------------------ mount60    def _try_mount(self, mount: str) -> bool:61        try:62            r = self.get(f"{self.base}{mount}/fr")63        except Exception:64            return False65        html = r.text or ""66        return (r.status_code == 20067                and any(m in html for m in _OC_MARKERS)68                and (PROD_URL_RE.search(html) is not None69                     or re.search(r'-[pc]\d+[c\d]*/"', html) is not None70                     or "-c" in html))7172    def _discover_mount(self) -> str:73        ep = (self.store.get("catalog_endpoint") or "").strip().rstrip("/")74        candidates: list[str] = []75        if ep and ep != "__generic__":76            candidates.append(ep if ep.startswith("/") else f"/{ep}")77        # page d'accueil Drupal : le lien vers la boutique révèle le montage78        try:79            html = self.get(self.base + "/").text80            for m in re.finditer(81                    r'href="https?://[^"/]+(/[a-z0-9_-]+)(?:/fr)?/?"', html, re.I):82                p = m.group(1)83                if re.search(r"boutique|produit|magasin|commander|achat", p, re.I):84                    candidates.append(p)85            for m in re.finditer(86                    r'href="https?://[^"/]+(/[a-z0-9_-]+)/fr/[^"]*-[pc]\d', html):87                candidates.append(m.group(1))88        except Exception:89            pass90        candidates += CANDIDATE_MOUNTS91        seen: set[str] = set()92        for c in candidates:93            c = c.rstrip("/")94            if not c or c in seen:95                continue96            seen.add(c)97            if self._try_mount(c):98                return c99        raise RuntimeError("boutique Votresite (OpenCart) introuvable "100                           f"(montages essayés : {sorted(seen)})")101102    # -------------------------------------------------------------- inventaire103    def _sitemap_urls(self, mount: str) -> dict[str, str]:104        """{pid: url produit} depuis <mount>/sitemap.xml (souvent absent)."""105        out: dict[str, str] = {}106        try:107            r = self.get(f"{self.base}{mount}/sitemap.xml")108        except Exception:109            return out110        xml = r.text or ""111        if r.status_code != 200 or "<urlset" not in xml:112            return out113        for loc in LOC_RE.findall(xml):114            m = PROD_URL_RE.search(loc)115            if m:116                out.setdefault(m.group(2), loc)117        return out118119    def _crawl_urls(self, mount: str) -> dict[str, str]:120        """Crawl des pages catégories : {pid: url produit}."""121        root = f"{self.base}{mount}"122        prods: dict[str, str] = {}123        seen: set[str] = set()124        queue: list[str] = [f"{root}/fr"]125        fetched = 0126        while queue and fetched < self.max_listing_pages:127            url = queue.pop(0)128            key = url.rstrip("/")129            if key in seen:130                continue131            seen.add(key)132            try:133                html = self.get(url).text or ""134            except Exception:135                continue136            fetched += 1137            for href in HREF_RE.findall(html):138                href = unescape(href).split("#")[0]139                if not href.startswith(root) or SKIP_RE.search(href):140                    continue141                pm = PROD_URL_RE.search(href.split("?")[0])142                if pm:143                    prods.setdefault(pm.group(2), href.split("?")[0])144                    continue145                path = href.split("?")[0]146                if CAT_URL_RE.search(path):147                    # 100 produits par page de liste (défaut : 15)148                    if path.rstrip("/") + "/?limit=100" not in seen:149                        queue.append(path.rstrip("/") + "/?limit=100")150                elif "page=" in href and "limit=" in href:151                    queue.append(href)     # pagination déjà paramétrée152            if len(prods) >= self.max_products:153                break154        return prods155156    # ------------------------------------------------------------------ fiche157    def _parse_product(self, pid: str, url: str, html: str) -> Product | None:158        # le 1er <h1> est souvent le logo du site (un lien) ; le titre du159        # produit est le dernier <h1> SANS ancre (ex. lemieldabee.ca)160        title, h1_pos = None, -1161        h1s = list(re.finditer(r"<h1[^>]*>(.*?)</h1>", html, re.S))162        plain = [m for m in h1s if "<a " not in m.group(1)]163        chosen = (plain or h1s)[-1] if h1s else None164        if chosen and plain:165            title, h1_pos = _clean(chosen.group(1)), chosen.start()166        if not title:167            m = re.search(r'<meta[^>]+property="og:title"[^>]+content="([^"]+)"',168                          html)169            title = _clean(m.group(1)) if m else None170            if chosen:171                h1_pos = chosen.start()172        if not title:173            return None174175        # zone d'info produit : entre le <h1> et le formulaire d'achat176        if h1_pos < 0:177            h1_pos = 0178        zone_end = html.find('<div id="product"', h1_pos)179        zone = html[h1_pos: zone_end if zone_end > h1_pos else h1_pos + 6000]180181        compare = None182        m = re.search(r"line-through[^>]*>\s*([^<]*\d[^<]*)</", zone)183        if m:184            compare = parse_price(m.group(1))185        price = None186        m = re.search(r"<h2[^>]*>\s*([^<]*\d[^<]*\$[^<]*)</h2>", zone)187        if m:188            price = parse_price(m.group(1))189        if price is None:      # secours : meta twitter (prix régulier)190            m = re.search(r'twitter:data1"\s+content="([^"]+)"', html)191            if m:192                price = parse_price(m.group(1))193        if compare is not None and price is not None and compare <= price:194            compare = None195196        available = None197        m = (re.search(r"Disponibilité\s*:\s*</?[^>]*>?\s*([^<]+)<", zone)198             or re.search(r'twitter:data2"\s+content="([^"]+)"', html))199        dispo = _clean(m.group(1)) if m else ""200        if dispo:201            low = dispo.lower()202            if "rupture" in low or "épuis" in low or "epuis" in low:203                available = False204            elif "stock" in low or "commande" in low or "jour" in low:205                available = True206207        det: dict = {}208        m = re.search(r"Modèle\s*:\s*([^<]+)<", zone)209        if m and _clean(m.group(1)):210            det["model"] = _clean(m.group(1))211        if dispo:212            det["availability_text"] = dispo213214        # options (listes déroulantes OpenCart)215        opts = []216        for sel in re.finditer(217                r'<label[^>]*control-label[^>]*>(.*?)</label>\s*<select[^>]*'218                r'name="option\[\d+\]"[^>]*>(.*?)</select>', html, re.S):219            name = _clean(sel.group(1))220            values = [_clean(v) for v in221                      re.findall(r"<option[^>]*>(.*?)</option>", sel.group(2), re.S)]222            values = [v for v in values if v and "choisir" not in v.lower()]223            if name and values:224                opts.append({"name": name, "values": values[:30]})225        if opts:226            det["options"] = opts227228        # description : div#tab-description jusqu'au prochain onglet/bloc229        desc = ""230        start = html.find('id="tab-description"')231        if start != -1:232            start = html.find(">", start) + 1233            cuts = [html.find(marker, start) for marker in234                    ('id="tab-', 'class="related', "<footer", "Produits apparentés")]235            cuts = [c for c in cuts if c > start]236            desc = html[start: min(cuts) if cuts else start + 20000]237        if not _clean(desc):238            m = (re.search(r'<meta[^>]+property="og:description"[^>]+content="([^"]+)"', html)239                 or re.search(r'<meta[^>]+name="description"[^>]+content="([^"]+)"', html))240            desc = m.group(1) if m else ""241242        images: list[str] = []243        for m in re.finditer(r'<a[^>]+class="thumbnail"[^>]+href="([^"]+)"', html):244            u = _enc(unescape(m.group(1)))245            if u.startswith("http") and u not in images:246                images.append(u)247        m = re.search(r'<meta[^>]+property="og:image"[^>]+content="([^"]+)"', html)248        if m:249            u = _enc(unescape(m.group(1)))250            if u.startswith("http") and u not in images:251                images.append(u)252253        # catégories du fil d'Ariane (sans l'accueil ni le produit lui-même)254        cats: list[str] = []255        m = re.search(r'<ul class="breadcrumb"[^>]*>(.*?)</ul>', html, re.S)256        if m:257            entries = re.findall(r"<a[^>]*>(.*?)</a>", m.group(1), re.S)258            cats = [_clean(e) for e in entries259                    if _clean(e) and "fa-home" not in e and _clean(e) != title]260261        return Product(262            store_id=self.store_id,263            external_id=pid,264            url=url,265            title=title,266            description=desc,267            price=price,268            price_max=price,269            compare_at_price=compare,270            images=images,271            product_type=cats[-1] if cats else "",272            tags=cats,273            available=available,274            details=det,275        )276277    # ------------------------------------------------------------------ fetch278    def fetch(self) -> list[Product]:279        mount = self._discover_mount()280        urls = self._sitemap_urls(mount)281        if len(urls) < 3:          # sitemap absent ou quasi vide -> crawl282            crawled = self._crawl_urls(mount)283            for pid, u in crawled.items():284                urls.setdefault(pid, u)285        items = list(urls.items())[: self.max_products]286        out: list[Product] = []287288        def work(item: tuple[str, str]) -> Product | None:289            pid, u = item290            try:291                html = self.get(u).text or ""292            except Exception:293                return None294            if len(html) < 500:295                return None296            return self._parse_product(pid, u, html)297298        with cf.ThreadPoolExecutor(6) as ex:299            for rec in ex.map(work, items):300                if rec:301                    out.append(rec)302        return out303