"""run_pipeline: the worker's single hook. Same contract as engine_stub.
Also: backtest() — accuracy + interval coverage against known truth."""
from __future__ import annotations

import json
from pathlib import Path

import numpy as np
import pandas as pd

from .core import attach_intervals, estimate_row, index_evidence
from .epg_adapter import adapt_epg, is_epg_export
from .reference import ReferenceData, validate_schedule
from .vr_report import broadcast_hours, platform_totals


def _load_reference(cfg_path: Path | None) -> ReferenceData:
    """fixtures mode via worker/config.json {'reference_mode':'fixtures',
    'fixtures_dir': ...}; mysql mode on the server."""
    if cfg_path and cfg_path.exists():
        cfg = json.loads(cfg_path.read_text())
        mode = cfg.get("reference_mode", "fixtures")
        if mode == "mysql":
            return ReferenceData.from_mysql(cfg)
        return ReferenceData.from_fixtures(cfg.get("fixtures_dir", "fixtures"))
    raise FileNotFoundError("No worker config found for reference data.")


def estimate_frame(schedule: pd.DataFrame, ref: ReferenceData,
                   on_progress=None, sigma_by_tier: dict | None = None,
                   correction_by_tier: dict | None = None) -> pd.DataFrame:
    idx = index_evidence(ref)
    n = len(schedule)
    results = []
    for i, (_, row) in enumerate(schedule.iterrows(), start=1):
        results.append(estimate_row(row.to_dict(), ref, idx, correction_by_tier))
        if on_progress and (i % 25 == 0 or i == n):
            pct = 30 + int(45 * i / max(1, n))
            on_progress(pct, "Estimating audiences", f"row {i:,} of {n:,}")
    out = pd.concat([schedule.reset_index(drop=True), pd.DataFrame(results)], axis=1)
    return attach_intervals(out, sigma_by_tier)


