"""Unit normalisation for grocery listings (CAD; g / ml / unit). Re-implements and extends Food-Ka's parser (data/raw/foodka-src/foodka/normalize.py) so that the research pipeline does not depend on production code. Differences are documented in reports/phase1_data_audit.md. """ from __future__ import annotations import re, unicodedata from dataclasses import dataclass _UNITS = {"kg": (1000.0, "g"), "g": (1.0, "g"), "mg": (0.001, "g"), "lb": (453.592, "g"), "lbs": (453.592, "g"), "oz": (28.3495, "g"), "l": (1000.0, "ml"), "ml": (1.0, "ml"), "cl": (10.0, "ml"), "un": (1.0, "un"), "unite": (1.0, "un"), "unites": (1.0, "un"), "ea": (1.0, "un"), "each": (1.0, "un"), "pk": (1.0, "un"), "ch": (1.0, "un"), "pc": (1.0, "un"), "pcs": (1.0, "un"), "units": (1.0, "un"), "unit": (1.0, "un")} _SIZE_RE = re.compile(r"(?:(\d+)\s*[x×*]\s*)?(\d+(?:[.,]\d+)?)\s*(kg|g|mg|lb|lbs|oz|ml|cl|l|un|unites?|ea|each|pk|ch|pcs?|units?)\b", re.I) _RANGE_RE = re.compile(r"(\d+(?:[.,]\d+)?)\s*(?:-|–|a|à)\s*(\d+(?:[.,]\d+)?)\s*(kg|g|lb|lbs|oz|ml|l)\b", re.I) _PERKG_RE = re.compile(r"/\s*(kg|lb|100\s*g|100\s*ml|l|ch|un)\b", re.I) _MULTI_RE = re.compile(r"^\s*(\d+)\s*(?:/|pour|for)\s*(?:(\d{1,4}(?:[.,]\d{1,2})?)\s*\$)?") _APPROX_RE = re.compile(r"environ|approx|moyenne|avg|about|~", re.I) _DEPOSIT_RE = re.compile(r"(cons(igne)?\.?\s*[\d,\.]+\s*\$\s*/\s*\w+\.?|\d+[,\.]\d+\s*\$\s*/\s*100\s*(g|ml))", re.I) _SOLD_INDIV_RE = re.compile(r"vendu individuellement|sold individually|a l'unite|à l'unité|l'unite|unite$", re.I) def strip_accents(s: str) -> str: return "".join(c for c in unicodedata.normalize("NFKD", s or "") if not unicodedata.combining(c)) @dataclass class Size: qty: float # in base unit (g, ml, un) base: str # 'g' | 'ml' | 'un' approx: bool = False source: str = "size_label" # where it was parsed from pack: int = 1 # multipack count ('12 x 355 ml' -> 12) def parse_size(label: str | None, name: str | None = None) -> Size | None: """Parse '450 g', '2 L', '12 x 355 ml', '6 un', 'Vendu individuellement', '1,36 kg'; falls back to the name.""" for text, src in ((label, "size_label"), (name, "name")): if not text: continue t = strip_accents(text).lower().replace("\u00a0", " ") m = _RANGE_RE.search(t) if m: # "1,2 - 1,5 kg" -> midpoint, approximate (checked before the simple pattern) a, b = float(m.group(1).replace(",", ".")), float(m.group(2).replace(",", ".")) factor, base = _UNITS[m.group(3).lower()] return Size((a + b) / 2 * factor, base, True, src) m = _SIZE_RE.search(t) if m: mult = int(m.group(1)) if m.group(1) else 1 qty = float(m.group(2).replace(",", ".")) factor, base = _UNITS[m.group(3).lower()] total = mult * qty * factor if total > 0: return Size(total, base, bool(_APPROX_RE.search(t)), src, mult) if src == "size_label" and _SOLD_INDIV_RE.search(t): return Size(1.0, "un", False, src) return None @dataclass class PriceLabel: per_weight_unit: str | None = None # 'kg','lb','100 g','100 ml','l' when the displayed price is per weight/volume multibuy_n: int | None = None # 'n / x $' multibuy_total: float | None = None def parse_price_label(label: str | None) -> PriceLabel: pl = PriceLabel() if not label: return pl t = strip_accents(label).lower().replace(" ", " ") m = _PERKG_RE.search(t) if m: pl.per_weight_unit = re.sub(r"\s+", " ", m.group(1)) m = _MULTI_RE.match(t) if m: pl.multibuy_n = int(m.group(1)) if m.group(2): pl.multibuy_total = float(m.group(2).replace(",", ".")) return pl def unit_price(price: float | None, size: Size | None, label: PriceLabel | None = None) -> tuple[float, str] | None: """Comparable price: $/100 g, $/100 ml or $/unit. Returns None when not computable.""" if price is None or price <= 0: return None if label and label.per_weight_unit: u = label.per_weight_unit if u == "kg": return (round(price / 10.0, 6), "100 g") if u == "lb": return (round(price / 4.53592, 6), "100 g") if u == "100 g": return (price, "100 g") if u == "100 ml": return (price, "100 ml") if u == "l": return (round(price / 10.0, 6), "100 ml") if size is None: return None if size.base == "un": return (round(price / size.qty, 6), "un") return (round(price / size.qty * 100.0, 6), f"100 {size.base}") def size_residual(label: str | None) -> str: """Text of the size label that is NOT the package size itself (e.g. 'calibre 21-25', 'gros', 'bio'), with deposit and unit-price fragments removed — used as extra discriminating tokens in matching.""" if not label: return "" t = strip_accents(label).lower().replace("\u00a0", " ") t = _DEPOSIT_RE.sub(" ", t) t = _RANGE_RE.sub(" ", t); t = _SIZE_RE.sub(" ", t) t = re.sub(r"[^a-z0-9%]+", " ", t) return " ".join(sorted(w for w in t.split() if len(w) > 1))