congress-analyzer/tests/parity_check.py

#!/usr/bin/env python3
"""Phase 3 KPI parity gate: compares legacy/*.html DATA blocks against
data/119/members/<id>.json metrics blocks for 8 sample members."""
import json
import re
import sys
from pathlib import Path

ROOT = Path("/home/user/polisci")
LEGACY = ROOT / "legacy"
MEMBERS = ROOT / "data" / "119" / "members"

PAIRS = [
    ("ThomasMassie119.html", "M001184", "Thomas Massie"),
    ("RoKhanna119.html", "K000389", "Ro Khanna"),
    ("AlexandriaOcasioCortez119.html", "O000172", "Alexandria Ocasio-Cortez"),
    ("IlhanOmar119.html", "O000173", "Ilhan Omar"),
    ("MarjorieTaylorGreene119.html", "G000596", "Marjorie Taylor Greene"),
    ("JimJordan119.html", "J000289", "Jim Jordan"),
    ("ByronDonalds119.html", "D000032", "Byron Donalds"),
    ("LindseyGraham119.html", "S293", "Lindsey Graham"),
]

SCALAR_KPIS = [
    "total", "voting", "yeas", "nays", "nv", "present",
    "blocked_dem_count", "blocked_rep_count",
    "voted_with_gop", "voted_against_gop",
    "voted_with_dem", "voted_against_dem",
]
CANONICAL_ALIGN = ["Helped Republicans", "Helped Democrats", "Helped Both", "Helped Neither"]


def extract_legacy(path):
    text = path.read_text(encoding="utf-8")
    m = re.search(r"const DATA\s*=\s*(\{.+?\});", text)
    if not m:
        raise RuntimeError(f"No DATA block in {path}")
    return json.loads(m.group(1))


def cmp_member(legacy_file, mid, display):
    out_lines = []
    fails = []

    legacy_path = LEGACY / legacy_file
    new_path = MEMBERS / f"{mid}.json"
    legacy = extract_legacy(legacy_path)
    new_full = json.loads(new_path.read_text(encoding="utf-8"))
    new = new_full["metrics"]

    # scalar KPIs
    for k in SCALAR_KPIS:
        lv = legacy.get(k)
        nv = new.get(k)
        if lv != nv:
            fails.append(f"  {k}: legacy={lv} new={nv}  (delta {(nv if nv is not None else 0) - (lv if lv is not None else 0)})")

    # lone_wolf (only if legacy has it)
    if "lone_wolf" in legacy:
        if legacy["lone_wolf"] != new.get("lone_wolf"):
            fails.append(f"  lone_wolf: legacy={legacy['lone_wolf']} new={new.get('lone_wolf')}")

    # alignment canonical 4
    la = legacy.get("alignment", {})
    na = new.get("alignment", {})
    for k in CANONICAL_ALIGN:
        lv, nv = la.get(k, 0), na.get(k, 0)
        if lv != nv:
            fails.append(f"  alignment[{k}]: legacy={lv} new={nv}  (delta {nv - lv})")
    # N/A sums
    la_na = sum(v for k, v in la.items() if k.startswith("N/A"))
    na_na = sum(v for k, v in na.items() if k.startswith("N/A"))
    if la_na != na_na:
        fails.append(f"  alignment[N/A sum]: legacy={la_na} new={na_na}  (delta {na_na - la_na})")

    # blocked
    lb = legacy.get("blocked", {})
    nb = new.get("blocked", {})
    for k in ("Democrat", "Republican"):
        if lb.get(k, 0) != nb.get(k, 0):
            fails.append(f"  blocked[{k}]: legacy={lb.get(k,0)} new={nb.get(k,0)}")

    # monthly: align by month string
    l_months = legacy.get("months", [])
    n_months = new.get("months", [])
    l_monthly = legacy.get("monthly", {})
    n_monthly = new.get("monthly", {})
    all_months = sorted(set(l_months) | set(n_months))
    for cat in CANONICAL_ALIGN:
        l_arr = l_monthly.get(cat, [])
        n_arr = n_monthly.get(cat, [])
        l_map = dict(zip(l_months, l_arr))
        n_map = dict(zip(n_months, n_arr))
        for mo in all_months:
            lv = l_map.get(mo, 0)
            nv = n_map.get(mo, 0)
            if lv != nv:
                fails.append(f"  monthly[{cat}][{mo}]: legacy={lv} new={nv}  (delta {nv - lv})")

    # rows length must equal total
    l_rows = len(legacy.get("rows", []))
    n_rows = len(new.get("rows", []))
    if l_rows != legacy.get("total"):
        fails.append(f"  legacy rows length {l_rows} != legacy total {legacy.get('total')}")
    if n_rows != new.get("total"):
        fails.append(f"  new rows length {n_rows} != new total {new.get('total')}")
    if l_rows != n_rows:
        fails.append(f"  rows length: legacy={l_rows} new={n_rows}  (delta {n_rows - l_rows})")

    if not fails:
        out_lines.append(f"{mid} {display} — PASS  (all {len(SCALAR_KPIS)}+1 scalar KPIs + alignment4 + blocked + monthly + rows)")
        return True, out_lines
    else:
        out_lines.append(f"{mid} {display} — FAIL")
        out_lines.extend(fails)
        return False, out_lines


