"""Phase 4 features — region × category count matrices under the three weighting schemes and the sample variants.

Outputs data/processed/rc_counts.parquet (long format): sample ∈ {S_R, S_R0, S_A}, scheme ∈ {L, M, T}, region, category, n
plus store-level table data/processed/store_panel.parquet used by every model (one row per store in S_A with region, weights, size).
"""
from __future__ import annotations
import json
import numpy as np, pandas as pd
from src.common import ROOT, P, REGIONS, get_logger, ledger, register_output, write_json, PROC_DIR, EXP_DIR, LABEL, BASE_PROC, BASE_EXP

log = get_logger("features.rc")
PROC = PROC_DIR


def store_category_table():
    li = pd.read_parquet(P["interim"] / "listings_clean.parquet")
    sm = pd.read_parquet(P["interim"] / "stores_master.parquet")
    a = li[li.f_analysis_assort]
    if LABEL == "model":
        cm = pd.read_parquet(BASE_PROC / "category_model.parquet").set_index("uid").category_model
        a = a.assign(category=a.uid.map(cm).fillna(a.category))
    sc = a.groupby(["store_id", "category"]).size().rename("n").reset_index()
    sc = sc.merge(sm[["id", "region_final", "region_source", "n_listings", "platform", "store_kind", "certified", "first_source", "origin_class"]], left_on="store_id", right_on="id").drop(columns="id")
    sc = sc.merge(sm[["id", "region_pipfirst", "region_db"]], left_on="store_id", right_on="id").drop(columns="id")
    return sc, sm


def build(sc: pd.DataFrame) -> pd.DataFrame:
    rows = []
    sizes = sc.groupby("store_id").n.sum()
    cut = sizes.quantile(0.99)
    for sample, mask in (("S_R", sc.region_final != "Inconnue"), ("S_R0", sc.region_db.notna()), ("S_Rpip", sc.region_pipfirst != "Inconnue"), ("S_A", pd.Series(True, index=sc.index))):
        d = sc[mask].copy()
        if sample == "S_R0": d["region_final"] = d.region_db       # platform label only (preregistered S_R0)
        if sample == "S_Rpip": d["region_final"] = d.region_pipfirst
        d["w_L"] = d.n.astype(float)
        d["w_M"] = d.n / d.store_id.map(sizes)
        d["w_T"] = np.where(d.store_id.map(sizes) > cut, 0.0, d.n.astype(float))
        for scheme in ("L", "M", "T"):
            g = d.groupby(["region_final", "category"])[f"w_{scheme}"].sum().reset_index()
            g.columns = ["region", "category", "n"]; g["sample"] = sample; g["scheme"] = scheme
            rows.append(g)
    out = pd.concat(rows, ignore_index=True)
    return out


def main():
    sc, sm = store_category_table()
    rc = build(sc)
    rc.to_parquet(PROC / "rc_counts.parquet", index=False); register_output(PROC / "rc_counts.parquet", "features.rc")
    sc.to_parquet(PROC / "store_category.parquet", index=False); register_output(PROC / "store_category.parquet", "features.rc")
    sp = sm[["id", "region_final", "region_source", "region_conflict", "n_listings", "n_categories", "platform", "store_kind", "certified", "cert_adq", "cert_lpq",
             "first_source", "origin_class", "median_price", "main_category", "lat", "lng", "postal_prefix", "mrc_pip"]].copy()
    sp = sp[sp.n_listings > 0]
    sp.to_parquet(PROC / "store_panel.parquet", index=False); register_output(PROC / "store_panel.parquet", "features.rc")
    summ = {"stores_S_A": int(sp.shape[0]), "stores_S_R": int((sp.region_final != "Inconnue").sum()), "stores_S_R0": int(sm.region_db.notna().sum()),
            "listings_S_R": int(rc.query("sample=='S_R' and scheme=='L'").n.sum()), "listings_S_A": int(rc.query("sample=='S_A' and scheme=='L'").n.sum()),
            "top1pct_cut": float(sc.groupby('store_id').n.sum().quantile(0.99)),
            "stores_by_region_S_R": sp[sp.region_final != "Inconnue"].region_final.value_counts().to_dict()}
    write_json(summ, EXP_DIR / "features_rc.json", "features.rc"); ledger("features.region_category", **summ); log.info(json.dumps(summ, ensure_ascii=False))


if __name__ == "__main__":
    main()