def run_pipeline(*, upload_path: Path, exports_dir: Path, run_id: str, on_progress,
                 config_path: Path | None = None) -> dict:
    on_progress(3, "Checking your file", "Reading columns and dates")
    schedule = pd.read_excel(upload_path)

    # A raw EPG export has none of the columns the estimator needs, so it is
    # translated first. Detected by its own signature columns rather than by
    # filename, so either file type can be uploaded to the same box.
    if is_epg_export(schedule):
        on_progress(5, "Reading EPG listings", "Translating to a schedule")
        cfg_for_epg = config_path or (
            Path(__file__).resolve().parent.parent / "worker" / "config.json")
        schedule = adapt_epg(schedule, cfg_for_epg)
        on_progress(8, "Reading EPG listings",
                    f"{len(schedule):,} broadcasts after de-duplication")

    problems = validate_schedule(schedule)
    if problems:
        raise ValueError(" ".join(problems))

    # Which tournaments are actually in this file. Written on every run so the
    # enrichment screen can offer a pre-filled Wikipedia page per event without
    # the analyst having to tell us what they just uploaded. Cheap, and it never
    # blocks the estimate.
    _write_events(schedule, exports_dir, run_id)

    # Fold in any fixtures already confirmed for these events. Adds sports_teams
    # and match_level, which is what makes tier 2 and the stage weight reachable.
    schedule, merged = _merge_fixtures(schedule, config_path)
    if merged:
        on_progress(9, "Applying fixtures", f"{merged:,} rows matched to a known fixture")
    if "weekday" not in schedule.columns and "broadcast_date" in schedule.columns:
        schedule["weekday"] = pd.to_datetime(schedule["broadcast_date"]).dt.weekday
    on_progress(10, "Matching evidence", "Loading reference data")
    cfg = config_path or (Path(__file__).resolve().parent.parent / "worker" / "config.json")
    ref = _load_reference(cfg)
    on_progress(28, "Matching evidence", f"{len(ref.evidence()):,} historical broadcasts")

    # Use the measured calibration where one exists. Without this the engine
    # loads the fitted parameters and then ignores them: intervals fall back to
    # the assumed sigma constants and the bias correction stays at 1.0.
    sigma = ref.sigma_by_tier()
    corr = ref.correction_by_tier()
    if sigma or corr:
        on_progress(29, "Matching evidence",
                    f"calibration: {getattr(ref, 'calibration_source', 'assumed')}")
    out = estimate_frame(schedule, ref, on_progress,
                         sigma_by_tier=sigma, correction_by_tier=corr)

    on_progress(88, "Running safety checks", "Comparing against market sizes")
    flagged = out[out["flag"].astype(str).str.startswith(("REVIEW", "NO_EVIDENCE"))]
    on_progress(94, "Preparing your reports", "Formatting workbooks")

    summary_cols = ["event_name", "country_name", "channel_name", "telecast_type",
                    "estimated_ama_000", "interval80_low", "interval80_high",
                    "confidence", "strategy"]
    summary = out[[c for c in summary_cols if c in out.columns]]
    summary_path = exports_dir / f"{run_id}_summary.xlsx"
    detailed_path = exports_dir / f"{run_id}_detailed.xlsx"

    # The summary workbook is what an analyst actually opens and forwards, so it
    # is formatted and carries a Method sheet: the fitted exponents, the measured
    # intervals and — explicitly — what the model does NOT apply. A number is
    # only as good as the reader's ability to check it.
    try:
        from .excel_report import write_report
        write_report(out, summary_path, run_id,
                     title=str(upload_path.name), ref=ref)
    except Exception:                       # noqa: BLE001 — never lose a run over formatting
        summary.to_excel(summary_path, index=False)

    # The detailed workbook stays a plain dump: it is the machine-readable copy.
    out.to_excel(detailed_path, index=False)

    # chart/table data for the results dashboard (public/results.php).
    # allow_nan=False on purpose: json.dumps otherwise emits bare NaN, which is
    # not valid JSON, and the browser's JSON.parse rejects the whole file — the
    # analytics page then renders empty with no visible error. Better to fail
    # here, loudly, than to ship a file the dashboard cannot read.
    chart_payload = _json_safe(_build_chart_payload(out, ref))
    (exports_dir / f"{run_id}_chartdata.json").write_text(
        json.dumps(chart_payload, allow_nan=False))

    flags = [{"row": int(i) + 2,
              "channel": str(r.get("channel_name", "")),
              "country": str(r.get("country_name", "")),
              "programme": str(r.get("programme_name", r.get("event_name", ""))),
              "ama": (None if pd.isna(r["estimated_ama_000"]) else float(r["estimated_ama_000"])),
              "why": {"REVIEW_ABOVE_15PCT_TVU": "Estimate is above 15% of this market's TV universe",
                      "REVIEW_LOW_CONFIDENCE": "Evidence for this row is thin — confidence is low",
                      "NO_EVIDENCE": "No usable historical evidence was found for this row",
                      "MISSING_TVU_REFUSED": "Market TV population missing — Crystal refused to guess",
                      }.get(str(r["flag"]), str(r["flag"]))}
             for i, r in flagged.iterrows()]
    (exports_dir / f"{run_id}_flags.json").write_text(json.dumps(flags))

    return {"rows_total": len(out),
            "rows_estimated": int(out["estimated_ama_000"].notna().sum()),
            "rows_flagged": len(flags),
            "summary_path": summary_path, "detailed_path": detailed_path}


