1"""Phase 1.4 — cross-banner product matching with confidence tiers.23Tiers (calibrated on UPC ground truth Metro<->Super C, src/matching/calibrate_thresholds.py, then frozen):4 A exact GS1-valid GTIN equality (only feeds whose external_id is a UPC: metro, superc — 100 % check-digit valid)5 B brand key equal + package size equal (0.1-unit bucket) + identical cleaned token sets + identical numeric tokens6 (labelled-set precision 0.913, recall 0.984)7 C brand key equal + size equal + token_sort_ratio >= 0.90 + identical numeric tokens, not already B8 (labelled-set precision 0.845 cumulative)9 D (brand missing on one side) size equal + identical full-name token sets (brand tokens kept on both sides) +10 identical numeric tokens + brand tokens of the branded side present in the brandless name11 (validation round 1 showed that a 0.92 sort-ratio rule let 'spaghetti' ~ 'spaghettini' through: 3/60 errors)12Excluded from matching: listings without a parsable package size (circular items are mostly size-less) and13multi-product circular offers ("X ou Y", "assortis", "sélectionnés", comma lists). A first version using14rapidfuzz token_set_ratio was rejected because a name that is a token-subset of another scores 1.015("beurre végétal salé" ~ "beurre végétal non salé"); see data/processed/match_validation_sample_v0_tokenset_rejected.csv.16Pairs are formed only across different banners (same company allowed, flagged `same_company`); same-banner links17between the storefront feed and the circular feed are recorded separately (`link_type = 'channel'`).18Outputs: data/processed/product_matches.parquet (pairs), match_groups_{tier}.parquet (union-find groups), and a19stratified validation sample (data/processed/match_validation_sample.csv) for manual + LLM annotation.20"""21from __future__ import annotations22import re, unicodedata, itertools, hashlib23import numpy as np, pandas as pd24from rapidfuzz import fuzz, process25from src.common import path, get_logger, ledger, seed_everything, SEED, SETTINGS26from src.cleaning.units import size_residual2728log = get_logger("match")29_ALNUM = re.compile(r"[^a-z0-9]+")30_SIZE_TOK = re.compile(r"\b\d+(?:[.,]\d+)?\s*(?:x|×)?\s*\d*(?:[.,]\d+)?\s*(?:kg|g|mg|lb|lbs|oz|ml|cl|l|un|unites?|ea|each|pk|ch)\b")31_NUM = re.compile(r"\d+(?:[.,]\d+)?")32STOP = {"de","du","des","la","le","les","l","d","et","a","au","aux","en","pour","the","of","and","with","un","une",33 "sans","avec","ou","or","pc","chaque","each","marque","brand"}34GTIN_FEEDS = {"metro", "superc"}35MULTI_RE = re.compile(r"\b(ou|or|assortis?|assorted|selectionne?s?|varietes?|au choix|et plus|et autres)\b|,|/", re.I)3637def strip_accents(s: str) -> str:38 return "".join(c for c in unicodedata.normalize("NFKD", s or "") if not unicodedata.combining(c))3940def brand_key(b: str) -> str:41 k = _ALNUM.sub("", strip_accents(b or "").lower())42 return k if len(k) >= 2 else ""4344def name_key(name: str, brand: str = "") -> str:45 s = strip_accents(name or "").lower().replace(" ", " ")46 s = _SIZE_TOK.sub(" ", s); s = _ALNUM.sub(" ", s)47 drop = set(STOP)48 if brand:49 drop |= set(_ALNUM.sub(" ", strip_accents(brand).lower()).split())50 toks = [t for t in s.split() if t and t not in drop]51 return " ".join(sorted(toks))5253def num_tokens(nk: str) -> frozenset:54 return frozenset(x.replace(",", ".") for x in _NUM.findall(nk))5556def size_bucket(qty: float, base: str, pack=1) -> str:57 """Package key: total quantity in base unit AND multipack count ('12x355 ml' != '6x710 ml')."""58 if pd.isna(qty) or base is None: return ""59 pk = 1 if pd.isna(pack) else int(pack)60 return f"{round(float(qty), 1):g}|{base}|x{pk}"6162def prepare(prod: pd.DataFrame, ret: pd.DataFrame) -> pd.DataFrame:63 excl = set(SETTINGS["scope"]["exclude_sources"])64 p = prod[~prod.source.isin(excl)].copy()65 p = p[p.name.fillna("").str.len() > 0]66 p = p.merge(ret[["feed_id", "banner", "company", "channel"]], left_on="source", right_on="feed_id", how="left")67 p["bkey"] = p.brand.fillna("").map(brand_key)68 resid = p.size_label.fillna("").map(size_residual)69 p["nkey"] = [(name_key(n, b) + " " + r).strip() for n, b, r in zip(p.name.fillna(""), p.brand.fillna(""), resid)]70 p["nkey"] = p.nkey.map(lambda k: " ".join(sorted(k.split())))71 p["nkey_full"] = [(name_key(n, "") + " " + r).strip() for n, r in zip(p.name.fillna(""), resid)]72 p["nkey_full"] = p.nkey_full.map(lambda k: " ".join(sorted(k.split())))73 p["ntoks"] = p.nkey.map(num_tokens)74 p["skey"] = [size_bucket(q, b, k) for q, b, k in zip(p.size_qty, p.size_base, p.size_pack)]75 p["gtin"] = np.where(p.source.isin(GTIN_FEEDS) & p.gtin_valid, p.external_id.str.lstrip("0"), None)76 p["multi_offer"] = p.name.fillna("").map(lambda n: bool(MULTI_RE.search(strip_accents(n))))77 return p7879def tier_a(p: pd.DataFrame) -> pd.DataFrame:80 g = p[p.gtin.notna()]81 pairs = []82 for gt, grp in g.groupby("gtin"):83 if len(grp) < 2: continue84 for i, j in itertools.combinations(grp.index, 2):85 if grp.at[i, "banner"] == grp.at[j, "banner"] and grp.at[i, "channel"] == grp.at[j, "channel"]: continue86 pairs.append((grp.at[i, "uid"], grp.at[j, "uid"], "A", 1.0, True, True, 1.0))87 return pd.DataFrame(pairs, columns=["uid_a", "uid_b", "tier", "score", "brand_match", "size_match", "name_sim"])8889def _block_pairs(block: pd.DataFrame, thr: float) -> list:90 out = []91 if len(block) < 2 or len(block) > 4000: return out92 names = block.nkey.tolist(); uids = block.uid.tolist(); banners = block.banner.tolist(); chans = block.channel.tolist()93 ntoks = block.ntoks.tolist(); tsets = [frozenset(n.split()) for n in names]94 sim = process.cdist(names, names, scorer=fuzz.token_sort_ratio, workers=-1) / 100.095 iu = np.triu_indices(len(block), k=1)96 for i, j in zip(*iu):97 s = sim[i, j]98 if s < thr: continue99 if ntoks[i] != ntoks[j]: continue100 if banners[i] == banners[j] and chans[i] == chans[j]: continue101 out.append((uids[i], uids[j], float(s), tsets[i] == tsets[j]))102 return out103104def tier_bc(p: pd.DataFrame) -> pd.DataFrame:105 rows = []106 q = p[(p.bkey != "") & (p.skey != "") & (~p.multi_offer)]107 for (bk, sk), block in q.groupby(["bkey", "skey"]):108 for ua, ub, s, eq in _block_pairs(block, 0.90):109 rows.append((ua, ub, "B" if eq else "C", s, True, True, s))110 return pd.DataFrame(rows, columns=["uid_a", "uid_b", "tier", "score", "brand_match", "size_match", "name_sim"])111112def tier_d(p: pd.DataFrame) -> pd.DataFrame:113 """Brandless feeds: block on size, compare full names (brand kept in the name), require the branded side's brand114 tokens to appear in the brandless name."""115 rows = []116 q = p[(p.skey != "") & (~p.multi_offer)]117 for sk, block in q.groupby("skey"):118 bl = block[block.bkey == ""]; br = block[block.bkey != ""]119 if len(bl) == 0 or len(br) == 0 or len(block) > 6000: continue120 sim = process.cdist(bl.nkey_full.tolist(), br.nkey_full.tolist(), scorer=fuzz.token_sort_ratio, workers=-1) / 100.0121 ii, jj = np.where(sim >= 0.92)122 for i, j in zip(ii, jj):123 a, b = bl.iloc[i], br.iloc[j]124 if a.banner == b.banner and a.channel == b.channel: continue125 if set(a.nkey_full.split()) != set(b.nkey_full.split()): continue126 if num_tokens(a.nkey_full) != num_tokens(b.nkey_full): continue127 btoks = [t for t in _ALNUM.sub(" ", strip_accents(b.brand).lower()).split() if len(t) >= 3]128 if btoks and not all(t in a.nkey_full.split() for t in btoks): continue129 rows.append((a.uid, b.uid, "D", float(sim[i, j]), False, True, float(sim[i, j])))130 return pd.DataFrame(rows, columns=["uid_a", "uid_b", "tier", "score", "brand_match", "size_match", "name_sim"])131132def union_groups(pairs: pd.DataFrame, tiers: set[str]) -> pd.DataFrame:133 parent = {}134 def find(x):135 while parent.setdefault(x, x) != x:136 parent[x] = parent[parent[x]]; x = parent[x]137 return x138 for ua, ub in pairs.loc[pairs.tier.isin(tiers), ["uid_a", "uid_b"]].itertuples(index=False):139 ra, rb = find(ua), find(ub)140 if ra != rb: parent[ra] = rb141 rows = [(hashlib.md5(find(u).encode()).hexdigest()[:12], u) for u in parent]142 return pd.DataFrame(rows, columns=["group_id", "uid"])143144def main() -> None:145 seed_everything()146 prod = pd.read_parquet(path("processed") / "products.parquet")147 ret = pd.read_parquet(path("processed") / "retailers.parquet")148 p = prepare(prod, ret)149 log.info("matching universe: %d listings (active+inactive, excl. %s); with size %d; multi-product offers excluded %d",150 len(p), SETTINGS["scope"]["exclude_sources"], (p.skey != "").sum(), p.multi_offer.sum())151 parts = [tier_a(p), tier_bc(p), tier_d(p)]152 pairs = pd.concat(parts, ignore_index=True)153 # keep the best tier per unordered pair154 pairs["key"] = [tuple(sorted(x)) for x in zip(pairs.uid_a, pairs.uid_b)]155 order = {"A": 0, "B": 1, "C": 2, "D": 3}156 pairs["rank"] = pairs.tier.map(order)157 pairs = pairs.sort_values(["key", "rank"]).drop_duplicates("key").drop(columns=["key", "rank"])158 meta = p.set_index("uid")[["banner", "company", "channel", "source", "category", "active"]]159 for side in ("a", "b"):160 m = meta.loc[pairs[f"uid_{side}"]].reset_index(drop=True); m.columns = [f"{c}_{side}" for c in m.columns]161 pairs = pd.concat([pairs.reset_index(drop=True), m], axis=1)162 pairs["same_company"] = pairs.company_a == pairs.company_b163 pairs["link_type"] = np.where(pairs.banner_a == pairs.banner_b, "channel", "cross_banner")164 pairs["both_active"] = (pairs.active_a == 1) & (pairs.active_b == 1)165 pairs.to_parquet(path("processed") / "product_matches.parquet", index=False)166 log.info("pairs: %s ; cross-banner by tier: %s", len(pairs),167 pairs[pairs.link_type == "cross_banner"].tier.value_counts().to_dict())168 for name, tiers in {"A": {"A"}, "AB": {"A", "B"}, "ABC": {"A", "B", "C"}, "ABCD": {"A", "B", "C", "D"}}.items():169 g = union_groups(pairs[pairs.link_type == "cross_banner"], tiers)170 g = g.merge(meta[["banner"]], left_on="uid", right_index=True)171 nb = g.groupby("group_id").banner.nunique()172 g = g[g.group_id.isin(nb[nb >= 2].index)]173 g.to_parquet(path("processed") / f"match_groups_{name}.parquet", index=False)174 log.info("groups tiers %-4s: %d groups, %d listings, median banners/group %.0f, max %d", name,175 g.group_id.nunique(), len(g), nb[nb >= 2].median(), nb.max())176 # Food-Ka's own links for comparison177 fk = pd.read_parquet(path("interim") / "raw_product_links.parquet")178 fkpairs = set()179 for gid, grp in fk.groupby("group_id"):180 for a, b in itertools.combinations(sorted(grp.uid), 2): fkpairs.add((a, b))181 ours = set(tuple(sorted(x)) for x in zip(pairs.uid_a, pairs.uid_b))182 log.info("Food-Ka product_links pairs: %d ; recovered by our tiers A-D: %d (%.1f%%)", len(fkpairs),183 len(fkpairs & ours), 100 * len(fkpairs & ours) / max(len(fkpairs), 1))184 # stratified validation sample: 40 per tier (A,B,C,D) × link_type cross_banner, with listing text for annotators185 rng = np.random.default_rng(SEED)186 cb = pairs[pairs.link_type == "cross_banner"]187 samp = cb.groupby("tier", group_keys=False).apply(lambda g: g.sample(min(len(g), 60), random_state=SEED))188 txt = prod.set_index("uid")[["name", "brand", "size_label", "price", "source"]]189 for side in ("a", "b"):190 t = txt.loc[samp[f"uid_{side}"]].reset_index(drop=True); t.columns = [f"{c}_{side}_txt" for c in t.columns]191 samp = pd.concat([samp.reset_index(drop=True), t], axis=1)192 samp = samp.sample(frac=1, random_state=SEED).reset_index(drop=True)193 samp.insert(0, "sample_id", range(1, len(samp) + 1))194 samp["label_llm"] = ""; samp["label_operator"] = ""; samp["note"] = ""195 target = path("processed") / "match_validation_sample.csv"196 if target.exists() and pd.read_csv(target).get("label_llm", pd.Series(dtype=str)).fillna("").ne("").any():197 # never overwrite an annotated sample (labels are hand-reviewed); write candidates alongside and report drift198 old = pd.read_csv(target); same = set(zip(old.uid_a, old.uid_b)) == set(zip(samp.uid_a, samp.uid_b))199 samp.to_csv(path("processed") / "match_validation_sample_candidates.csv", index=False)200 log.info("annotated validation sample preserved (%s); candidate sample identical to annotated: %s", target.name, same)201 else:202 samp.to_csv(target, index=False)203 log.info("validation sample: %d pairs (%s)", len(samp), samp.tier.value_counts().to_dict())204 ledger("1", "match_products", f"{len(pairs)} pairs; sample {len(samp)}")205206if __name__ == "__main__":207 main()208