"""Phase 2.2 — Listing-level cleaning and the sample-flow (attrition) table.

Input : data/interim/listings_raw.parquet, data/interim/stores_raw.parquet
Output: data/interim/listings_clean.parquet  (all published-active listings of live stores, with flags; nothing deleted)
        reports/sample_flow.md, experiments/sample_flow.json

Flags (all kept, so every analysis can state its own exclusion set):
  f_live_store      store enabled with >= 1 published listing at the snapshot
  f_published       active=1 and listing_status='published' (platform's own publication rule)
  f_cad             store currency is CAD for >= 95 % of its listings and this listing is CAD
  f_price_ok        price > 0, not price_on_request, price <= 50 000 CAD (p99.9 guard, see report)
  f_dup_in_store    normalised title duplicated inside the same store (variants listed as separate products); the
                    first-seen listing is kept as the representative (f_dup_rep)
  f_hors_mission    store_kind == 'hors_mission'
  f_analysis_assort assortment sample = live & published & not f_dup_in_store-non-rep & not hors_mission
  f_analysis_price  price sample = f_analysis_assort & f_cad & f_price_ok
"""
from __future__ import annotations
import json, re, unicodedata
import numpy as np, pandas as pd
from src.common import ROOT, P, get_logger, ledger, register_output, write_json

log = get_logger("clean.listings")
PRICE_CAP = 50_000.0


def norm_title(t: str) -> str:
    t = unicodedata.normalize("NFKD", str(t or "")).encode("ascii", "ignore").decode().lower()
    t = re.sub(r"&amp;|&#\d+;", " ", t)
    t = re.sub(r"[^a-z0-9]+", " ", t)
    return re.sub(r"\s+", " ", t).strip()