# --------------------------------------------------------------- backtest
def backtest(estimates: pd.DataFrame, truth: pd.Series) -> dict:
    ok = estimates["estimated_ama_000"].notna()
    p = estimates.loc[ok, "estimated_ama_000"].to_numpy(float)
    a = truth[ok].to_numpy(float)
    err = p - a
    wape = float(np.abs(err).sum() / a.sum())
    bias = float(err.sum() / a.sum())
    hit20 = float((np.abs(err) / a <= 0.20).mean())
    cov = float(((estimates.loc[ok, "interval80_low"] <= a)
                 & (a <= estimates.loc[ok, "interval80_high"])).mean())
    per_tier = (pd.DataFrame({"tier": estimates.loc[ok, "strategy_tier"].to_numpy(),
                              "ape": np.abs(err) / a})
                .groupby("tier")["ape"].agg(["mean", "count"]).round(3).to_dict("index"))
    return {"rows": int(ok.sum()), "coverage_rate": round(ok.mean(), 3),
            "WAPE": round(wape, 4), "bias_pct": round(bias, 4),
            "hit_within_20pct": round(hit20, 3), "interval80_coverage": round(cov, 3),
            "per_tier_mean_APE": per_tier}


def fit_calibration(estimates: pd.DataFrame, truth: pd.Series,
                    shrink_n: float = 8.0) -> tuple[dict, dict]:
    """Per-tier residual calibration in log-space (Crystal's dormant module, live):
    correction = exp(w*mu_tier + (1-w)*mu_global), w = n/(n+shrink_n);
    sigma feeds the 80% intervals. Returns (correction_by_tier, sigma_by_tier)."""
    ok = estimates["estimated_ama_000"].notna() & (truth > 0)
    lr = np.log(truth[ok].to_numpy(float) / estimates.loc[ok, "estimated_ama_000"].to_numpy(float))
    df = pd.DataFrame({"tier": estimates.loc[ok, "strategy_tier"].to_numpy(), "lr": lr})
    mu_g = float(df["lr"].mean())
    corr, sig = {}, {}
    for t, g in df.groupby("tier"):
        n = len(g)
        if n < 8:
            continue
        w = n / (n + shrink_n)
        corr[int(t)] = float(np.exp(w * g["lr"].mean() + (1 - w) * mu_g))
        sig[int(t)] = float(np.clip(g["lr"].std(ddof=1), 0.05, 0.8))
    return corr, sig


def _write_events(schedule: pd.DataFrame, exports_dir: Path, run_id: str) -> None:
    """List the distinct tournaments in this schedule, with enough context for
    the enrichment screen to guess the right Wikipedia page."""
    if "event_name" not in schedule.columns:
        return
    dcol = next((c for c in ("broadcast_date", "prog_date", "date") if c in schedule.columns), None)
    rows = []
    for name, g in schedule[schedule["event_name"].notna()].groupby("event_name"):
        dates = pd.to_datetime(g[dcol], errors="coerce").dropna() if dcol else pd.Series(dtype="datetime64[ns]")
        titles = (g["prog_title"].dropna().unique().tolist()[:4]
                  if "prog_title" in g.columns else [])
        rows.append({
            "event_name": str(name),
            "rows": int(len(g)),
            "countries": sorted({str(c) for c in g.get("country_name", pd.Series(dtype=object)).dropna().unique()})[:12],
            "date_from": dates.min().strftime("%Y-%m-%d") if len(dates) else "",
            "date_to": dates.max().strftime("%Y-%m-%d") if len(dates) else "",
            "season": int(dates.max().year) if len(dates) else None,
            "sample_titles": [str(t) for t in titles],
        })
    rows.sort(key=lambda r: -r["rows"])
    (exports_dir / f"{run_id}_events.json").write_text(json.dumps(rows, indent=1))


