"""Phase 2.2b — Mirror / sibling domains: one business publishing the same catalogue under two store ids
(language mirrors en.X / X, .ca / .com twins, rebranded domains). They would double-count assortments and create
within-firm 'cross-merchant' price pairs.

Rule: two live stores are siblings if (a) their domain stems match after stripping language/www prefixes, TLDs and the
tokens 'votre|boutique|shop|store', OR (b) the Jaccard overlap of their normalised-title sets is >= 0.9 of the smaller
catalogue and the smaller catalogue has >= 10 listings. Connected components give sibling groups; the representative is
the store with the most listings (ties: alphabetical). Output: data/interim/sibling_stores.parquet
"""
from __future__ import annotations
import itertools, json, re
import pandas as pd, networkx as nx
from src.common import ROOT, P, get_logger, ledger, register_output, write_json

log = get_logger("clean.siblings")


def stem(domain: str) -> str:
    d = domain.lower()
    d = re.sub(r"^(www|en|fr|shop|boutique|store|ca|us)\.", "", d)
    d = re.sub(r"\.(com|ca|co|net|org|quebec|qc\.ca|io|shop|store|biz|info)$", "", d)
    d = re.sub(r"^(votre|boutique|shop|store|la|le|les)[-]?", "", d)
    d = re.sub(r"[-](boutique|shop|store|quebec|qc|canada|ca)$", "", d)
    return re.sub(r"[^a-z0-9]", "", d)


def main():
    li = pd.read_parquet(P["interim"] / "listings_clean.parquet")
    st = pd.read_parquet(P["interim"] / "stores_raw.parquet")
    live = st[(st.enabled == 1) & (st.product_count > 0)].id.tolist()
    a = li[li.f_published & li.f_live_store & li.f_dup_rep]
    titles = a.groupby("store_id").title_norm.apply(lambda s: frozenset(x for x in s if len(x) >= 8))
    sizes = titles.map(len)
    G = nx.Graph(); G.add_nodes_from(live)
    reasons = []
    by_stem = pd.Series(live, index=[stem(s) for s in live])
    for s, grp in by_stem.groupby(level=0):
        ids = list(grp.values)
        if len(ids) > 1 and len(s) >= 4:
            for x, y in itertools.combinations(ids, 2):
                G.add_edge(x, y); reasons.append((x, y, "domain_stem", None))
    # catalogue overlap — candidates = store pairs sharing >= 5 identical titles (via inverted index)
    inv = {}
    for sid, ts in titles.items():
        for t in ts: inv.setdefault(t, []).append(sid)
    pc = {}
    for t, sids in inv.items():
        if 1 < len(sids) <= 30:
            for x, y in itertools.combinations(sorted(sids), 2): pc[(x, y)] = pc.get((x, y), 0) + 1
    for (x, y), n in pc.items():
        if n < 5: continue
        small = min(sizes.get(x, 0), sizes.get(y, 0))
        if small >= 10 and n / small >= 0.9:
            G.add_edge(x, y); reasons.append((x, y, "title_overlap", round(n / small, 3)))
    rows = []
    for gid, comp in enumerate(nx.connected_components(G)):
        if len(comp) < 2: continue
        comp = sorted(comp, key=lambda s: (-sizes.get(s, 0), s))
        for s in comp: rows.append((gid, s, comp[0], s == comp[0], sizes.get(s, 0)))
    sib = pd.DataFrame(rows, columns=["sibling_group", "id", "sibling_rep", "is_rep", "n_titles"])
    sib.to_parquet(P["interim"] / "sibling_stores.parquet", index=False); register_output(P["interim"] / "sibling_stores.parquet", "siblings")
    pd.DataFrame(reasons, columns=["a", "b", "reason", "overlap"]).to_csv(ROOT / "experiments" / "sibling_pairs.csv", index=False)
    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,
            "non_rep_listings": int(a[a.store_id.isin(sib[~sib.is_rep].id)].shape[0]) if len(sib) else 0,
            "reasons": pd.Series([r[2] for r in reasons]).value_counts().to_dict(),
            "largest_groups": sib.groupby("sibling_group").id.apply(list).map(lambda l: l[:4]).head(12).tolist()}
    write_json(summ, ROOT / "experiments" / "siblings.json", "siblings"); ledger("clean.siblings", **summ)
    log.info(json.dumps(summ, ensure_ascii=False))


if __name__ == "__main__":
    main()