pipeline_miami/verify/survey_source_check.py

"""Compare regenerated survey composites with the independent raw-workbook reader.

Run: python -m pipeline_miami.verify.survey_source_check
This checks source artifacts without a database; it is not a served-data verdict.
"""
from __future__ import annotations

import csv
import hashlib
import json
import math
from pathlib import Path
import statistics

from pipeline_miami.verify import common
from pipeline_miami.verify.connectors import source_file


def read_csv(path: Path) -> list[dict[str, str]]:
    with path.open(newline="") as handle:
        return list(csv.DictReader(handle))


def main() -> None:
    reference = common.REPO_ROOT / "data/cities/miami/reference"
    items = read_csv(reference / "survey_items.csv")
    crosswalk = {(r["respondent"], r["year"], r["item_no"]): r for r in items}
    results, verified_sources = [], []

    for key in source_file._SURVEY_COLUMNS:
        path = common.latest_source_path(f"mdcps_climate_survey_{key}")
        provenance = json.loads((path.parent / "provenance.json").read_text())
        if hashlib.sha256(path.read_bytes()).hexdigest() != provenance["sha256"]:
            raise ValueError(f"Source hash does not match its provenance: {key}")
        verified_sources.append(key)
        respondent, year = key.split("_")
        rows = read_csv(reference / f"parsed/survey/mdcps_climate_survey_{key}.csv")
        for domain in ("safety", "leadership", "teaching", "resources"):
            core = {r["canonical"] for r in items
                    if r["respondent"] == respondent and r["domain"] == domain and r["core"] == "True"}
            scores: dict[str, list[float]] = {}
            for row in rows:
                values = scores.setdefault(row["school_number"], [])
                item = crosswalk[(respondent, year, row["item_no"])]
                if row["suppressed"] == "True" or item["domain"] != domain or item["core"] != "True":
                    continue
                selected = row["pct_disagree"] if item["reverse"] == "True" else row["pct_agree"]
                if selected:
                    value = float(selected)
                    if math.isfinite(value) and 0 <= value <= 100:
                        values.append(value)

            actual = source_file.climate_survey(year, respondent, domain)
            mismatches = []
            for (school, _), observed in actual.items():
                values = scores[school]
                expected = statistics.mean(values) if len(values) * 2 >= len(core) else None
                missing_differs = (expected is None) != (observed["value"] is None)
                value_differs = (expected is not None and observed["value"] is not None
                                 and abs(expected - observed["value"]) > 1e-6)
                if missing_differs or value_differs:
                    mismatches.append((school, expected, observed))
            results.append(dict(key=key, domain=domain, schools=len(actual), mismatches=len(mismatches)))
            print(key, domain, len(actual), "schools; mismatches", len(mismatches), flush=True)
            if mismatches:
                raise ValueError(f"{key} {domain}: {mismatches[:3]}")

    common.ensure_qa_dir()
    output = common.QA_REPORTS_DIR / "survey_favorable_response_source_check.json"
    output.write_text(json.dumps({
        "scoring_version": "favorable-response-v2",
        "source_sha256_verified": len(verified_sources),
        "checks": results,
        "note": "Independent raw workbook reader compared with regenerated parsed CSVs and the declared favorable scoring, without a served database.",
    }, indent=2) + "\n")


if __name__ == "__main__":
    main()