def _merge_fixtures(schedule: pd.DataFrame, config_path: Path | None) -> tuple[pd.DataFrame, int]:
    """Attach teams and stage from confirmed Wikipedia fixtures.

    Joined on (event_name, date). Where a date carries several fixtures every one
    of them is kept in the table; here the closest by kickoff time is chosen when
    the schedule has an hour, otherwise the first. Nothing is ever blanked out --
    an ambiguous date still yields a stage and a best-guess pairing, and
    fixture_candidates records how many it was chosen from.
    """
    if config_path is None or "event_name" not in schedule.columns:
        return schedule, 0
    try:
        cfg = json.loads(Path(config_path).read_text())
        import pymysql
        conn = pymysql.connect(
            host=cfg["db_host"], port=int(cfg.get("db_port", 3306)),
            user=cfg["db_user"], password=cfg["db_pass"],
            database=cfg.get("db_name", "crystal"), charset="utf8mb4",
            cursorclass=pymysql.cursors.DictCursor)
    except Exception:
        return schedule, 0

    events = [str(e) for e in schedule["event_name"].dropna().unique()]
    if not events:
        return schedule, 0
    with conn.cursor() as c:
        c.execute(
            "SELECT event_name, match_date, match_time, team1, team2, stage "
            "FROM wiki_fixtures WHERE match_date IS NOT NULL AND event_name IN (%s)"
            % ",".join(["%s"] * len(events)), events)
        fixtures = c.fetchall()
    conn.close()
    if not fixtures:
        return schedule, 0

    by_key: dict[tuple, list] = {}
    for f in fixtures:
        by_key.setdefault((f["event_name"], str(f["match_date"])), []).append(f)

    dcol = next((c for c in ("broadcast_date", "prog_date", "date") if c in schedule.columns), None)
    teams, levels, cands = [], [], []
    hit = 0
    for _, r in schedule.iterrows():
        got = None
        n = 0
        if dcol and pd.notna(r.get("event_name")) and pd.notna(r.get(dcol)):
            key = (str(r["event_name"]), str(pd.to_datetime(r[dcol], errors="coerce").date()))
            pool = by_key.get(key) or []
            n = len(pool)
            if n == 1:
                got = pool[0]
            elif n > 1:
                hour = r.get("hour")
                timed = [f for f in pool if f.get("match_time")]
                if timed and pd.notna(hour):
                    got = min(timed, key=lambda f: abs(int(str(f["match_time"])[:2]) - int(hour)))
                else:
                    got = pool[0]
        if got:
            hit += 1
            teams.append(f"{got['team1']}|{got['team2']}")
            levels.append(got["stage"])
        else:
            teams.append(r.get("sports_teams", "") or "")
            levels.append(r.get("match_level", "") or "")
        cands.append(n)

    schedule = schedule.copy()
    schedule["sports_teams"] = teams
    schedule["match_level"] = levels
    schedule["fixture_candidates"] = cands
    return schedule, hit


def _json_safe(obj):
    """Recursively turn NaN/NaT/numpy scalars into JSON-legal values.

    pandas keeps NaN inside float columns even after .where(notna, None), so
    to_dict('records') hands back float('nan'). Left alone those reach the
    browser as the bare token NaN and kill JSON.parse for the entire file.
    """
    if isinstance(obj, dict):
        return {k: _json_safe(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple)):
        return [_json_safe(v) for v in obj]
    if obj is None or isinstance(obj, (str, bool)):
        return obj
    if isinstance(obj, (np.integer,)):
        return int(obj)
    if isinstance(obj, (float, np.floating)):
        f = float(obj)
        return None if (np.isnan(f) or np.isinf(f)) else f
    if isinstance(obj, (int,)):
        return obj
    if obj is pd.NaT or (hasattr(pd, "isna") and not isinstance(obj, (list, dict))
                         and pd.isna(obj)):
        return None
    return str(obj)


