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 + full-name token_sort_ratio >= 0.92 + identical numeric tokens + brand10 tokens of the branded side present in the brandless name11Excluded from matching: listings without a parsable package size (circular items are mostly size-less) and12multi-product circular offers ("X ou Y", "assortis", "sélectionnés", comma lists). A first version using13rapidfuzz token_set_ratio was rejected because a name that is a token-subset of another scores 1.014("beurre végétal salé" ~ "beurre végétal non salé"); see data/processed/match_validation_sample_v0_tokenset_rejected.csv.15Pairs are formed only across different banners (same company allowed, flagged `same_company`); same-banner links16between the storefront feed and the circular feed are recorded separately (`link_type = 'channel'`).17Outputs: data/processed/product_matches.parquet (pairs), match_groups_{tier}.parquet (union-find groups), and a18stratified validation sample (data/processed/match_validation_sample.csv) for manual + LLM annotation.19"""20from __future__ import annotations21import re, unicodedata, itertools, hashlib22import numpy as np, pandas as pd23from rapidfuzz import fuzz, process24from src.common import path, get_logger, ledger, seed_everything, SEED, SETTINGS25from src.cleaning.units import size_residual2627log = get_logger("match")28_ALNUM = re.compile(r"[^a-z0-9]+")29_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")30_NUM = re.compile(r"\d+(?:[.,]\d+)?")31STOP = {"de","du","des","la","le","les","l","d","et","a","au","aux","en","pour","the","of","and","with","un","une",32 "sans","avec","ou","or","pc","chaque","each","marque","brand"}33GTIN_FEEDS = {"metro", "superc"}34MULTI_RE = re.compile(r"\b(ou|or|assortis?|assorted|selectionne?s?|varietes?|au choix|et plus|et autres)\b|,|/", re.I)3536def strip_accents(s: str) -> str:37 return "".join(c for c in unicodedata.normalize("NFKD", s or "") if not unicodedata.combining(c))3839def brand_key(b: str) -> str:40 k = _ALNUM.sub("", strip_accents(b or "").lower())41 return k if len(k) >= 2 else ""4243def name_key(name: str, brand: str = "") -> str:44 s = strip_accents(name or "").lower().replace(" ", " ")45 s = _SIZE_TOK.sub(" ", s); s = _ALNUM.sub(" ", s)46 drop = set(STOP)47 if brand:48 drop |= set(_ALNUM.sub(" ", strip_accents(brand).lower()).split())49 toks = [t for t in s.split() if t and t not in drop]50 return " ".join(sorted(toks))5152def num_tokens(nk: str) -> frozenset:53 return frozenset(x.replace(",", ".") for x in _NUM.findall(nk))5455def size_bucket(qty: float, base: str, pack=1) -> str:56 """Package key: total quantity in base unit AND multipack count ('12x355 ml' != '6x710 ml')."""57 if pd.isna(qty) or base is None: return ""58 pk = 1 if pd.isna(pack) else int(pack)59 return f"{round(float(qty), 1):g}|{base}|x{pk}"6061def prepare(prod: pd.DataFrame, ret: pd.DataFrame) -> pd.DataFrame:62 excl = set(SETTINGS["scope"]["exclude_sources"])63 p = prod[~prod.source.isin(excl)].copy()64 p = p[p.name.fillna("").str.len() > 0]65 p = p.merge(ret[["feed_id", "banner", "company", "channel"]], left_on="source", right_on="feed_id", how="left")66 p["bkey"] = p.brand.fillna("").map(brand_key)67 resid = p.size_label.fillna("").map(size_residual)68 p["nkey"] = [(name_key(n, b) + " " + r).strip() for n, b, r in zip(p.name.fillna(""), p.brand.fillna(""), resid)]69 p["nkey"] = p.nkey.map(lambda k: " ".join(sorted(k.split())))70 p["nkey_full"] = [(name_key(n, "") + " " + r).strip() for n, r in zip(p.name.fillna(""), resid)]71 p["nkey_full"] = p.nkey_full.map(lambda k: " ".join(sorted(k.split())))72 p["ntoks"] = p.nkey.map(num_tokens)73 p["skey"] = [size_bucket(q, b, k) for q, b, k in zip(p.size_qty, p.size_base, p.size_pack)]74 p["gtin"] = np.where(p.source.isin(GTIN_FEEDS) & p.gtin_valid, p.external_id.str.lstrip("0"), None)75 p["multi_offer"] = p.name.fillna("").map(lambda n: bool(MULTI_RE.search(strip_accents(n))))76 return p7778def tier_a(p: pd.DataFrame) -> pd.DataFrame:79 g = p[p.gtin.notna()]80 pairs = []81 for gt, grp in g.groupby("gtin"):82 if len(grp) < 2: continue83 for i, j in itertools.combinations(grp.index, 2):84 if grp.at[i, "banner"] == grp.at[j, "banner"] and grp.at[i, "channel"] == grp.at[j, "channel"]: continue85 pairs.append((grp.at[i, "uid"], grp.at[j, "uid"], "A", 1.0, True, True, 1.0))86 return pd.DataFrame(pairs, columns=["uid_a", "uid_b", "tier", "score", "brand_match", "size_match", "name_sim"])8788def _block_pairs(block: pd.DataFrame, thr: float) -> list:89 out = []90 if len(block) < 2 or len(block) > 4000: return out91 names = block.nkey.tolist(); uids = block.uid.tolist(); banners = block.banner.tolist(); chans = block.channel.tolist()92 ntoks = block.ntoks.tolist(); tsets = [frozenset(n.split()) for n in names]93 sim = process.cdist(names, names, scorer=fuzz.token_sort_ratio, workers=-1) / 100.094 iu = np.triu_indices(len(block), k=1)95 for i, j in zip(*iu):96 s = sim[i, j]97 if s < thr: continue98 if ntoks[i] != ntoks[j]: continue99 if banners[i] == banners[j] and chans[i] == chans[j]: continue100 out.append((uids[i], uids[j], float(s), tsets[i] == tsets[j]))101 return out102103def tier_bc(p: pd.DataFrame) -> pd.DataFrame:104 rows = []105 q = p[(p.bkey != "") & (p.skey != "") & (~p.multi_offer)]106 for (bk, sk), block in q.groupby(["bkey", "skey"]):107 for ua, ub, s, eq in _block_pairs(block, 0.90):108 rows.append((ua, ub, "B" if eq else "C", s, True, True, s))109 return pd.DataFrame(rows, columns=["uid_a", "uid_b", "tier", "score", "brand_match", "size_match", "name_sim"])110111def tier_d(p: pd.DataFrame) -> pd.DataFrame:112 """Brandless feeds: block on size, compare full names (brand kept in the name), require the branded side's brand113 tokens to appear in the brandless name."""114 rows = []115 q = p[(p.skey != "") & (~p.multi_offer)]116 for sk, block in q.groupby("skey"):117 bl = block[block.bkey == ""]; br = block[block.bkey != ""]118 if len(bl) == 0 or len(br) == 0 or len(block) > 6000: continue119 sim = process.cdist(bl.nkey_full.tolist(), br.nkey_full.tolist(), scorer=fuzz.token_sort_ratio, workers=-1) / 100.0120 ii, jj = np.where(sim >= 0.92)121 for i, j in zip(ii, jj):122 a, b = bl.iloc[i], br.iloc[j]123 if a.banner == b.banner and a.channel == b.channel: continue124 if num_tokens(a.nkey_full) != num_tokens(b.nkey_full): continue125 btoks = [t for t in _ALNUM.sub(" ", strip_accents(b.brand).lower()).split() if len(t) >= 3]126 if btoks and not all(t in a.nkey_full.split() for t in btoks): continue127 rows.append((a.uid, b.uid, "D", float(sim[i, j]), False, True, float(sim[i, j])))128 return pd.DataFrame(rows, columns=["uid_a", "uid_b", "tier", "score", "brand_match", "size_match", "name_sim"])129130def union_groups(pairs: pd.DataFrame, tiers: set[str]) -> pd.DataFrame:131 parent = {}132 def find(x):133 while parent.setdefault(x, x) != x:134 parent[x] = parent[parent[x]]; x = parent[x]135 return x136 for ua, ub in pairs.loc[pairs.tier.isin(tiers), ["uid_a", "uid_b"]].itertuples(index=False):137 ra, rb = find(ua), find(ub)138 if ra != rb: parent[ra] = rb139 rows = [(hashlib.md5(find(u).encode()).hexdigest()[:12], u) for u in parent]140 return pd.DataFrame(rows, columns=["group_id", "uid"])141142def main() -> None:143 seed_everything()144 prod = pd.read_parquet(path("processed") / "products.parquet")145 ret = pd.read_parquet(path("processed") / "retailers.parquet")146 p = prepare(prod, ret)147 log.info("matching universe: %d listings (active+inactive, excl. %s); with size %d; multi-product offers excluded %d",148 len(p), SETTINGS["scope"]["exclude_sources"], (p.skey != "").sum(), p.multi_offer.sum())149 parts = [tier_a(p), tier_bc(p), tier_d(p)]150 pairs = pd.concat(parts, ignore_index=True)151 # keep the best tier per unordered pair152 pairs["key"] = [tuple(sorted(x)) for x in zip(pairs.uid_a, pairs.uid_b)]153 order = {"A": 0, "B": 1, "C": 2, "D": 3}154 pairs["rank"] = pairs.tier.map(order)155 pairs = pairs.sort_values(["key", "rank"]).drop_duplicates("key").drop(columns=["key", "rank"])156 meta = p.set_index("uid")[["banner", "company", "channel", "source", "category", "active"]]157 for side in ("a", "b"):158 m = meta.loc[pairs[f"uid_{side}"]].reset_index(drop=True); m.columns = [f"{c}_{side}" for c in m.columns]159 pairs = pd.concat([pairs.reset_index(drop=True), m], axis=1)160 pairs["same_company"] = pairs.company_a == pairs.company_b161 pairs["link_type"] = np.where(pairs.banner_a == pairs.banner_b, "channel", "cross_banner")162 pairs["both_active"] = (pairs.active_a == 1) & (pairs.active_b == 1)163 pairs.to_parquet(path("processed") / "product_matches.parquet", index=False)164 log.info("pairs: %s ; cross-banner by tier: %s", len(pairs),165 pairs[pairs.link_type == "cross_banner"].tier.value_counts().to_dict())166 for name, tiers in {"A": {"A"}, "AB": {"A", "B"}, "ABC": {"A", "B", "C"}, "ABCD": {"A", "B", "C", "D"}}.items():167 g = union_groups(pairs[pairs.link_type == "cross_banner"], tiers)168 g = g.merge(meta[["banner"]], left_on="uid", right_index=True)169 nb = g.groupby("group_id").banner.nunique()170 g = g[g.group_id.isin(nb[nb >= 2].index)]171 g.to_parquet(path("processed") / f"match_groups_{name}.parquet", index=False)172 log.info("groups tiers %-4s: %d groups, %d listings, median banners/group %.0f, max %d", name,173 g.group_id.nunique(), len(g), nb[nb >= 2].median(), nb.max())174 # Food-Ka's own links for comparison175 fk = pd.read_parquet(path("interim") / "raw_product_links.parquet")176 fkpairs = set()177 for gid, grp in fk.groupby("group_id"):178 for a, b in itertools.combinations(sorted(grp.uid), 2): fkpairs.add((a, b))179 ours = set(tuple(sorted(x)) for x in zip(pairs.uid_a, pairs.uid_b))180 log.info("Food-Ka product_links pairs: %d ; recovered by our tiers A-D: %d (%.1f%%)", len(fkpairs),181 len(fkpairs & ours), 100 * len(fkpairs & ours) / max(len(fkpairs), 1))182 # stratified validation sample: 40 per tier (A,B,C,D) × link_type cross_banner, with listing text for annotators183 rng = np.random.default_rng(SEED)184 cb = pairs[pairs.link_type == "cross_banner"]185 samp = cb.groupby("tier", group_keys=False).apply(lambda g: g.sample(min(len(g), 60), random_state=SEED))186 txt = prod.set_index("uid")[["name", "brand", "size_label", "price", "source"]]187 for side in ("a", "b"):188 t = txt.loc[samp[f"uid_{side}"]].reset_index(drop=True); t.columns = [f"{c}_{side}_txt" for c in t.columns]189 samp = pd.concat([samp.reset_index(drop=True), t], axis=1)190 samp = samp.sample(frac=1, random_state=SEED).reset_index(drop=True)191 samp.insert(0, "sample_id", range(1, len(samp) + 1))192 samp["label_llm"] = ""; samp["label_human"] = ""; samp["note"] = ""193 samp.to_csv(path("processed") / "match_validation_sample.csv", index=False)194 log.info("validation sample: %d pairs (%s)", len(samp), samp.tier.value_counts().to_dict())195 ledger("1", "match_products", f"{len(pairs)} pairs; sample {len(samp)}")196197if __name__ == "__main__":198 main()199