1"""Phase 2.2b — Mirror / sibling domains: one business publishing the same catalogue under two store ids2(language mirrors en.X / X, .ca / .com twins, rebranded domains). They would double-count assortments and create3within-firm 'cross-merchant' price pairs.45Rule: two live stores are siblings if (a) their domain stems match after stripping language/www prefixes, TLDs and the6tokens 'votre|boutique|shop|store', OR (b) the Jaccard overlap of their normalised-title sets is >= 0.9 of the smaller7catalogue and the smaller catalogue has >= 10 listings. Connected components give sibling groups; the representative is8the store with the most listings (ties: alphabetical). Output: data/interim/sibling_stores.parquet9"""10from __future__ import annotations11import itertools, json, re12import pandas as pd, networkx as nx13from src.common import ROOT, P, get_logger, ledger, register_output, write_json, BASE_PROC, BASE_EXP1415log = get_logger("clean.siblings")161718def stem(domain: str) -> str:19 d = domain.lower()20 d = re.sub(r"^(www|en|fr|shop|boutique|store|ca|us)\.", "", d)21 d = re.sub(r"\.(com|ca|co|net|org|quebec|qc\.ca|io|shop|store|biz|info)$", "", d)22 d = re.sub(r"^(votre|boutique|shop|store|la|le|les)[-]?", "", d)23 d = re.sub(r"[-](boutique|shop|store|quebec|qc|canada|ca)$", "", d)24 return re.sub(r"[^a-z0-9]", "", d)252627def main():28 li = pd.read_parquet(P["interim"] / "listings_clean.parquet")29 st = pd.read_parquet(P["interim"] / "stores_raw.parquet")30 live = st[(st.enabled == 1) & (st.product_count > 0)].id.tolist()31 a = li[li.f_published & li.f_live_store & li.f_dup_rep]32 titles = a.groupby("store_id").title_norm.apply(lambda s: frozenset(x for x in s if len(x) >= 8))33 sizes = titles.map(len)34 G = nx.Graph(); G.add_nodes_from(live)35 reasons = []36 by_stem = pd.Series(live, index=[stem(s) for s in live])37 for s, grp in by_stem.groupby(level=0):38 ids = list(grp.values)39 if len(ids) > 1 and len(s) >= 4:40 for x, y in itertools.combinations(ids, 2):41 G.add_edge(x, y); reasons.append((x, y, "domain_stem", None))42 # catalogue overlap — candidates = store pairs sharing >= 5 identical titles (via inverted index)43 inv = {}44 for sid, ts in titles.items():45 for t in ts: inv.setdefault(t, []).append(sid)46 pc = {}47 for t, sids in inv.items():48 if 1 < len(sids) <= 30:49 for x, y in itertools.combinations(sorted(sids), 2): pc[(x, y)] = pc.get((x, y), 0) + 150 for (x, y), n in pc.items():51 if n < 5: continue52 small = min(sizes.get(x, 0), sizes.get(y, 0))53 if small >= 10 and n / small >= 0.9:54 G.add_edge(x, y); reasons.append((x, y, "title_overlap", round(n / small, 3)))55 rows = []56 for gid, comp in enumerate(nx.connected_components(G)):57 if len(comp) < 2: continue58 comp = sorted(comp, key=lambda s: (-sizes.get(s, 0), s))59 for s in comp: rows.append((gid, s, comp[0], s == comp[0], sizes.get(s, 0)))60 sib = pd.DataFrame(rows, columns=["sibling_group", "id", "sibling_rep", "is_rep", "n_titles"])61 sib.to_parquet(P["interim"] / "sibling_stores.parquet", index=False); register_output(P["interim"] / "sibling_stores.parquet", "siblings")62 pd.DataFrame(reasons, columns=["a", "b", "reason", "overlap"]).to_csv(BASE_EXP / "sibling_pairs.csv", index=False)63 summ = {"sibling_groups": int(sib.sibling_group.nunique()) if len(sib) else 0, "stores_in_groups": len(sib), "non_rep_stores": int((~sib.is_rep).sum()) if len(sib) else 0,64 "non_rep_listings": int(a[a.store_id.isin(sib[~sib.is_rep].id)].shape[0]) if len(sib) else 0,65 "reasons": pd.Series([r[2] for r in reasons]).value_counts().to_dict(),66 "largest_groups": sib.groupby("sibling_group").id.apply(list).map(lambda l: l[:4]).head(12).tolist()}67 write_json(summ, BASE_EXP / "siblings.json", "siblings"); ledger("clean.siblings", **summ)68 log.info(json.dumps(summ, ensure_ascii=False))697071if __name__ == "__main__":72 main()73