def main():
    pr = pd.read_parquet(P["interim"] / "listings_raw.parquet")
    st = pd.read_parquet(P["interim"] / "stores_raw.parquet")
    live_ids = set(st.loc[(st.enabled == 1) & (st.product_count > 0), "id"])
    kind = st.set_index("id")["store_kind"].fillna("")
    flow = [("DB product rows (all statuses, all stores)", len(pr))]

    pr["f_published"] = (pr.active == 1) & (pr.listing_status == "published")
    flow.append(("active = 1 and listing_status = 'published'", int(pr.f_published.sum())))
    pr["f_live_store"] = pr.store_id.isin(live_ids)
    base = pr[pr.f_published & pr.f_live_store].copy()
    flow.append(("… of an enabled store with >= 1 published listing", len(base)))

    base["store_kind"] = base.store_id.map(kind).fillna("")
    base["f_hors_mission"] = base.store_kind == "hors_mission"
    flow.append(("… excluding store_kind = hors_mission", int((~base.f_hors_mission).sum())))

    base["title_norm"] = base.title.map(norm_title)
    base = base.sort_values(["store_id", "title_norm", "first_seen", "uid"])
    grp = base.groupby(["store_id", "title_norm"], sort=False)
    base["dup_group_size"] = grp["uid"].transform("size")
    base["f_dup_in_store"] = base.dup_group_size > 1
    base["f_dup_rep"] = ~base.duplicated(["store_id", "title_norm"], keep="first")
    flow.append(("… keeping one representative per (store, normalised title) and non-empty title", int(((~base.f_hors_mission) & base.f_dup_rep & (base.title_norm.str.len() > 0)).sum())))
    sibp = P["interim"] / "sibling_stores.parquet"
    nonrep = set(pd.read_parquet(sibp).query("not is_rep").id) if sibp.exists() else set()
    base["f_sibling_dup"] = base.store_id.isin(nonrep)
    base["f_analysis_assort"] = (~base.f_hors_mission) & base.f_dup_rep & (base.title_norm.str.len() > 0) & (~base.f_sibling_dup)
    flow.append(("… excluding mirror domains of the same business (sibling stores, non-representative)  [ASSORTMENT SAMPLE]", int(base.f_analysis_assort.sum())))

    cur_share = base.groupby("store_id")["currency"].apply(lambda s: (s == "CAD").mean())
    base["store_cad_share"] = base.store_id.map(cur_share)
    base["f_cad"] = (base.currency == "CAD") & (base.store_cad_share >= 0.95)
    base["f_price_ok"] = base.price.notna() & (base.price > 0) & (base.price_on_request == 0) & (base.price <= PRICE_CAP)
    base["f_analysis_price"] = base.f_analysis_assort & base.f_cad & base.f_price_ok
    flow.append(("… and CAD-priced store/listing", int((base.f_analysis_assort & base.f_cad).sum())))
    flow.append(("… and a positive listed price <= 50 000 CAD, not on request  [PRICE SAMPLE]", int(base.f_analysis_price.sum())))

    base["log_price"] = np.where(base.f_price_ok, np.log(base.price.where(base.price > 0)), np.nan)
    base["on_sale"] = base.compare_at_price.notna() & (base.compare_at_price > base.price)
    base["has_variants"] = (base.price_max.notna() & (base.price_max > base.price)) | (base.n_variants.fillna(1) > 1)
    base["available"] = base.available.astype("Int64")
    base.to_parquet(P["interim"] / "listings_clean.parquet", index=False)
    register_output(P["interim"] / "listings_clean.parquet", "clean.listings")

    # ---- diagnostics for the data-quality report ----------------------------------
    a = base[base.f_analysis_assort]
    diag = {
        "flow": flow,
        "assortment_sample": {"listings": len(a), "stores": int(a.store_id.nunique()), "categories": int(a.category.nunique()),
                               "dup_in_store_listings_removed": int((base.f_dup_in_store & ~base.f_dup_rep & ~base.f_hors_mission).sum()),
                               "share_autre": round(float((a.category == "autre").mean()), 4),
                               "available_0_share": round(float((a.available == 0).mean()), 4),
                               "on_sale_share": round(float(a.on_sale.mean()), 4), "has_variants_share": round(float(a.has_variants.mean()), 4),
                               "sku_share": round(float(a.has_sku.mean()), 4), "vendor_share": round(float((a.vendor.fillna("") != "").mean()), 4)},
        "price_exclusions": {"non_cad_listing_or_store": int((base.f_analysis_assort & ~base.f_cad).sum()),
                              "price_missing_or_on_request": int((base.f_analysis_assort & base.f_cad & (base.price.isna() | (base.price_on_request == 1))).sum()),
                              "price_zero": int((base.f_analysis_assort & base.f_cad & (base.price == 0)).sum()),
                              "price_gt_cap": int((base.f_analysis_assort & base.f_cad & (base.price > PRICE_CAP)).sum())},
        "price_quantiles_cad": {str(q): round(float(base.loc[base.f_analysis_price, "price"].quantile(q)), 2) for q in (0.001, 0.01, 0.1, 0.5, 0.9, 0.99, 0.999)},
        "top10_stores_share_of_assortment": round(float(a.store_id.value_counts().head(10).sum() / len(a)), 4),
    }
    write_json(diag, ROOT / "experiments" / "sample_flow.json", "clean.listings")
    md = ["# Sample flow (attrition) — WP38", "", f"Snapshot 2026-11-01T17:38:30Z. Nothing is deleted; every step is a flag in `data/interim/listings_clean.parquet`.", "",
          "| step | listings |", "|---|---:|"] + [f"| {k} | {v:,} |".replace(",", " ") for k, v in flow] + ["",
          "## Assortment sample diagnostics", "```", json.dumps(diag["assortment_sample"], indent=1), "```",
          "## Price-sample exclusions (from the assortment sample)", "```", json.dumps(diag["price_exclusions"], indent=1), "```",
          "## Price quantiles in the price sample (CAD)", "```", json.dumps(diag["price_quantiles_cad"], indent=1), "```",
          f"Top-10 stores hold {diag['top10_stores_share_of_assortment']:.1%} of the assortment sample."]
    (ROOT / "reports" / "sample_flow.md").write_text("\n".join(md))
    ledger("clean.listings", **{k: v for k, v in diag.items() if k != "flow"}, flow=flow)
    log.info("\n".join(f"{k}: {v}" for k, v in flow))


if __name__ == "__main__":
    main()