"""Phase 2.5 — Cross-merchant product matching (candidates for H5 price dispersion).

There is no GTIN in the source, so identity across merchants is inferred, tiered, and later validated on a labelled sample
(src/validation/match_validation.py). Only listings in the price sample are matched.

Tiers (pairs of listings in DIFFERENT stores):
  A  same merchant SKU (>= 4 chars, not a short pure number) and token_set_ratio(title) >= 70
  B  same normalised brand (vendor), identical normalised title
  C  same normalised brand, same category, token_sort_ratio(title) >= 90
  D  identical normalised title (>= 3 tokens, >= 15 chars), same category, brand missing or different   [sensitivity only]
Comparability guard (kept as a flag, used by the primary H5 specification):
  comparable = both single-variant, or equal variant counts and (grams equal or both missing)

Outputs: data/interim/match_pairs.parquet, data/interim/match_groups.parquet (connected components over tiers A–C),
         experiments/matching.json
"""
from __future__ import annotations
import itertools, json, re, unicodedata
import numpy as np, pandas as pd, networkx as nx
from rapidfuzz import fuzz
from src.common import ROOT, P, get_logger, ledger, register_output, write_json
from src.clean.listings import norm_title

log = get_logger("features.match")
GENERIC_VENDORS = {"", "general", "generale", "default", "default title", "unknown", "inconnu", "divers", "vendor", "brand", "marque", "none", "n a", "na", "autre", "autres"}


def norm_vendor(v: str) -> str:
    v = norm_title(v)
    v = re.sub(r"\b(inc|ltee|ltd|llc|enr|senc|co|cie|boutique|la|le|les|the|de|du|des|d)\b", " ", v)
    return re.sub(r"\s+", " ", v).strip()


def sku_ok(s) -> bool:
    if not isinstance(s, str): return False
    s = s.strip()
    if len(s) < 4: return False
    if s.isdigit() and len(s) < 6: return False
    return not re.fullmatch(r"(default|none|null|n/a|sku|test)[-_ ]*\d*", s.lower())


