# ----------------------------------------------------------------------------- # Food-Ka — Agrégateur de produits d'épicerie (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # schema.py : modèle de données standardisé (Product) + enrichissement central # ----------------------------------------------------------------------------- """Schéma standard d'un produit d'épicerie (Product) et normalisation. Chaque connecteur, peu importe la bannière (Metro, IGA, Maxi…), doit produire des objets `Product` conformes à ce schéma. La méthode `finalize()` applique ensuite la couche de normalisation commune (foodka/normalize.py) : prix, prix unitaire $/100 g, catégorie canonique… — les connecteurs restent simples et remplissent les champs bruts. """ from __future__ import annotations import hashlib import json from dataclasses import dataclass, field, asdict from .normalize import ( # ré-exportés pour les connecteurs clean_text, format_unit_price, normalize_category, parse_price, parse_size, strip_accents, unit_price, ) __all__ = [ "Product", "normalize_category", "parse_price", "parse_size", "clean_text", "strip_accents", ] @dataclass class Product: """Produit d'épicerie standardisé Food-Ka.""" source: str # id de la bannière (voir data/sources.json) external_id: str # identifiant/SKU chez la source url: str # fiche produit chez la source name: str = "" # ex. "Beurre d'arachide croquant" brand: str = "" # ex. "Kraft" category: str = "" # catégorie canonique Food-Ka category_raw: str = "" # taxonomie originale de la bannière size_label: str = "" # format affiché, ex. "500 g", "2 L" price: float | None = None # prix courant ($ CAD), rabais inclus regular_price: float | None = None # prix régulier si le produit est en solde price_label: str = "" # texte original (ex. "2 / 5,00 $") on_sale: bool = False # produit en promotion unit_price: float | None = None # prix par unité comparable unit_price_label: str = "" # ex. "0,70 $ / 100 g" in_stock: bool | None = None # None = inconnu description: str = "" keywords: list[str] = field(default_factory=list) # tags source (bio, sans gluten…) details: dict = field(default_factory=dict) # champs structurés (JSON) images: list[str] = field(default_factory=list) # URLs absolues @property def uid(self) -> str: return f"{self.source}:{self.external_id}" def content_hash(self) -> str: """Hash du contenu pour la détection de changements (pseudo-webhook).""" payload = asdict(self) blob = json.dumps(payload, sort_keys=True, ensure_ascii=False) return hashlib.sha256(blob.encode("utf-8")).hexdigest() def finalize(self) -> "Product": """Applique la normalisation commune. Appelé par le pipeline d'ingestion. Idempotent ; ne remplace jamais une valeur explicite du connecteur. """ self.name = clean_text(self.name) self.brand = clean_text(self.brand) self.size_label = clean_text(self.size_label) self.description = clean_text(self.description) if not self.category: self.category = normalize_category(self.category_raw) if self.price is None: self.price = parse_price(self.price_label) # cohérence solde : un "prix régulier" égal ou inférieur au prix # courant n'apporte rien — on le laisse tomber. if self.regular_price is not None and self.price is not None: if self.regular_price <= self.price: self.regular_price = None else: self.on_sale = True if self.regular_price is None and not self.on_sale: self.on_sale = False if self.unit_price is None: up = unit_price(self.price, self.size_label) if up: self.unit_price = up[0] if not self.unit_price_label: self.unit_price_label = format_unit_price(up) # garde-fou : prix absurdes (0 $, > 2000 $) = donnée cassée, pas un prix if self.price is not None and not (0 < self.price <= 2000): self.price = None self.regular_price = None self.on_sale = False return self