def main():
    report = []
    report.append("# Phase 3 — KPI Parity Gate Report")
    report.append("")
    report.append("Compared legacy/*.html DATA blocks vs data/119/members/<id>.json metrics.")
    report.append("")
    report.append("## Per-member results")
    report.append("")
    passes = 0
    for legacy_file, mid, display in PAIRS:
        ok, lines = cmp_member(legacy_file, mid, display)
        if ok:
            passes += 1
        report.extend(lines)
        report.append("")
    summary = f"Phase 3 gate: {passes}/{len(PAIRS)} members PASS"
    report.append("## Summary")
    report.append("")
    report.append(summary)
    report.append("")

    # MTG banner check
    mtg = json.loads((MEMBERS / "G000596.json").read_text(encoding="utf-8"))
    report.append("## MTG (G000596) banner check")
    report.append("")
    report.append(f"- served_partial: {mtg.get('served_partial')}")
    report.append(f"- metrics.total: {mtg['metrics'].get('total')}")
    report.append(f"- served_from: {mtg.get('served_from')}, served_to: {mtg.get('served_to')}")
    if mtg.get("served_partial") or mtg["metrics"].get("total") == 0:
        report.append("- Banner SHOULD trigger.")
    else:
        report.append("- No banner needed (full term, has votes).")
    report.append("")

    # deep-link in template/app.js
    appjs = (ROOT / "template" / "app.js").read_text(encoding="utf-8")
    has_pushstate = "pushState" in appjs
    has_popstate = "popstate" in appjs
    report.append("## Deep-link URL behavior (template/app.js)")
    report.append("")
    report.append(f"- pushState present: {has_pushstate}")
    report.append(f"- popstate present: {has_popstate}")
    report.append("")

    # CDN check
    import subprocess
    cdn = subprocess.run(
        ["grep", "-RE", r"cdn\.|cdnjs|jsdelivr|unpkg", str(ROOT / "template")],
        capture_output=True, text=True,
    )
    report.append("## CDN traffic check (template/)")
    report.append("")
    # Filter out comment lines inside vendored files (documentation references, not runtime URLs)
    real_hits = []
    for line in cdn.stdout.strip().splitlines():
        if "/vendor/" in line and ("More information" in line or "Do NOT use SRI" in line):
            continue  # vendored chart.js header comment
        real_hits.append(line)
    if real_hits:
        report.append("- FAIL — CDN references found:")
        for line in real_hits:
            report.append(f"  {line}")
    else:
        report.append("- PASS — no runtime CDN references (vendored-file documentation comments ignored).")
        if cdn.stdout.strip():
            report.append("  Note: documentation/comment-only mentions inside template/vendor/ were ignored:")
            for line in cdn.stdout.strip().splitlines():
                report.append(f"    {line}")
    report.append("")

    # decision
    report.append("## Decision")
    report.append("")
    if passes == len(PAIRS):
        report.append("**GATE PASSED — safe to proceed to Phase 4.**")
    else:
        report.append("**GATE FAILED — investigate the delta(s) before proceeding.**")
    report.append("")

    text = "\n".join(report)
    print(text)
    (ROOT / "research" / "PHASE3_PARITY.md").write_text(text, encoding="utf-8")
    return 0 if passes == len(PAIRS) else 1


if __name__ == "__main__":
    sys.exit(main())