pipeline_philly/verify/reconcile.py

"""Base case (ii) reconciliation: served DB vs exact ingested source file.

For each component in components.yaml, reads the served value from
philly_school_year_metrics, re-reads the same cell from the ingested
source file, and compares within tolerance.

Mirrors verify/reconcile.py (NYC). Writes a Markdown + JSON report per metric
to docs/qa_reports/philly/<metric_key>__validation.md (+ .json).

Run:
  python pipeline_philly/verify/reconcile.py --metric pssa_all_ela_proficiency
  python pipeline_philly/verify/reconcile.py --all
"""
from __future__ import annotations

import argparse
import json
import pathlib
import sys

# Make pipeline_philly importable when this script is run as a path.
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2]))

import yaml  # noqa: E402

from pipeline_philly.verify import common  # noqa: E402
from pipeline_philly.verify.connectors import served_db  # noqa: E402
from pipeline_philly.verify.connectors import source_file  # noqa: E402


def load_components() -> list[dict]:
    with open(common.COMPONENTS_YAML) as f:
        return yaml.safe_load(f)


def read_source(component: dict) -> dict[tuple[str, str], dict]:
    """Dispatch a component to the right source-file reader."""
    sid = component["served_source"]
    args = component.get("source_args", {})
    subgroup = component["subgroup"]
    year = component.get("year")

    if sid == "sdp_pssa_keystone_2024-25":
        return source_file.read_sdp_pssa_zip_schools(
            cut=args["cut"],
            year=year,
            testname=args["testname"],
            subject=args["subject"],
            grade=args["grade"],
            subgroup=subgroup,
        )
    elif sid == "sdp_attendance_90_school":
        return source_file.read_sdp_csv_long(
            source_id=sid,
            metric_value_col=args["metric_value_col"],
            subgroup=subgroup,
        )
    elif sid == "sdp_graduation_school_2014-2025":
        return source_file.read_sdp_csv_long(
            source_id=sid,
            metric_value_col=args["metric_value_col"],
            school_col=args["school_col"],
            year_col=args["year_col"],
            group_col=args["group_col"],
            subgroup=subgroup,
            extra_filter=args.get("extra_filter"),
        )
    else:
        raise SystemExit(f"no source-reader dispatch for source: {sid}")


def reconcile_one(component: dict) -> dict:
    """Run base-case reconciliation for one metric. Returns summary dict."""
    metric = component["metric_key"]
    subgroup = component["subgroup"]
    cut = component["population_cut"]
    tol = float(component["tolerance_abs"])
    year_filter = component.get("year")

    served = served_db.fetch(metric, subgroup=subgroup, population_cut=cut)
    source = read_source(component)

    # If a single year was specified in the config, restrict served + source to it
    if year_filter:
        served = {k: v for k, v in served.items() if k[1] == year_filter}
        source = {k: v for k, v in source.items() if k[1] == year_filter}

    served_keys = set(served.keys())
    source_keys = set(source.keys())
    common_keys = served_keys & source_keys

    # Cell-level fidelity within common keys
    matches = 0
    mismatches = []
    suppressed = 0
    for k in common_keys:
        s_val = served[k]["value"]
        f_val = source[k]["value"]
        if served[k]["suppressed"]:
            suppressed += 1
            continue
        if s_val is None and f_val is None:
            matches += 1
            continue
        if s_val is None or f_val is None:
            mismatches.append((k, s_val, f_val, None))
            continue
        diff = s_val - f_val
        if abs(diff) <= tol:
            matches += 1
        else:
            mismatches.append((k, s_val, f_val, diff))

    return {
        "metric_key": metric,
        "subgroup": subgroup,
        "population_cut": cut,
        "tolerance_abs": tol,
        "n_served": len(served),
        "n_source": len(source),
        "n_common": len(common_keys),
        "matches": matches,
        "mismatches": mismatches,
        "suppressed": suppressed,
        "served_only_count": len(served_keys - source_keys),
        "source_only_count": len(source_keys - served_keys),
        "served_only_examples": sorted([f"{k[0]} {k[1]}" for k in (served_keys - source_keys)])[:20],
        "source_only_examples": sorted([f"{k[0]} {k[1]}" for k in (source_keys - served_keys)])[:20],
        "agreement_rate": matches / (matches + len(mismatches)) if (matches + len(mismatches)) else None,
        "status": "PASS" if matches and len(mismatches) == 0 else "MISMATCH",
    }


def render_report(summary: dict) -> str:
    out: list[str] = []
    out.append(f"# Validation (ii) base case — {summary['metric_key']}\n")
    out.append("served DB cell-by-cell vs an independent re-read of the exact ingested file.\n")
    out.append(f"## Summary\n")
    for k in [
        "metric_key", "subgroup", "population_cut", "tolerance_abs",
        "n_served", "n_source", "n_common", "matches", "suppressed",
        "served_only_count", "source_only_count", "agreement_rate", "status",
    ]:
        out.append(f"- {k}: {summary.get(k)}")
    if summary.get("served_only_examples"):
        out.append("\n## Served-only examples\n")
        for e in summary["served_only_examples"]:
            out.append(f"- {e}")
    if summary.get("source_only_examples"):
        out.append("\n## Source-only examples\n")
        for e in summary["source_only_examples"]:
            out.append(f"- {e}")
    if summary["mismatches"]:
        out.append(f"\n## Mismatches ({len(summary['mismatches'])})\n")
        for (key, s, f, d) in summary["mismatches"][:50]:
            out.append(f"- {key[0]} {key[1]}: served={s} source={f} Δ={d}")
        if len(summary["mismatches"]) > 50:
            out.append(f"- ... and {len(summary['mismatches']) - 50} more")
    return "\n".join(out)


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--metric", default=None)
    ap.add_argument("--all", action="store_true")
    args = ap.parse_args()

    common.ensure_qa_dir()
    components = load_components()
    if args.metric:
        components = [c for c in components if c["metric_key"] == args.metric]
        if not components:
            print(f"no component matching --metric={args.metric}", file=sys.stderr)
            sys.exit(1)
    if not args.metric and not args.all:
        print("must pass --metric=<key> or --all")
        sys.exit(1)

    for component in components:
        summary = reconcile_one(component)
        # Persist the report
        slug = component["metric_key"]
        md = render_report(summary)
        (common.QA_REPORTS_DIR / f"{slug}__validation.md").write_text(md)
        json_summary = {k: v for k, v in summary.items() if k != "mismatches"}
        json_summary["n_mismatches"] = len(summary["mismatches"])
        (common.QA_REPORTS_DIR / f"{slug}__validation.json").write_text(
            json.dumps(json_summary, indent=2),
        )
        print(
            f"{slug:38s} STATUS: {summary['status']:9s}  "
            f"common={summary['n_common']:>5d}  matches={summary['matches']:>5d}  "
            f"mismatches={len(summary['mismatches']):>4d}  "
            f"served_only={summary['served_only_count']:>3d}  source_only={summary['source_only_count']:>3d}",
        )


if __name__ == "__main__":
    main()