# -----------------------------------------------------------------------------
# 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