def _build_chart_payload(out: pd.DataFrame, ref=None) -> dict:
    """Aggregates + row table for the results dashboard. Pure JSON, no pandas
    objects leak out, so plain PHP/JS can render it with zero dependencies."""
    ok = out["estimated_ama_000"].notna()
    est = out.loc[ok, "estimated_ama_000"]
    conf = out.loc[ok, "confidence"]
    tiers = out.loc[ok, "strategy_tier"].astype("Int64")
    tier_names = {1: "Same channel", 2: "Same teams", 3: "Same market",
                 4: "Cross-market", 5: "Time-slot history"}

    # AMA distribution histogram (log-scaled bins read well for audience data)
    bins = [0, 5, 20, 50, 100, 250, 500, 1000, 5000, 1e12]
    labels = ["0-5k", "5-20k", "20-50k", "50-100k", "100-250k", "250-500k",
             "500k-1M", "1M-5M", "5M+"]
    hist = pd.cut(est, bins=bins, labels=labels, right=False).value_counts().reindex(labels).fillna(0)

    # confidence bands
    band = pd.cut(conf, bins=[0, 60, 75, 90, 100],
                  labels=["Low", "Medium", "High", "Very High"], right=True)
    conf_counts = band.value_counts().reindex(["Very High", "High", "Medium", "Low"]).fillna(0)

    tier_counts = tiers.map(tier_names).value_counts()
    tier_order = [tier_names[t] for t in sorted(tier_names) if tier_names[t] in tier_counts.index]

    top_markets = (out.loc[ok].groupby("country_name")["estimated_ama_000"]
                  .sum().sort_values(ascending=False).head(8))

    flag_counts = out["flag"].replace("", "None").value_counts()
    # The helper returns exact hours because the Summary sheet renders them as
    # hh:mm:ss. The tiles want a whole number, so round here.
    _hours, _hours_note = broadcast_hours(out)
    _hours = int(round(_hours))
    # OTT and out-of-home come from the same helper the workbook's platform
    # sheets use, so the tiles and the book cannot disagree. Without `ref`
    # these are zero -- which is what those sheets show too.
    _plat = platform_totals(out, ref)

    table_cols = ["event_name", "country_name", "channel_name", "telecast_type",
                 "estimated_ama_000", "interval80_low", "interval80_high",
                 "confidence", "strategy", "flag"]
    table = out[[c for c in table_cols if c in out.columns]].copy()
    table = table.where(pd.notna(table), None)

    return {
        "kpi": {
            "rows_total": int(len(out)),
            "rows_estimated": int(ok.sum()),
            "median_ama": round(float(est.median()), 1) if len(est) else None,
            "mean_confidence": round(float(conf.mean()), 1) if len(conf) else None,
            # fillna first: a flag column round-tripped through Excel comes back
            # with NaN for clean rows, and str(NaN) == "nan" counts as flagged.
            "flagged": int((out["flag"].fillna("").astype(str).str.strip() != "").sum()),
            # Whole hours -- the same figure and the same provenance note the
            # workbook's Summary sheet carries, so the two never disagree.
            # Summed AMA, computed once here so the run page and the analytics
            # page cannot show two different totals for the same run.
            "total_ama_000": round(float(est.sum()), 1) if len(est) else 0.0,
            "broadcast_hours": _hours,
            "broadcast_hours_note": _hours_note,
            "ott_000": round(_plat["ott_000"], 1),
            "ooh_000": round(_plat["ooh_000"], 1),
            "ott_markets": _plat["ott_markets"],
            "ooh_markets": _plat["ooh_markets"],
            "markets": int(out["country_name"].nunique()) if "country_name" in out else 0,
            "channels": int(out["channel_name"].nunique()) if "channel_name" in out else 0,
        },
        "distribution": {"labels": labels, "values": [int(v) for v in hist.tolist()]},
        "confidence_bands": {"labels": list(conf_counts.index.astype(str)),
                             "values": [int(v) for v in conf_counts.tolist()]},
        "tiers": {"labels": tier_order,
                 "values": [int(tier_counts.get(t, 0)) for t in tier_order]},
        "top_markets": {"labels": [str(x) for x in top_markets.index.tolist()],
                        "values": [round(float(v), 1) for v in top_markets.tolist()]},
        "flags": {"labels": [str(x) for x in flag_counts.index.tolist()],
                 "values": [int(v) for v in flag_counts.tolist()]},
        "rows": table.to_dict(orient="records"),
    }