def main():
    li = pd.read_parquet(P["interim"] / "listings_clean.parquet")
    u = li[li.f_analysis_price].copy()
    u["vendor_norm"] = u.vendor.fillna("").map(norm_vendor)
    u.loc[u.vendor_norm.isin(GENERIC_VENDORS), "vendor_norm"] = ""
    u["sku_norm"] = u.sku.where(u.sku.map(sku_ok)).fillna("").str.strip().str.upper()
    u["n_tok"] = u.title_norm.str.split().str.len()
    u["nv"] = u.n_variants.fillna(1).astype(int)
    idx = u.set_index("uid")
    pairs = []

    def add(a, b, tier, score):
        if a == b: return
        pairs.append((min(a, b), max(a, b), tier, score))

    # Tier A — shared SKU across stores
    sk = u[u.sku_norm != ""].groupby("sku_norm")
    for s, g in sk:
        if g.store_id.nunique() < 2 or len(g) > 60: continue
        for (ua, ta, sa), (ub, tb, sb) in itertools.combinations(zip(g.uid, g.title_norm, g.store_id), 2):
            if sa == sb: continue
            sc = fuzz.token_set_ratio(ta, tb)
            if sc >= 70: add(ua, ub, "A", sc)
    nA = len(pairs); log.info("tier A pairs %s", nA)

    # Tier B/C — same brand
    vb = u[u.vendor_norm != ""].groupby("vendor_norm")
    for v, g in vb:
        if g.store_id.nunique() < 2 or len(g) > 4000: continue
        # B: identical title
        for t, gg in g.groupby("title_norm"):
            if gg.store_id.nunique() < 2 or len(t) < 8: continue
            for (ua, sa), (ub, sb) in itertools.combinations(zip(gg.uid, gg.store_id), 2):
                if sa != sb: add(ua, ub, "B", 100)
        # C: fuzzy within category
        for c, gc in g.groupby("category"):
            if gc.store_id.nunique() < 2 or len(gc) > 1500: continue
            stores = gc.store_id.values; uids = gc.uid.values; titles = gc.title_norm.values
            for i in range(len(gc)):
                for j in range(i + 1, len(gc)):
                    if stores[i] == stores[j] or titles[i] == titles[j]: continue
                    sc = fuzz.token_sort_ratio(titles[i], titles[j])
                    if sc >= 90: add(uids[i], uids[j], "C", sc)
    nBC = len(pairs) - nA; log.info("tier B+C pairs %s", nBC)

    # Tier D — identical title, brand missing/different
    gt = u[(u.n_tok >= 3) & (u.title_norm.str.len() >= 15)].groupby(["category", "title_norm"])
    for (c, t), g in gt:
        if g.store_id.nunique() < 2 or len(g) > 40: continue
        for (ua, sa, va), (ub, sb, vb_) in itertools.combinations(zip(g.uid, g.store_id, g.vendor_norm), 2):
            if sa == sb: continue
            if va and vb_ and va == vb_: continue   # already tier B
            add(ua, ub, "D", 100)
    nD = len(pairs) - nA - nBC; log.info("tier D pairs %s", nD)

    pr = pd.DataFrame(pairs, columns=["uid_a", "uid_b", "tier", "score"])
    pr = pr.sort_values(["uid_a", "uid_b", "tier"]).drop_duplicates(["uid_a", "uid_b"], keep="first")  # best tier (A<B<C<D)
    for side in ("a", "b"):
        pr[f"store_{side}"] = pr[f"uid_{side}"].map(idx.store_id); pr[f"price_{side}"] = pr[f"uid_{side}"].map(idx.price)
        pr[f"nv_{side}"] = pr[f"uid_{side}"].map(idx.nv); pr[f"grams_{side}"] = pr[f"uid_{side}"].map(idx.grams)
        pr[f"cat_{side}"] = pr[f"uid_{side}"].map(idx.category)
    pr["comparable"] = ((pr.nv_a == 1) & (pr.nv_b == 1)) | ((pr.nv_a == pr.nv_b) & ((pr.grams_a == pr.grams_b) | (pr.grams_a.isna() & pr.grams_b.isna())))
    pr["log_gap"] = (np.log(pr.price_a) - np.log(pr.price_b)).abs()
    pr.to_parquet(P["interim"] / "match_pairs.parquet", index=False); register_output(P["interim"] / "match_pairs.parquet", "match")

    # groups over A-C
    G = nx.Graph(); G.add_edges_from(pr.loc[pr.tier.isin(list("ABC")), ["uid_a", "uid_b"]].itertuples(index=False, name=None))
    rows = []
    for gid, comp in enumerate(nx.connected_components(G)):
        for uid in comp: rows.append((gid, uid))
    gr = pd.DataFrame(rows, columns=["group_id", "uid"])
    gr["store_id"] = gr.uid.map(idx.store_id); gr["price"] = gr.uid.map(idx.price); gr["category"] = gr.uid.map(idx.category)
    gr["nv"] = gr.uid.map(idx.nv); gr["grams"] = gr.uid.map(idx.grams); gr["vendor_norm"] = gr.uid.map(idx.vendor_norm)
    gs = gr.groupby("group_id").agg(n=("uid", "size"), n_stores=("store_id", "nunique"), n_cat=("category", "nunique"))
    gr = gr.merge(gs, left_on="group_id", right_index=True)
    gr.to_parquet(P["interim"] / "match_groups.parquet", index=False); register_output(P["interim"] / "match_groups.parquet", "match")

    summ = {"universe_listings": len(u), "universe_stores": int(u.store_id.nunique()), "with_sku": int((u.sku_norm != "").sum()), "with_brand": int((u.vendor_norm != "").sum()),
            "pairs_by_tier": pr.tier.value_counts().to_dict(), "pairs_comparable_by_tier": pr[pr.comparable].tier.value_counts().to_dict(),
            "median_abs_log_gap_by_tier": pr.groupby("tier").log_gap.median().round(4).to_dict(),
            "share_zero_gap_by_tier": pr.groupby("tier").log_gap.apply(lambda s: (s < 1e-9).mean()).round(4).to_dict(),
            "groups_AC": int(gs.shape[0]), "groups_AC_multi_store": int((gs.n_stores >= 2).sum()), "listings_in_groups": len(gr),
            "stores_in_groups": int(gr.store_id.nunique()), "groups_size_gt_10": int((gs.n > 10).sum()),
            "regions_pairs": None}
    write_json(summ, ROOT / "experiments" / "matching.json", "match"); ledger("features.match_products", **summ)
    log.info(json.dumps(summ, ensure_ascii=False))


if __name__ == "__main__":
    main()