# ----------------------------------------------------------------------------- # Food-Ka — connecteur Marché Tau (marchestau.com — supermarchés santé) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # k-eCommerce : les grilles de catégories sont chargées en AJAX (aucun prix # dans le HTML), mais les FICHES produit sont rendues côté serveur # (« 16,99$ CAD », fil d'Ariane, marque, image). Stratégie : sitemap # kSitemap-1.xml (~14 000 URLs), filtrage des slugs produit (suffixe de # format « -227gr »), échantillon réparti sur tout le catalogue, puis une # requête par fiche (throttling de BaseConnector). # ----------------------------------------------------------------------------- from __future__ import annotations import re from bs4 import BeautifulSoup from ..schema import Product, normalize_category, parse_price from .base import BaseConnector BASE = "https://marchestau.com" SITEMAP = f"{BASE}/kSitemap-1.xml" _LOC_RE = re.compile(r"([^<]+)") # slug produit : se termine par un format, ex. « -227gr », « -1.36l », « -60un » _SLUG_SIZE_RE = re.compile( r"-(\d+(?:[.,]\d+)?)\s*(gr|g|kg|ml|l|lb|oz|un)$", re.IGNORECASE) _CODE_RE = re.compile(r"(\d{6,})") class TauConnector(BaseConnector): source_id = "tau" max_products: int = 180 # une requête par fiche : rester raisonnable def _product_urls(self) -> list[str]: """URLs de fiches produit, échantillonnées uniformément (tout l'alphabet).""" xml = self.get(SITEMAP).text urls = [u for u in _LOC_RE.findall(xml) if _SLUG_SIZE_RE.search(u.rsplit("/", 1)[-1])] if len(urls) <= self.max_products: return urls step = len(urls) / self.max_products return [urls[int(i * step)] for i in range(self.max_products)] def _parse_product(self, html: str, url: str) -> Product | None: soup = BeautifulSoup(html, "html.parser") h1 = soup.select_one(".product-detail h1") or soup.select_one("h1") if h1 is None: return None name = h1.get_text(" ", strip=True) if not name: return None code_el = soup.select_one(".product-details-code") code = "" if code_el is not None: m = _CODE_RE.search(code_el.get_text()) code = m.group(1) if m else "" slug = url.rstrip("/").rsplit("/", 1)[-1] external_id = code or slug # prix courant + prix régulier barré (solde) price_el = soup.select_one(".price-current") regular_el = soup.select_one( ".price-regular, .price-before, .single-price-display del, " ".single-price-display s") unit_el = soup.select_one(".price-per-unit small") # fil d'Ariane : Accueil > Catalogue > > > crumbs = [li.get_text(" ", strip=True) for li in soup.select("ul.breadcrumb li a")] crumbs = [c for c in crumbs if c and c not in ("Page d'accueil", "Catalogue")] category_raw = " / ".join(crumbs) brand_el = soup.select_one(".product-brand a") img = soup.select_one("#product-detail-gallery-main-img[src]") image = str(img["src"]) if img else "" # format depuis le slug : « mures-noires-227gr » -> « 227 g » size_label = "" m = _SLUG_SIZE_RE.search(slug) if m: unit = m.group(2).lower() size_label = f"{m.group(1)} {'g' if unit == 'gr' else unit}" desc_el = soup.select_one(".product-details-desc") return Product( source=self.source_id, external_id=external_id, url=url, name=name, brand=brand_el.get_text(" ", strip=True) if brand_el else "", category=normalize_category(category_raw), category_raw=category_raw, size_label=size_label, price_label=price_el.get_text(" ", strip=True) if price_el else "", regular_price=(parse_price(regular_el.get_text(" ", strip=True)) if regular_el else None), on_sale=regular_el is not None, unit_price_label=unit_el.get_text(" ", strip=True) if unit_el else "", description=desc_el.get_text(" ", strip=True) if desc_el else "", images=[image] if image.startswith("http") else [], ) def fetch(self) -> list[Product]: products: list[Product] = [] seen: set[str] = set() for url in self._product_urls(): try: html = self.get(url).text except Exception: # une fiche qui casse ne bloque pas le reste continue prod = self._parse_product(html, url) if prod is not None and prod.uid not in seen: seen.add(prod.uid) products.append(prod) return products