"""QA of the Philly absenteeism data: served attendance cells against the
publishers' files, plus a specified set of packet comparisons (not every
displayed value — see the not_checked section the report writes).

Five sections, one report (docs/qa_reports/philly/__absenteeism_full_qa.{md,json}):

  A. Base case, district file → served DB: attendance_rate_above_90 for EVERY
     year and EVERY subgroup (value, and the student count compared
     independently of whether the rate is suppressed), and
     average_daily_attendance for every year.
  B. Base case, state files → served DB: attendance_persistence_rate for every
     workbook (wide-format 2021-22..2024-25, long-format 2018-19, 2019-20, and
     2020-21) and every subgroup. YEAR ALIGNMENT: the state's attendance element
     is a lagging indicator (PDE glossary: "data is from the year prior to the
     reporting year"), so each workbook's values are expected under served year
     = workbook year − 1 (scripts/loaders/philly/_lib.ts STATE_METRIC_YEAR_OFFSET).
     Also counts blank vs numeric PercentChronicAbsenteeism cells per workbook.
  C. Completeness: every (school, year, subgroup) cell in each source file is
     served, for schools in the directory; rows for schools outside the
     directory (closed) are counted and disclosed, not hidden.
  D. Packet recomputation: a specified set of fields of data/analysis/
     philly-absenteeism-descriptives.json recomputed here, in Python, from the
     source files and the DB, and compared one by one to the committed JSON
     (the report's not_checked section lists what is outside that set).
  E. Verdict: PASS only if A, B, and D have zero mismatches and C has zero
     source-only cells among directory schools.

Read-only against the served DB. Run:
  python pipeline_philly/verify/absenteeism_full_qa.py
"""
from __future__ import annotations

import json
import math
import pathlib
import re
import sys
from collections import defaultdict

import numpy as np
import pandas as pd

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2]))

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

REPO = common.REPO_ROOT
PACKET = REPO / "data" / "analysis" / "philly-absenteeism-descriptives.json"
YEARS = ["2020-21", "2021-22", "2022-23", "2023-24", "2024-25"]
TOL_RATE = 0.0051   # packet values are stored at 4dp
TOL_VALUE = 0.005   # served values vs source (source is 2dp)


def prev_year(y: str) -> str:
    """'2024-25' → '2023-24' (the attendance year a state workbook describes)."""
    start = int(y[:4]) - 1
    return f"{start}-{str((start + 1) % 100).zfill(2)}"


def canon_year(sy: str) -> str:
    m = re.match(r"^(\d{4})-(\d{4})$", str(sy).strip())
    return f"{m.group(1)}-{m.group(2)[2:]}" if m else str(sy).strip()


def num(v) -> float | None:
    if v is None or (isinstance(v, float) and math.isnan(v)):
        return None
    s = str(v).strip().replace("%", "")
    if s.upper() in {"", "S", "*", "IS", "--", "INS", "NA", "N/A", "N.A."}:
        return None
    try:
        return float(s)
    except ValueError:
        return None


def db_query(sql: str, params=()) -> list[tuple]:
    import psycopg
    with psycopg.connect(common.served_db_dsn()) as conn:
        conn.read_only = True
        with conn.cursor() as cur:
            cur.execute(sql, params)
            return cur.fetchall()


def compare_cells(name: str, source: dict, served: dict, directory: set[str] | None,
                  tol: float, with_n: bool) -> dict:
    """source/served: {(ulcs, year, subgroup): {"value","n","suppressed"}}."""
    src_keys, srv_keys = set(source), set(served)
    src_only_all = src_keys - srv_keys
    src_only_dir = {k for k in src_only_all if directory is None or k[0] in directory}
    common_keys = src_keys & srv_keys
    mism, nmism, supp_mismatch, examples = 0, 0, 0, []
    n_compared = 0
    for k in sorted(common_keys):
        a, b = source[k], served[k]
        # Student counts are compared INDEPENDENTLY of the attendance rate:
        # the district publishes a count even when it suppresses the rate, so
        # a suppressed cell still has a denominator to check, and a count
        # missing on one side only is a mismatch, not a skip. (An earlier
        # version compared counts only after the rate branch, which silently
        # skipped every suppressed cell's count — flagged by the fourth
        # external audit, 2026-09-08.)
        if with_n:
            an, bn = a.get("n"), b.get("n")
            if an is not None or bn is not None:
                n_compared += 1
                if (an is None) != (bn is None):
                    nmism += 1
                    if len(examples) < 10:
                        examples.append({"key": list(k), "source_n": an, "served_n": bn, "kind": "denominator-presence"})
                elif abs(float(an) - float(bn)) > 0.5:
                    nmism += 1
                    if len(examples) < 10:
                        examples.append({"key": list(k), "source_n": an, "served_n": bn, "kind": "denominator"})
        if a["value"] is None or b["value"] is None:
            if (a["value"] is None) != (b["value"] is None):
                supp_mismatch += 1
                if len(examples) < 10:
                    examples.append({"key": list(k), "source": a["value"], "served": b["value"], "kind": "suppression"})
            continue
        if abs(a["value"] - b["value"]) > tol:
            mism += 1
            if len(examples) < 10:
                examples.append({"key": list(k), "source": a["value"], "served": b["value"], "kind": "value"})
    served_only_keys = sorted(srv_keys - src_keys)
    return {
        "name": name,
        "source_only_examples": [list(k) for k in sorted(src_only_all)[:12]],
        "served_only_examples": [list(k) for k in served_only_keys[:12]],
        "source_cells": len(src_keys),
        "served_cells": len(srv_keys),
        "common_cells": len(common_keys),
        "value_mismatches": mism,
        "denominator_comparisons": n_compared,
        "denominator_mismatches": nmism,
        "suppression_mismatches": supp_mismatch,
        "served_only": len(srv_keys - src_keys),
        "source_only_directory_schools": len(src_only_dir),
        "source_only_non_directory_schools": len(src_only_all - src_only_dir),
        "status": "PASS" if (mism == 0 and nmism == 0 and supp_mismatch == 0 and len(src_only_dir) == 0 and len(srv_keys - src_keys) == 0) else "FAIL",
        "examples": examples,
    }


