"""Phase 1.5 — coverage and attrition tables (inputs of Table 1 / Figure 1).

Outputs (data/processed/coverage_*.csv): by feed (listings, active, days observed, events, unit-price coverage,
flags), by category, by week (listings observed, feeds reporting), and the attrition ledger.
"""
import pandas as pd, numpy as np
from src.common import path, get_logger, ledger, SETTINGS

log = get_logger("coverage")

def main():
    P = path("processed")
    prod = pd.read_parquet(P / "products.parquet"); ret = pd.read_parquet(P / "retailers.parquet")
    wk = pd.read_parquet(P / "offers_weekly.parquet"); day = pd.read_parquet(P / "offers_daily.parquet", columns=["uid", "day", "source"])
    prod = prod.merge(ret[["feed_id", "banner", "company", "channel", "store_format", "is_grocery"]], left_on="source", right_on="feed_id", how="left")
    ev = pd.read_parquet(path("interim") / "raw_price_log.parquet").groupby("uid").size().rename("n_events")
    prod = prod.merge(ev, left_on="uid", right_index=True, how="left")
    dpu = day.groupby("uid").size().rename("days_obs"); prod = prod.merge(dpu, left_on="uid", right_index=True, how="left")
    by_feed = prod.groupby(["company", "banner", "source", "channel", "store_format"]).agg(
        listings=("uid", "size"), active=("active", "sum"), on_sale_now=("on_sale", "sum"),
        median_days_obs=("days_obs", "median"), price_events=("n_events", "sum"),
        unit_price_cov=("unit_price_rp", lambda s: s.notna().mean()), brand_cov=("brand", lambda s: (s.fillna("") != "").mean()),
        gtin=("gtin_valid", "sum"), weighted=("flag_weighted", "sum"), multibuy=("flag_multibuy", "sum"),
        loyalty=("flag_loyalty", "sum"), unrealistic=("flag_unrealistic", "sum")).reset_index().sort_values("listings", ascending=False)
    by_feed.to_csv(P / "coverage_by_feed.csv", index=False)
    by_cat = prod.groupby("category").agg(listings=("uid", "size"), active=("active", "sum"), feeds=("source", "nunique"),
        unit_price_cov=("unit_price_rp", lambda s: s.notna().mean()), price_events=("n_events", "sum")).reset_index().sort_values("listings", ascending=False)
    by_cat.to_csv(P / "coverage_by_category.csv", index=False)
    wk2 = wk.merge(prod[["uid", "channel", "category", "is_grocery"]], on="uid")
    by_week = wk2.groupby("week").agg(listings=("uid", "nunique"), feeds=("source", "nunique"),
        changes=("n_changes", "sum"), online=("channel", lambda s: (s == "online_storefront").sum()),
        circular=("channel", lambda s: s.str.startswith("circular").sum())).reset_index()
    by_week.to_csv(P / "coverage_by_week.csv", index=False)
    # attrition ledger (listing level)
    steps = []
    n0 = len(prod); steps.append(("All listings in snapshot (products table)", n0))
    n = prod[~prod.source.isin(SETTINGS["scope"]["exclude_sources"])]; steps.append(("excl. SAQ (alcohol monopoly)", len(n)))
    n = n[n.is_grocery]; steps.append(("excl. pharmacy feeds", len(n)))
    n = n[~n.category.isin(SETTINGS["scope"]["non_grocery_categories"])]; steps.append(("excl. non-food categories (health & beauty, household, pets, baby, alcohol)", len(n)))
    n = n[n.price.notna() | (n.n_events.fillna(0) > 0)]; steps.append(("with at least one observed price", len(n)))
    n = n[~n.flag_unrealistic]; steps.append(("excl. unrealistic prices", len(n)))
    n1 = n[n.size_qty.notna()]; steps.append(("… with parsable package size (unit-price / matching universe)", len(n1)))
    g = pd.read_parquet(P / "match_groups_ABC.parquet"); steps.append(("… in a cross-banner match group (tiers A–C)", int(n1.uid.isin(g.uid).sum())))
    att = pd.DataFrame(steps, columns=["step", "listings"]); att["share_of_initial"] = att.listings / n0
    att.to_csv(P / "attrition_listings.csv", index=False)
    log.info("\n%s", att.to_string(index=False)); log.info("weeks: %d (%s → %s)", len(by_week), by_week.week.min(), by_week.week.max())
    ledger("1", "coverage", f"attrition {n0} -> {att.listings.iloc[-1]} matched listings")

if __name__ == "__main__":
    main()