# ---------------------------------------------------------------------------
def main() -> None:
    common.ensure_qa_dir()
    packet = json.loads(PACKET.read_text())
    report: dict = {"generated_from_packet": packet["meta"]["generated"], "sections": {}}

    directory = {str(r[0]) for r in db_query("SELECT ulcs_code FROM philly_schools")}
    school_meta = {str(r[0]): {"type": r[1], "level": r[2]} for r in db_query(
        "SELECT ulcs_code, school_type::text, school_level FROM philly_schools")}

    # ---------------- A. district file base case ----------------
    sdp = pd.read_csv(common.latest_source_path("sdp_attendance_90_school"), low_memory=False)
    sdp = sdp[sdp["Sector"].astype(str) == "District"]
    src90: dict = {}
    grade_rows: list[dict] = []
    for _, r in sdp.iterrows():
        ulcs = str(r["ULCS Code"]).strip()
        y = canon_year(r["School Year"])
        cat, grp = str(r["Category"]).strip(), str(r["Group"]).strip()
        val = num(r["% with 90%+ Attendance (Yearly)"])
        denom = num(r["Total Students (Yearly)"])
        n90 = num(r["# with 90%+ Attendance (Yearly)"])
        if cat == "Grade Level":
            grade_rows.append({"ulcs": ulcs, "year": y, "grade": grp, "denom": denom, "n90": n90, "rate": val})
            continue
        sg = SDP_GROUP_MAP.get(grp)
        if sg is None:
            continue
        src90[(ulcs, y, sg)] = {"value": val, "n": denom, "n90": n90, "suppressed": val is None}
    served90: dict = {}
    for sg in sorted({k[2] for k in src90}):
        for (ulcs, y), v in served_db.fetch("attendance_rate_above_90", subgroup=sg).items():
            served90[(ulcs, y, sg)] = v
    secA = {"attendance_rate_above_90": compare_cells("attendance_rate_above_90 (all years × all subgroups)", src90, served90, directory, TOL_VALUE, True)}

    ada = pd.read_csv(common.latest_source_path("sdp_attendance_ada_school"), low_memory=False)
    ada = ada[ada["Sector"].astype(str) == "District"]
    srcAda = {(str(r["ULCS Code"]).strip(), canon_year(r["School Year"]), "ALL"): {"value": num(r["Average Daily Attendance (YTD)"]), "n": None, "suppressed": num(r["Average Daily Attendance (YTD)"]) is None} for _, r in ada.iterrows()}
    servedAda = {(u, y, "ALL"): v for (u, y), v in served_db.fetch("average_daily_attendance").items()}
    secA["average_daily_attendance"] = compare_cells("average_daily_attendance (all years)", srcAda, servedAda, directory, TOL_VALUE, False)
    report["sections"]["A_base_case_district_file"] = secA

    raw90 = pd.read_csv(common.latest_source_path("sdp_attendance_90_school"), low_memory=False, dtype=str)
    raw90 = raw90[raw90["Sector"].astype(str) == "District"]
    tok: dict[str, dict[str, int]] = {}
    for col in ["% with 90%+ Attendance (Yearly)", "# with 90%+ Attendance (Yearly)", "Total Students (Yearly)"]:
        v = raw90[col].fillna("<blank>").astype(str).str.strip()
        nonnum = v[pd.to_numeric(v, errors="coerce").isna()]
        tok[col] = {k: int(c) for k, c in nonnum.value_counts().items()}
    report["district_file_nonnumeric_tokens"] = {"note": "Every non-numeric token in the district yearly file's value columns (District sector). 's' is the district's suppression marker; anything else here would be an unrecognized code.", "tokens": tok}

    # ---------------- B. state files base case ----------------
    master = pd.read_csv(common.latest_source_path("sdp_master_school_list_2024-25"), low_memory=False)
    # Same rule as scripts/loaders/philly/_lib.ts buildPdeToUlcs: a state code
    # shared by a school and its Continuation Academy belongs to the school.
    cand: dict[str, list[tuple[str, str]]] = {}
    for _, r in master.iterrows():
        try:
            k = f"{int(float(r['AUN Code']))}-{int(float(r['PA Code']))}"
            cand.setdefault(k, []).append((str(int(float(r["ULCS Code"]))), str(r.get("Publication Name", r.get("School Name", "")))))
        except (ValueError, TypeError):
            continue
    aunpa_to_ulcs: dict[str, str] = {}
    shared_codes = []
    alias_codes = []
    for k, lst in cand.items():
        distinct = {u for u, _ in lst}
        pick = next((u for u, n in lst if "continuation" not in n.lower()), lst[0][0]) if len(distinct) > 1 else lst[0][0]
        aunpa_to_ulcs[k] = pick
        if len(distinct) > 1:
            # A genuine collision: two different school records share one state code.
            shared_codes.append({"state_code": k, "schools": lst, "assigned": pick})
        elif len(lst) > 1:
            # Alias: the same ULCS listed more than once under name variants (not a collision).
            alias_codes.append({"state_code": k, "ulcs": pick, "names": [n for _, n in lst]})
    report["shared_state_codes"] = shared_codes
    report["alias_state_codes"] = alias_codes
    srcState: dict = {}
    chronic_col_census: dict[str, dict] = {}
    for workbook in ["2021-22", "2022-23", "2023-24", "2024-25"]:
        year = prev_year(workbook)  # attendance year (lagging indicator)
        path = common.latest_source_path(f"futurereadypa_performance_{workbook}")
        df = pd.read_excel(path, sheet_name="School On Track Measures")
        cols = {c: FRPI_GROUP_MAP[c.split("_", 1)[1]] for c in df.columns
                if c.startswith("PercentPersistentAttendance_") and c.split("_", 1)[1] in FRPI_GROUP_MAP}
        chronic_cols = [c for c in df.columns if c.startswith("PercentChronicAbsenteeism_")]
        census = {"workbook": workbook, "attendance_year": year, "philly_rows": 0, "chronic_columns_present": len(chronic_cols),
                  "schema_status": "columns present" if chronic_cols else "columns absent from the School On Track Measures sheet",
                  "chronic_cells_examined": 0, "chronic_cells_blank": 0, "chronic_cells_numeric": 0, "chronic_cells_other": 0}
        token_census: dict[str, int] = {}
        for _, r in df.iterrows():
            if pd.isna(r["AUN"]) or pd.isna(r["Schl"]):
                continue
            ulcs = aunpa_to_ulcs.get(f"{int(r['AUN'])}-{int(r['Schl'])}")
            if not ulcs:
                continue
            census["philly_rows"] += 1
            for c in cols:
                raw = r[c]
                if num(raw) is None:
                    tok = "<blank>" if (raw is None or (isinstance(raw, float) and math.isnan(raw)) or str(raw).strip() == "") else str(raw).strip()
                    token_census[tok] = token_census.get(tok, 0) + 1
            for c in chronic_cols:
                raw = r[c]
                census["chronic_cells_examined"] += 1
                if raw is None or (isinstance(raw, float) and math.isnan(raw)) or str(raw).strip() == "":
                    census["chronic_cells_blank"] += 1
                elif num(raw) is not None:
                    census["chronic_cells_numeric"] += 1
                else:
                    census["chronic_cells_other"] += 1
            for c, sg in cols.items():
                v = num(r[c])
                srcState[(ulcs, year, sg)] = {"value": v, "n": None, "suppressed": v is None}
        assert census["chronic_cells_examined"] == census["philly_rows"] * len(chronic_cols)
        assert census["chronic_cells_examined"] == census["chronic_cells_blank"] + census["chronic_cells_numeric"] + census["chronic_cells_other"]
        census["nonnumeric_attendance_tokens"] = token_census
        census["attendance_cells_examined"] = census["philly_rows"] * len(cols)
        chronic_col_census[workbook] = census

    LONG_SG = {"All Student": "ALL", "American Indian/Alaska Native": "AMER_INDIAN_AK_NATIVE", "Asian": "ASIAN",
               "Hawaiian/Pacific Islander": "HAWAIIAN_PAC_ISL", "Black": "BLACK", "Hispanic": "HISPANIC", "White": "WHITE",
               "2 or More Races": "TWO_OR_MORE_RACES", "Economically Disadvantaged": "ECON_DISADV",
               "English Learner": "ELL", "Students with Disabilities": "IEP"}
    for workbook in ["2018-19", "2019-20", "2020-21"]:
        year = prev_year(workbook)  # attendance year (lagging indicator)
        long_tokens: dict[str, int] = {}
        long_cells = 0
        path = common.latest_source_path(f"futurereadypa_performance_{workbook}")
        xf = pd.ExcelFile(path, engine="openpyxl")
        for sheet in [s for s in xf.sheet_names if s.lower().startswith("schools")]:
            df = pd.read_excel(xf, sheet_name=sheet, usecols=["AUN", "Schl", "DataElement", "DisplayValue"])
            df = df[df["DataElement"].astype(str).str.replace(r"\s+", " ", regex=True).str.startswith("Percent Regular Attendance")]
            for _, r in df.iterrows():
                el = re.sub(r"\s+", " ", str(r["DataElement"])).strip()
                m = re.search(r"\(([^)]+)\)\s*$", el)
                sg = LONG_SG.get(m.group(1).strip()) if m else None
                if not sg or pd.isna(r["AUN"]) or pd.isna(r["Schl"]):
                    continue
                ulcs = aunpa_to_ulcs.get(f"{int(r['AUN'])}-{int(r['Schl'])}")
                if not ulcs:
                    continue
                v = num(r["DisplayValue"])
                long_cells += 1
                if v is None:
                    raw = r["DisplayValue"]
                    tok = "<blank>" if (raw is None or (isinstance(raw, float) and math.isnan(raw)) or str(raw).strip() == "") else str(raw).strip()
                    long_tokens[tok] = long_tokens.get(tok, 0) + 1
                srcState[(ulcs, year, sg)] = {"value": v, "n": None, "suppressed": v is None}
        chronic_col_census[workbook] = {"workbook": workbook, "attendance_year": year, "philly_rows": None, "chronic_columns_present": 0,
                                        "schema_status": "long format (one row per school × element); no chronic-absenteeism element",
                                        "chronic_cells_examined": 0, "chronic_cells_blank": 0, "chronic_cells_numeric": 0, "chronic_cells_other": 0,
                                        "attendance_cells_examined": long_cells, "nonnumeric_attendance_tokens": long_tokens}
    report["state_chronic_absenteeism_column_census"] = dict(sorted(chronic_col_census.items()))
    KNOWN_STATE_TOKENS = {"IS", "--", "INS", "*", "<blank>", "INSUFFICIENT SAMPLE"}
    report["state_nonnumeric_tokens"] = {
        "note": "Every non-numeric token in the matched-school attendance cells of all seven state workbooks. Known suppression codes: IS / Insufficient Sample (insufficient sample), --, INS, *, blank. Anything else is flagged as unexpected.",
        "by_workbook": {wb: c["nonnumeric_attendance_tokens"] for wb, c in chronic_col_census.items()},
        "unexpected": sorted({t for c in chronic_col_census.values() for t in c["nonnumeric_attendance_tokens"] if t.upper() not in {k.upper() for k in KNOWN_STATE_TOKENS}}),
    }
    servedState: dict = {}
    for sg in sorted({k[2] for k in srcState}):
        for (u, y), v in served_db.fetch("attendance_persistence_rate", subgroup=sg).items():
            servedState[(u, y, sg)] = v
    report["sections"]["B_base_case_state_files"] = {
        "attendance_persistence_rate": compare_cells("attendance_persistence_rate (attendance years 2017-18..2023-24 from workbooks 2018-19..2024-25 × all subgroups)", srcState, servedState, None, TOL_VALUE, False),
    }

    # ---------------- C. completeness (by year) ----------------
    compl = []
    for metric, src, srv in [("attendance_rate_above_90", src90, served90), ("average_daily_attendance", srcAda, servedAda), ("attendance_persistence_rate", srcState, servedState)]:
        for y in sorted({k[1] for k in src}):
            sk = {k for k in src if k[1] == y}
            vk = {k for k in srv if k[1] == y}
            so = sk - vk
            compl.append({"metric": metric, "year": y, "source_cells": len(sk), "served_cells": len(vk),
                          "source_only_directory": len({k for k in so if k[0] in directory}),
                          "source_only_non_directory": len({k for k in so if k[0] not in directory}),
                          "served_only": len(vk - sk)})
    report["sections"]["C_completeness"] = compl

    # ---------------- D. packet recomputation ----------------
    mismatches: list[dict] = []
    compared = 0

    def check(path: str, got, exp, tol: float):
        nonlocal compared
        compared += 1
        if got is None or exp is None:
            if got is not exp:
                mismatches.append({"path": path, "packet": exp, "recomputed": got})
            return
        if isinstance(got, str) or isinstance(exp, str):
            if str(got) != str(exp):
                mismatches.append({"path": path, "packet": exp, "recomputed": got})
            return
        if abs(float(got) - float(exp)) > tol:
            mismatches.append({"path": path, "packet": exp, "recomputed": round(float(got), 4)})

    all_cells = {y: [] for y in YEARS}
    for (ulcs, y, sg), v in src90.items():
        if sg == "ALL" and v["value"] is not None and v["n"] is not None and v["n90"] is not None and y in all_cells:
            all_cells[y].append({"ulcs": ulcs, "v": 100 - v["value"], "denom": v["n"], "n90": v["n90"]})

    def weighted(cells):
        tn = sum(c["denom"] for c in cells); t9 = sum(c["n90"] for c in cells)
        return 100 * (1 - t9 / tn)

    ct = packet["citywide_trend"]
    for y in YEARS:
        cells = all_cells[y]; vs = np.array(sorted(c["v"] for c in cells))
        check(f"citywide_trend.{y}.weighted", weighted(cells), ct[y]["weighted"], TOL_RATE)
        check(f"citywide_trend.{y}.unweighted_mean", vs.mean(), ct[y]["unweighted_mean"], TOL_RATE)
        for q, key in [(10, "p10"), (25, "p25"), (50, "p50"), (75, "p75"), (90, "p90")]:
            check(f"citywide_trend.{y}.{key}", np.percentile(vs, q), ct[y][key], TOL_RATE)
        check(f"citywide_trend.{y}.sd", vs.std(), ct[y]["sd"], TOL_RATE)
        check(f"citywide_trend.{y}.n_schools", len(cells), ct[y]["n_schools"], 0)
        check(f"citywide_trend.{y}.n_students", sum(c["denom"] for c in cells), ct[y]["n_students"], 0)

    pg = packet["per_grade"]
    for g in ["00", "01", "02", "03", "04", "05", "06", "07", "08", "09", "10", "11", "12"]:
        label = "K" if g == "00" else str(int(g))
        for y in YEARS:
            rows = [r for r in grade_rows if r["grade"] == g and r["year"] == y and r["denom"] is not None and r["n90"] is not None]
            tn = sum(r["denom"] for r in rows); t9 = sum(r["n90"] for r in rows)
            check(f"per_grade.{label}.{y}.weighted", 100 * (1 - t9 / tn) if tn else None, pg[label][y]["weighted"], TOL_RATE)
            check(f"per_grade.{label}.{y}.n_schools", len(rows), pg[label][y]["n_schools"], 0)

    # school_level_trend (served cells joined to directory levels)
    slt = packet["school_level_trend"]
    for lvl in ["Elementary", "Elementary-Middle", "Middle", "Middle-High", "High"]:
        for y in YEARS:
            cells = [(100 - v["value"], v["n"]) for (u, yy, sg), v in served90.items()
                     if sg == "ALL" and yy == y and not v["suppressed"] and v["value"] is not None and v["n"] is not None
                     and school_meta.get(u, {}).get("level") == lvl]
            tn = sum(n for _, n in cells)
            check(f"school_level_trend.{lvl}.{y}.weighted", sum(v * n for v, n in cells) / tn if tn else None, slt[lvl][y]["weighted"], TOL_RATE)
            check(f"school_level_trend.{lvl}.{y}.n_schools", len(cells), slt[lvl][y]["n_schools"], 0)

    # ca_vs_ada
    cva = packet["ca_vs_ada"]
    for y in YEARS:
        pairs = []
        for (u, yy, sg), v in served90.items():
            if sg != "ALL" or yy != y or v["suppressed"] or v["value"] is None or v["n"] is None:
                continue
            a = servedAda.get((u, y, "ALL"))
            if not a or a["value"] is None or a["suppressed"]:
                continue
            pairs.append((100 - v["value"], a["value"], v["n"]))
        ca = np.array([p[0] for p in pairs]); ad = np.array([p[1] for p in pairs]); w = np.array([p[2] for p in pairs])
        check(f"ca_vs_ada.{y}.weighted_ca", (ca * w).sum() / w.sum(), cva[y]["weighted_ca"], TOL_RATE)
        check(f"ca_vs_ada.{y}.weighted_ada", (ad * w).sum() / w.sum(), cva[y]["weighted_ada"], TOL_RATE)
        check(f"ca_vs_ada.{y}.school_level_r", np.corrcoef(ca, ad)[0, 1], cva[y]["school_level_r"], 0.0006)
        rk = np.corrcoef(pd.Series(ca).rank().values, pd.Series(ad).rank().values)[0, 1]
        check(f"ca_vs_ada.{y}.spearman_r", rk, cva[y]["spearman_r"], 0.0006)
        check(f"ca_vs_ada.{y}.n_schools", len(pairs), cva[y]["n_schools"], 0)

    # subgroup_trend (race, all reporting schools) + gender common + black-white common
    RACE = {"BLACK", "HISPANIC", "WHITE", "ASIAN", "TWO_OR_MORE_RACES"}
    st = packet["subgroup_trend"]
    for sg in RACE:
        for y in YEARS:
            cells = [v for (u, yy, s), v in src90.items() if s == sg and yy == y and v["value"] is not None and v["n"] is not None and v["n90"] is not None]
            tn = sum(c["n"] for c in cells); t9 = sum(c["n90"] for c in cells)
            check(f"subgroup_trend.{sg}.{y}.weighted", 100 * (1 - t9 / tn) if tn else None, st[sg][y]["weighted"], TOL_RATE)
            check(f"subgroup_trend.{sg}.{y}.n_schools", len(cells), st[sg][y]["n_schools"], 0)
    gt = packet["gender_trend_common"]
    for y in YEARS:
        m = {u: v for (u, yy, s), v in src90.items() if s == "MALE" and yy == y and v["value"] is not None and v["n"] is not None}
        f = {u: v for (u, yy, s), v in src90.items() if s == "FEMALE" and yy == y and v["value"] is not None and v["n"] is not None}
        both = set(m) & set(f)
        for key, d in [("MALE", m), ("FEMALE", f)]:
            cells = [d[u] for u in both]
            tn = sum(c["n"] for c in cells); t9 = sum(c["n90"] for c in cells)
            check(f"gender_trend_common.{key}.{y}.weighted", 100 * (1 - t9 / tn) if tn else None, gt[key][y]["weighted"], TOL_RATE)
            check(f"gender_trend_common.{key}.{y}.n_schools", len(cells), gt[key][y]["n_schools"], 0)
    gp = packet["gender_trend_fixed_panel"]
    panel_years = gp["years"]
    panel = None
    for y in panel_years:
        m = {u for (u, yy, s), v in src90.items() if s == "MALE" and yy == y and v["value"] is not None and v["n"] is not None}
        f = {u for (u, yy, s), v in src90.items() if s == "FEMALE" and yy == y and v["value"] is not None and v["n"] is not None}
        both = m & f
        panel = both if panel is None else (panel & both)
    check("gender_trend_fixed_panel.n_schools", len(panel), gp["n_schools"], 0)
    for y in panel_years:
        for key in ["MALE", "FEMALE"]:
            cells = [v for (u, yy, s), v in src90.items() if s == key and yy == y and u in panel]
            tn = sum(c["n"] for c in cells); t9 = sum(c["n90"] for c in cells)
            check(f"gender_trend_fixed_panel.{key}.{y}.weighted", 100 * (1 - t9 / tn) if tn else None, gp[key][y]["weighted"], TOL_RATE)
            check(f"gender_trend_fixed_panel.{key}.{y}.n_schools", len(cells), gp[key][y]["n_schools"], 0)
    bw = packet["black_white_common"]
    for y in YEARS:
        b = {u: v for (u, yy, s), v in src90.items() if s == "BLACK" and yy == y and v["value"] is not None and v["n"] is not None}
        w_ = {u: v for (u, yy, s), v in src90.items() if s == "WHITE" and yy == y and v["value"] is not None and v["n"] is not None}
        both = set(b) & set(w_)
        bb = [b[u] for u in both]; ww = [w_[u] for u in both]
        bl = 100 * (1 - sum(c["n90"] for c in bb) / sum(c["n"] for c in bb)); wh = 100 * (1 - sum(c["n90"] for c in ww) / sum(c["n"] for c in ww))
        check(f"black_white_common.{y}.black", bl, bw[y]["black"], TOL_RATE)
        check(f"black_white_common.{y}.white", wh, bw[y]["white"], TOL_RATE)
        check(f"black_white_common.{y}.gap", bl - wh, bw[y]["gap"], 2 * TOL_RATE)
        check(f"black_white_common.{y}.n_schools", len(both), bw[y]["n_schools"], 0)
        diffs = [(100 * (1 - b[u]["n90"] / b[u]["n"])) - (100 * (1 - w_[u]["n90"] / w_[u]["n"])) for u in both]
        wts = [b[u]["n"] + w_[u]["n"] for u in both]
        check(f"black_white_common.{y}.gap_within_school_equal_weight", float(np.mean(diffs)), bw[y]["gap_within_school_equal_weight"], TOL_RATE)
        check(f"black_white_common.{y}.gap_within_school_bw_weight", float(np.average(diffs, weights=wts)), bw[y]["gap_within_school_bw_weight"], TOL_RATE)
        bl_rates = [100 * (1 - b[u]["n90"] / b[u]["n"]) for u in both]; wh_rates = [100 * (1 - w_[u]["n90"] / w_[u]["n"]) for u in both]
        check(f"black_white_common.{y}.black_mean_school", float(np.mean(bl_rates)), bw[y]["black_mean_school"], TOL_RATE)
        check(f"black_white_common.{y}.white_mean_school", float(np.mean(wh_rates)), bw[y]["white_mean_school"], TOL_RATE)
        check(f"black_white_common.{y}.black_bw_weighted", float(np.average(bl_rates, weights=wts)), bw[y]["black_bw_weighted"], TOL_RATE)
        check(f"black_white_common.{y}.white_bw_weighted", float(np.average(wh_rates, weights=wts)), bw[y]["white_bw_weighted"], TOL_RATE)
    bwp = packet["black_white_fixed_panel"]
    bw_panel = None
    for y in bwp["years"]:
        b = {u for (u, yy, s_), v in src90.items() if s_ == "BLACK" and yy == y and v["value"] is not None and v["n"] is not None}
        w_ = {u for (u, yy, s_), v in src90.items() if s_ == "WHITE" and yy == y and v["value"] is not None and v["n"] is not None}
        bw_panel = (b & w_) if bw_panel is None else (bw_panel & b & w_)
    check("black_white_fixed_panel.n_schools", len(bw_panel), bwp["n_schools"], 0)
    for y in bwp["years"]:
        b = {u: v for (u, yy, s_), v in src90.items() if s_ == "BLACK" and yy == y and u in bw_panel}
        w_ = {u: v for (u, yy, s_), v in src90.items() if s_ == "WHITE" and yy == y and u in bw_panel}
        us = sorted(bw_panel)
        diffs = [(100 * (1 - b[u]["n90"] / b[u]["n"])) - (100 * (1 - w_[u]["n90"] / w_[u]["n"])) for u in us]
        wts = [b[u]["n"] + w_[u]["n"] for u in us]
        check(f"black_white_fixed_panel.{y}.gap_within_school_equal_weight", float(np.mean(diffs)), bwp[y]["gap_within_school_equal_weight"], TOL_RATE)
        check(f"black_white_fixed_panel.{y}.gap_within_school_bw_weight", float(np.average(diffs, weights=wts)), bwp[y]["gap_within_school_bw_weight"], TOL_RATE)
        check(f"black_white_fixed_panel.{y}.n_schools", len(us), bwp[y]["n_schools"], 0)
    ur = packet["universe_reconciliation"]
    for y in YEARS:
        full = all_cells[y]
        dir_only = [c for c in full if c["ulcs"] in directory]
        wf = 100 * (1 - sum(c["n90"] for c in full) / sum(c["denom"] for c in full))
        wd = 100 * (1 - sum(c["n90"] for c in dir_only) / sum(c["denom"] for c in dir_only))
        check(f"universe_reconciliation.{y}.weighted_full_universe", wf, ur[y]["weighted_full_universe"], TOL_RATE)
        check(f"universe_reconciliation.{y}.weighted_directory_only", wd, ur[y]["weighted_directory_only"], TOL_RATE)
        check(f"universe_reconciliation.{y}.effect_of_including_source_only", wf - wd, ur[y]["effect_of_including_source_only"], 2 * TOL_RATE)

    # distribution
    dist = packet["distribution"]
    for y in ["2021-22", "2024-25"]:
        vs = [c["v"] for c in all_cells[y]]
        counts = [0] * 20
        for v in vs:
            counts[min(int(v // 5), 19)] += 1
        for i, c in enumerate(counts):
            check(f"distribution.histograms.{y}.counts[{i}]", c, dist["histograms"][y]["counts"][i], 0)
    for y in YEARS:
        vs = [c["v"] for c in all_cells[y]]
        check(f"distribution.n_above_50_by_year.{y}.n_above_50", sum(1 for v in vs if v > 50), dist["n_above_50_by_year"][y]["n_above_50"], 0)

    # stability
    stab = packet["stability"]
    by = {y: {c["ulcs"]: c for c in all_cells[y]} for y in YEARS}
    for i in range(len(YEARS) - 1):
        y1, y2 = YEARS[i], YEARS[i + 1]
        pairs = [(by[y1][u]["v"], by[y2][u]["v"]) for u in by[y1] if u in by[y2]]
        a = np.array(pairs)
        check(f"stability.consecutive_r.{y1}->{y2}.pearson_r", np.corrcoef(a[:, 0], a[:, 1])[0, 1], stab["consecutive_r"][f"{y1}->{y2}"]["pearson_r"], 0.0006)
        check(f"stability.consecutive_r.{y1}->{y2}.n_schools", len(pairs), stab["consecutive_r"][f"{y1}->{y2}"]["n_schools"], 0)
    lp = stab["latest_pair_change"]
    ch = [(by["2024-25"][u]["v"] - by["2023-24"][u]["v"], by["2024-25"][u]["denom"]) for u in by["2023-24"] if u in by["2024-25"]]
    d = np.array([c for c, _ in ch])
    check("stability.latest_pair_change.n_schools", len(ch), lp["n_schools"], 0)
    check("stability.latest_pair_change.median_change", np.median(d), lp["median_change"], TOL_RATE)
    check("stability.latest_pair_change.median_abs_change", np.median(np.abs(d)), lp["median_abs_change"], TOL_RATE)
    check("stability.latest_pair_change.pct_improved_gt5pp", 100 * (d < -5).sum() / len(d), lp["pct_improved_gt5pp"], TOL_RATE)
    check("stability.latest_pair_change.pct_worsened_gt5pp", 100 * (d > 5).sum() / len(d), lp["pct_worsened_gt5pp"], TOL_RATE)
    for label, lo, hi in [("<300", 0, 300), ("300-599", 300, 600), ("600+", 600, 10**9)]:
        dd = np.array([c for c, n in ch if lo <= n < hi])
        check(f"stability.volatility_by_size_latest_pair.{label}.sd_change", dd.std(), stab["volatility_by_size_latest_pair"][label]["sd_change"], TOL_RATE)
        check(f"stability.volatility_by_size_latest_pair.{label}.n", len(dd), stab["volatility_by_size_latest_pair"][label]["n"], 0)
        check(f"stability.volatility_by_size_latest_pair.{label}.median_abs_change", float(np.median(np.abs(dd))) if len(dd) else None, stab["volatility_by_size_latest_pair"][label]["median_abs_change"], TOL_RATE)
    check("stability.latest_pair_change.n_moved_gt5", int(sum(1 for c, _ in ch if abs(c) > 5)), stab["latest_pair_change"]["n_moved_gt5"], 0)
    check("stability.latest_pair_change.n_moved_gt10", int(sum(1 for c, _ in ch if abs(c) > 10)), stab["latest_pair_change"]["n_moved_gt10"], 0)

    # district_monthly
    mon = pd.read_csv(common.latest_source_path("sdp_attendance_90_district_monthly"), low_memory=False)
    mon = mon[(mon["Sector"].astype(str) == "District") & (mon["Category"].astype(str) == "All Students")]
    dm = packet["district_monthly"]
    for _, r in mon.iterrows():
        y = canon_year(r["School Year"]); m_ = str(r["Month"]).strip()
        v = num(r["% with 90%+ Attendance (This Month)"])
        n_src = num(r["Total Students (This Month)"])
        if y in dm and m_ in dm[y] and v is not None:
            check(f"district_monthly.{y}.{m_}.pct_below_90", 100 - v, dm[y][m_]["pct_below_90"], TOL_RATE)
            check(f"district_monthly.{y}.{m_}.n_students", n_src, dm[y][m_]["n_students"], 0)
    # Window averages recomputed from SOURCE values (not the packet's own months).
    src_months: dict[tuple[str, str], tuple[float, float]] = {}
    for _, r in mon.iterrows():
        y = canon_year(r["School Year"]); m_ = str(r["Month"]).strip()
        v = num(r["% with 90%+ Attendance (This Month)"]); n_src = num(r["Total Students (This Month)"])
        if v is not None and n_src is not None:
            src_months[(y, m_)] = (100 - v, n_src)
    WINDOWS = {"sep_mar_weighted": ["Sep", "Oct", "Nov", "Dec", "Jan", "Feb", "Mar"], "sep_nov_weighted": ["Sep", "Oct", "Nov"], "dec_feb_weighted": ["Dec", "Jan", "Feb"]}
    for y, e in dm.items():
        for key, ms in WINDOWS.items():
            sm = [src_months[(y, m_)] for m_ in ms if (y, m_) in src_months]
            exp = sum(a * n for a, n in sm) / sum(n for _, n in sm) if len(sm) == len(ms) else None
            check(f"district_monthly.{y}.{key}", exp, e.get(key), TOL_RATE)
    # monthly_records: recomputed from source months
    mr = packet["monthly_records"]
    all_years = sorted({y for (y, _) in src_months}); in_person = [y for y in all_years if y != "2020-21"]
    for m_ in ["Sep", "Oct", "Nov", "Dec", "Jan", "Feb", "Mar", "Apr", "May", "Jun"]:
        for label, yrs in [("all_years", all_years), ("in_person", in_person)]:
            vals = [(y, src_months[(y, m_)][0]) for y in yrs if (y, m_) in src_months]
            exp = mr["by_month"][m_][label]
            if not vals:
                check(f"monthly_records.{m_}.{label}", None, exp, 0); continue
            lo = min(vals, key=lambda t: t[1]); hi = max(vals, key=lambda t: t[1])
            check(f"monthly_records.{m_}.{label}.lowest", lo[0], exp["lowest"], 0)
            check(f"monthly_records.{m_}.{label}.lowest_value", lo[1], exp["lowest_value"], TOL_RATE)
            check(f"monthly_records.{m_}.{label}.highest", hi[0], exp["highest"], 0)
            check(f"monthly_records.{m_}.{label}.highest_value", hi[1], exp["highest_value"], TOL_RATE)
            srt = sorted(v for _, v in vals)
            check(f"monthly_records.{m_}.{label}.second_highest_value", srt[-2] if len(srt) > 1 else None, exp["second_highest_value"], TOL_RATE)
            check(f"monthly_records.{m_}.{label}.n_years", len(vals), exp["n_years"], 0)
    for key, ms in WINDOWS.items():
        for label, yrs in [("all_years", all_years), ("in_person", in_person)]:
            vals = []
            for y in yrs:
                sm = [src_months[(y, m_)] for m_ in ms if (y, m_) in src_months]
                if len(sm) == len(ms):
                    vals.append((y, sum(a * n for a, n in sm) / sum(n for _, n in sm)))
            exp = mr["windows"][key][label]
            if not vals:
                check(f"monthly_records.windows.{key}.{label}", None, exp, 0); continue
            lo = min(vals, key=lambda t: t[1]); hi = max(vals, key=lambda t: t[1])
            check(f"monthly_records.windows.{key}.{label}.lowest", lo[0], exp["lowest"], 0)
            check(f"monthly_records.windows.{key}.{label}.lowest_value", lo[1], exp["lowest_value"], TOL_RATE)
            check(f"monthly_records.windows.{key}.{label}.highest", hi[0], exp["highest"], 0)
            check(f"monthly_records.windows.{key}.{label}.highest_value", hi[1], exp["highest_value"], TOL_RATE)
            check(f"monthly_records.windows.{key}.{label}.n_years", len(vals), exp["n_years"], 0)

    # derived fields the page composes wording from: per_grade_shape, subgroup_peak_year,
    # monthly_records (all fields), BW fixed-panel level pairs, size-band medians
    pg = packet["per_grade"]; GRADES_L = ["K"] + [str(i) for i in range(1, 13)]
    for y in ["2021-22", "2024-25"]:
        vals = [(g, pg[g][y]["weighted"]) for g in GRADES_L]
        trough = min(vals, key=lambda t: t[1]); sh = packet["per_grade_shape"][y]
        check(f"per_grade_shape.{y}.trough_grade", trough[0], sh["trough_grade"], 0)
        check(f"per_grade_shape.{y}.trough_value", trough[1], sh["trough_value"], TOL_RATE)
        ups = [vals[i][0] for i in range(1, len(vals)) if vals[i][1] > vals[i - 1][1]]
        check(f"per_grade_shape.{y}.grades_higher_than_previous", ",".join(ups), ",".join(sh["grades_higher_than_previous"]), 0)
        check(f"per_grade_shape.{y}.k_to_grade1_direction", "up" if vals[1][1] > vals[0][1] else "down" if vals[1][1] < vals[0][1] else "flat", sh["k_to_grade1_direction"], 0)
        ti = [g for g, _ in vals].index(trough[0])
        check(f"per_grade_shape.{y}.monotone_decline_k_to_trough", str(all(vals[i][1] <= vals[i - 1][1] for i in range(1, ti + 1))).lower(), str(sh["monotone_decline_k_to_trough"]).lower(), 0)
        i9 = [g for g, _ in vals].index("9")
        check(f"per_grade_shape.{y}.monotone_rise_9_to_12", str(all(vals[i][1] >= vals[i - 1][1] for i in range(i9 + 1, len(vals)))).lower(), str(sh["monotone_rise_9_to_12"]).lower(), 0)
    for g in GRADES_L:
        check(f"per_grade_change.{g}", pg[g]["2024-25"]["weighted"] - pg[g]["2021-22"]["weighted"], packet["per_grade_change"][g], 2 * TOL_RATE)
    for sg in RACE:
        ip = [(y, st[sg][y]["weighted"]) for y in YEARS if y != "2020-21"]
        pk = max(ip, key=lambda t: t[1]); spy = packet["subgroup_peak_year"][sg]
        check(f"subgroup_peak_year.{sg}.peak_in_person_year", pk[0], spy["peak_in_person_year"], 0)
        check(f"subgroup_peak_year.{sg}.peak_value", pk[1], spy["peak_value"], TOL_RATE)
        check(f"subgroup_peak_year.{sg}.range_in_person", max(v for _, v in ip) - min(v for _, v in ip), spy["range_in_person"], 2 * TOL_RATE)
    for y in bwp["years"]:
        b = {u: v for (u, yy, s_), v in src90.items() if s_ == "BLACK" and yy == y and u in bw_panel}
        w_ = {u: v for (u, yy, s_), v in src90.items() if s_ == "WHITE" and yy == y and u in bw_panel}
        us = sorted(bw_panel)
        check(f"black_white_fixed_panel.{y}.black_mean_school", float(np.mean([100 * (1 - b[u]["n90"] / b[u]["n"]) for u in us])), bwp[y]["black_mean_school"], TOL_RATE)
        check(f"black_white_fixed_panel.{y}.white_mean_school", float(np.mean([100 * (1 - w_[u]["n90"] / w_[u]["n"]) for u in us])), bwp[y]["white_mean_school"], TOL_RATE)
    sv = stab["small_vs_large_ratio"]; vb = stab["volatility_by_size_latest_pair"]
    check("stability.small_vs_large_ratio.sd_change", vb["<300"]["sd_change"] / vb["600+"]["sd_change"], sv["sd_change"], 0.001)
    check("stability.small_vs_large_ratio.median_abs_change", vb["<300"]["median_abs_change"] / vb["600+"]["median_abs_change"], sv["median_abs_change"], 0.001)

    # composition_segments (DB demographics 2024-25)
    seg = packet["composition_segments"]
    demo = {}
    for mk, u, v in db_query("SELECT metric_key, school_ulcs, value FROM philly_school_year_metrics WHERE year='2024-25' AND subgroup='ALL' AND NOT suppressed AND value IS NOT NULL AND metric_key = ANY(%s)", (["pct_special_ed", "pct_white", "pct_english_learner", "pct_econ_disadv", "enrollment"],)):
        demo.setdefault(mk, {})[str(u)] = float(v)
    ca25 = {u: (100 - v["value"], v["n"]) for (u, yy, sg), v in served90.items() if sg == "ALL" and yy == "2024-25" and not v["suppressed"] and v["value"] is not None and v["n"] is not None}
    for mk in ["pct_special_ed", "pct_white", "pct_english_learner"]:
        rows = sorted([(demo[mk][u], ca25[u]) for u in demo.get(mk, {}) if u in ca25])
        n = len(rows); cuts = [rows[: n // 3], rows[n // 3: 2 * n // 3], rows[2 * n // 3:]]
        for i, t in enumerate(cuts):
            tn = sum(c[1][1] for c in t)
            check(f"composition_segments.{mk}[{i}].weighted_below_90", sum(c[1][0] * c[1][1] for c in t) / tn, seg["characteristics"][mk][i]["weighted_below_90"], TOL_RATE)
            check(f"composition_segments.{mk}[{i}].n_schools", len(t), seg["characteristics"][mk][i]["n_schools"], 0)
    econ = [v for u, v in demo.get("pct_econ_disadv", {}).items() if u in ca25]
    check("composition_segments.econ_disadv_top_coded.n_at_100", sum(1 for v in econ if v >= 99.95), seg["econ_disadv_top_coded"]["n_at_100"], 0)

    # sector_state_measure (DB)
    ssm = packet["sector_state_measure"]
    enroll = demo.get("enrollment", {})
    state_all = {(u, y): 100 - v["value"] for (u, y, sg), v in servedState.items() if sg == "ALL" and not v["suppressed"] and v["value"] is not None}
    for sec in ["REGULAR", "CHARTER"]:
        for y in ssm["years"]:
            cells = [(u, val) for (u, yy), val in state_all.items() if yy == y and school_meta.get(u, {}).get("type") == sec]
            exp = ssm["trend"][sec][y]
            if not cells:
                check(f"sector_state_measure.trend.{sec}.{y}", None, exp, 0); continue
            vs = np.array([v for _, v in cells])
            check(f"sector_state_measure.trend.{sec}.{y}.n_schools", len(cells), exp["n_schools"], 0)
            check(f"sector_state_measure.trend.{sec}.{y}.mean_school", vs.mean(), exp["mean_school"], TOL_RATE)
            check(f"sector_state_measure.trend.{sec}.{y}.median_school", np.median(vs), exp["median_school"], TOL_RATE)
            wc = [(v, enroll[u]) for u, v in cells if u in enroll]
            check(f"sector_state_measure.trend.{sec}.{y}.weighted_by_2024_25_enrollment", sum(v * w for v, w in wc) / sum(w for _, w in wc) if wc else None, exp["weighted_by_2024_25_enrollment"], TOL_RATE)
    LY = ssm["latest_year"]
    assert LY == sorted({y for (_, y) in state_all})[-1], f"packet latest_year {LY} != served {sorted({y for (_, y) in state_all})[-1]}"
    # fixed panel: schools with a value in every state year
    every = {u for (u, _) in state_all if all((u, y) in state_all for y in ssm["years"])}
    for sec in ["REGULAR", "CHARTER"]:
        secset = {u for u in every if school_meta.get(u, {}).get("type") == sec}
        check(f"sector_state_measure.fixed_panel.n_schools.{sec}", len(secset), ssm["fixed_panel"]["n_schools"][sec], 0)
        for y in ssm["years"]:
            cells = [(u, state_all[(u, y)]) for u in secset]
            exp = ssm["fixed_panel"]["trend"][sec][y]
            vs = np.array([v for _, v in cells])
            check(f"sector_state_measure.fixed_panel.{sec}.{y}.mean_school", vs.mean(), exp["mean_school"], TOL_RATE)
            check(f"sector_state_measure.fixed_panel.{sec}.{y}.median_school", np.median(vs), exp["median_school"], TOL_RATE)
            wc = [(v, enroll[u]) for u, v in cells if u in enroll]
            check(f"sector_state_measure.fixed_panel.{sec}.{y}.weighted", sum(v * w for v, w in wc) / sum(w for _, w in wc) if wc else None, exp["weighted_by_2024_25_enrollment"], TOL_RATE)
    # like-for-like by school type (latest year)
    for lvl, d in ssm["by_level_latest"].items():
        for sec in ["REGULAR", "CHARTER"]:
            cells = [(u, state_all[(u, LY)]) for (u, y) in state_all if y == LY and school_meta.get(u, {}).get("type") == sec and school_meta.get(u, {}).get("level") == lvl]
            exp = d[sec]
            if not cells:
                check(f"sector_state_measure.by_level_latest.{lvl}.{sec}", None, exp, 0); continue
            vs = np.array([v for _, v in cells])
            check(f"sector_state_measure.by_level_latest.{lvl}.{sec}.n_schools", len(cells), exp["n_schools"], 0)
            check(f"sector_state_measure.by_level_latest.{lvl}.{sec}.mean_school", vs.mean(), exp["mean_school"], TOL_RATE)
            wc = [(v, enroll[u]) for u, v in cells if u in enroll]
            check(f"sector_state_measure.by_level_latest.{lvl}.{sec}.weighted", sum(v * w for v, w in wc) / sum(w for _, w in wc) if wc else None, exp["weighted_by_2024_25_enrollment"], TOL_RATE)
    # directory counts, re-derived from the served directory (the Master School
    # List itself is not re-parsed here — that remains an unchecked input).
    for sec in ["REGULAR", "CHARTER"]:
        secs = [u for u, m in school_meta.items() if m.get("type") == sec]
        check(f"sector_state_measure.directory.{sec}.n_schools", len(secs), ssm["directory"][sec]["n_schools"], 0)
        check(f"sector_state_measure.directory.{sec}.standalone_high", sum(1 for u in secs if school_meta[u].get("level") == "High"), ssm["directory"][sec]["standalone_high"], 0)
        check(f"sector_state_measure.directory.{sec}.serving_high_school_grades", sum(1 for u in secs if "High" in str(school_meta[u].get("level"))), ssm["directory"][sec]["serving_high_school_grades"], 0)
        check(f"sector_state_measure.directory.{sec}.mixed_grade_with_high_school", sum(1 for u in secs if "High" in str(school_meta[u].get("level")) and school_meta[u].get("level") != "High"), ssm["directory"][sec]["mixed_grade_with_high_school"], 0)
    for sg in ["ECON_DISADV", "IEP", "ELL", "BLACK", "WHITE"]:
        for sec in ["REGULAR", "CHARTER"]:
            allby = {u: state_all[(u, LY)] for (u, y) in state_all if y == LY and school_meta.get(u, {}).get("type") == sec}
            sub = {u: 100 - v["value"] for (u, y, s), v in servedState.items() if s == sg and y == LY and not v["suppressed"] and v["value"] is not None and u in allby}
            exp = ssm["subgroup_gaps_latest"][sg][sec]
            if not sub:
                check(f"sector_state_measure.gaps.{sg}.{sec}", None, exp, 0); continue
            gaps = [sub[u] - allby[u] for u in sub]
            check(f"sector_state_measure.gaps.{sg}.{sec}.n_schools", len(sub), exp["n_schools"], 0)
            check(f"sector_state_measure.gaps.{sg}.{sec}.mean_gap", float(np.mean(gaps)), exp["mean_gap"], TOL_RATE)
            check(f"sector_state_measure.gaps.{sg}.{sec}.mean_subgroup_below_90", float(np.mean(list(sub.values()))), exp["mean_subgroup_below_90"], TOL_RATE)

    report["sections"]["D_packet_recomputation"] = {"leaves_compared": compared, "mismatches": len(mismatches), "examples": mismatches[:40], "status": "PASS" if not mismatches else "FAIL",
        "scope_note": "These are specified scalar comparisons chosen by hand (one per packet value listed in this script), not a test of every displayed value or of the page's wording."}
    report["sections"]["not_checked"] = [
        "Enrollment weights (2024-25 enrollment per school) and directory classifications (sector, school type, admission type, Renaissance flag) are read from the served database, not re-derived from the district's enrollment and Master School List files.",
        "The New York City comparison on the analysis page is taken from that city's own analysis packet, which carries its own unverified warning; it is not checked here.",
        "The district's June 2023 attendance report figures in meta.external_context are hand-transcribed and not recomputed.",
        "Rendered prose, chart annotations, and accessibility text are not tested by this program. The packet fields the page composes universal statements from (monthly_records incl. window selectors, per_grade_shape, subgroup_peak_year, stability counts and ratios, Black–White fixed-panel level pairs) ARE directly compared above; the prose that uses them is not.",
        "The cross-publisher check covers district-run schools in attendance years 2020-21 through 2023-24 only; charter rows and the two pre-pandemic years are not cross-checked against any second publisher.",
        "Per-grade student counts (n_students) and the composition thirds' share boundaries are not separately compared.",
    ]

    # ---------------- E. verdict ----------------
    a_ok = all(v["status"] == "PASS" for v in secA.values())
    b_ok = report["sections"]["B_base_case_state_files"]["attendance_persistence_rate"]["status"] == "PASS"
    c_ok = all(r["source_only_directory"] == 0 and r["served_only"] == 0 for r in compl)
    d_ok = not mismatches
    verdict = "PASS" if (a_ok and b_ok and c_ok and d_ok) else "FAIL"
    report["verdict"] = {"overall": verdict, "A_district_base_case": a_ok, "B_state_base_case": b_ok, "C_completeness": c_ok, "D_packet_recomputation": d_ok}

    # ---------------- write ----------------
    md = ["# QA — Philadelphia chronic absenteeism: served attendance cells and specified packet comparisons\n",
          f"Packet generated: {packet['meta']['generated']}\n",
          f"\n## Verdict: **{verdict}**\n",
          "| Section | Result |\n|---|---|",
          f"| A. District file → database (attendance_rate_above_90, every year × subgroup, values + denominators; average_daily_attendance) | {'PASS' if a_ok else 'FAIL'} |",
          f"| B. State files → database (attendance_persistence_rate, attendance years 2017-18..2023-24 from the 2018-19..2024-25 workbooks, every subgroup) | {'PASS' if b_ok else 'FAIL'} |",
          f"| C. Completeness (every source cell served, directory schools) | {'PASS' if c_ok else 'FAIL'} |",
          f"| D. Packet recomputation ({compared} specified comparisons, numeric and year-label, recomputed independently in Python) | {'PASS' if d_ok else 'FAIL'} ({len(mismatches)} mismatches) |",
          "\n## A. District file base case\n", f"```json\n{json.dumps(secA, indent=2)}\n```",
          "\n## B. State files base case\n", f"```json\n{json.dumps(report['sections']['B_base_case_state_files'], indent=2)}\n```",
          "\n## Not checked by this program\n", *[f"- {t}\n" for t in report["sections"]["not_checked"]],
          "\n## C. Completeness by year\n", "| metric | year | source cells | served | source-only (directory) | source-only (closed/non-directory) | served-only |", "|---|---|---:|---:|---:|---:|---:|"]
    for r in compl:
        md.append(f"| {r['metric']} | {r['year']} | {r['source_cells']} | {r['served_cells']} | {r['source_only_directory']} | {r['source_only_non_directory']} | {r['served_only']} |")
    md += ["\n## D. Packet recomputation\n", f"{compared} specified comparisons (numeric values and year-label strings, listed in this script) recomputed from the source files and database; {len(mismatches)} differ beyond tolerance (rates ±0.0051 at 4dp, correlations ±0.0006, counts and labels exact).\n"]
    md += ["\n## Non-numeric token census\n", "District yearly file (District sector), by column:\n"]
    for col, toks in report["district_file_nonnumeric_tokens"]["tokens"].items():
        md.append(f"- {col}: {toks if toks else '(none)'}\n")
    md += ["\nState workbooks (matched-school attendance cells):\n"]
    for wb, toks in report["state_nonnumeric_tokens"]["by_workbook"].items():
        md.append(f"- {wb}: {toks if toks else '(none)'}\n")
    md.append(f"\nUnexpected tokens (not a documented suppression code): {report['state_nonnumeric_tokens']['unexpected'] or 'none'}\n")
    if mismatches:
        md.append("```json\n" + json.dumps(mismatches[:40], indent=2) + "\n```")
    (common.QA_REPORTS_DIR / "__absenteeism_full_qa.md").write_text("\n".join(md))
    (common.QA_REPORTS_DIR / "__absenteeism_full_qa.json").write_text(json.dumps(report, indent=2))
    print(f"VERDICT {verdict} | A={a_ok} B={b_ok} C={c_ok} D={d_ok} | comparisons={compared} mismatches={len(mismatches)}")
    for k, v in secA.items():
        print(f"  A {k}: {v['status']} value_mism={v['value_mismatches']} denom_mism={v['denominator_mismatches']} supp_mism={v['suppression_mismatches']} src_only_dir={v['source_only_directory_schools']} src_only_closed={v['source_only_non_directory_schools']} served_only={v['served_only']}")
    b = report["sections"]["B_base_case_state_files"]["attendance_persistence_rate"]
    print(f"  B state: {b['status']} common={b['common_cells']} value_mism={b['value_mismatches']} supp_mism={b['suppression_mismatches']} src_only={b['source_only_directory_schools']} served_only={b['served_only']}")
    for m in mismatches[:15]:
        print("  D mismatch:", m)


if __name__ == "__main__":
    main()