pipeline_miami/parse/abc_book.py

"""Parse an M-DCPS Attendance Zone Recommendations ("ABC book") PDF into a
per-school capacity/membership table.

2026-27 edition layout: a "Capacity Membership Data" page per targeted school
with a "NNNN - School Name" header and rows
  Grade Configuration | Student Enrollment | Perm FISH Cap | % Utilization
  Perm FISH Cap | Perm & Temp FISH Cap | % Utilization Perm & Temp
for the current year and two projected years (labels wrap across lines,
e.g. "% Utilization Perm FISH\n32% N/A N/A\nCap").

Independent check (no second parser needed): the printed utilization must
equal enrollment / capacity to within 1 point wherever both are numeric.

Output: data/cities/miami/reference/parsed/<source_id>.csv + .gate.json

Run:  python -m pipeline_miami.parse.abc_book mdcps_abc_book_2026-27
"""
from __future__ import annotations

import json
import pathlib
import re
import sys

import pandas as pd
import pdfplumber

REPO_ROOT = pathlib.Path(__file__).resolve().parents[2]
SOURCES = REPO_ROOT / "data" / "cities" / "miami" / "sources"
OUT_DIR = REPO_ROOT / "data" / "cities" / "miami" / "reference" / "parsed"

TOK = r"(N/A|[\d,]+(?:\.\d+)?%?)"
HEADER_RE = re.compile(r"^(\d{4})\s*[-–]\s*(.+?)\s*$", re.M)
YEARS_RE = re.compile(r"(\d{4})-(\d{4})\s+(\d{4})-(\d{4})\s+(\d{4})-(\d{4})")


def _latest_file(source_id: str) -> pathlib.Path:
    d = SOURCES / source_id
    for day in sorted([p for p in d.iterdir() if p.is_dir()], reverse=True):
        for f in day.iterdir():
            if f.suffix.lower() == ".pdf":
                return f
    raise FileNotFoundError(source_id)


def _val(tok: str | None) -> float | None:
    if tok is None or tok.upper() == "N/A":
        return None
    return float(tok.replace(",", "").replace("%", ""))


def _triple(text: str, label_re: str) -> list[float | None]:
    m = re.search(label_re + r"\s*" + TOK + r"\s+" + TOK + r"\s+" + TOK, text, re.S)
    if not m:
        return [None, None, None]
    return [_val(m.group(1)), _val(m.group(2)), _val(m.group(3))]


def parse(pdf: pdfplumber.PDF) -> list[dict]:
    recs: list[dict] = []
    for pno, page in enumerate(pdf.pages, start=1):
        t = page.extract_text() or ""
        if "Capacity Membership Data" not in t:
            continue
        # Footnote markers and parenthetical asides sit inside the number cells
        # ("1473*", "*1036", "240 (135 K-5)", "28.77% (16%)"); drop them before matching.
        t = re.sub(r"\([^)]*\)", " ", t).replace("*", " ")
        h = HEADER_RE.search(t)
        if not h:
            continue
        yrs = YEARS_RE.search(t)
        years = [f"{yrs.group(1)}-{yrs.group(2)[2:]}", f"{yrs.group(3)}-{yrs.group(4)[2:]}", f"{yrs.group(5)}-{yrs.group(6)[2:]}"] if yrs else [None, None, None]
        grade_m = re.search(r"Grade Configuration\s+(.+)", t)
        enr = _triple(t, r"Student Enrollment")
        perm = _triple(t, r"Perm FISH Cap")
        util_perm = _triple(t, r"% Utilization Perm FISH(?:\s*Cap)?")
        pt = _triple(t, r"Perm & Temp FISH Cap")
        util_pt = _triple(t, r"% Utilization Perm &(?:\s*Temp)?")
        recs.append({
            "page": pno, "school_number": h.group(1), "school_name": h.group(2).strip(),
            "year_current": years[0], "year_proj1": years[1], "year_proj2": years[2],
            "grade_configuration": grade_m.group(1).strip() if grade_m else None,
            "enrollment_current": enr[0], "enrollment_proj1": enr[1], "enrollment_proj2": enr[2],
            "perm_fish_current": perm[0], "perm_fish_proj1": perm[1], "perm_fish_proj2": perm[2],
            "util_perm_current": util_perm[0], "util_perm_proj1": util_perm[1], "util_perm_proj2": util_perm[2],
            "perm_temp_fish_current": pt[0], "perm_temp_fish_proj1": pt[1], "perm_temp_fish_proj2": pt[2],
            "util_perm_temp_current": util_pt[0], "util_perm_temp_proj1": util_pt[1], "util_perm_temp_proj2": util_pt[2],
        })
    return recs


def main(source_id: str) -> None:
    path = _latest_file(source_id)
    with pdfplumber.open(path) as pdf:
        recs = parse(pdf)
    df = pd.DataFrame(recs)
    checks = []
    for _, r in df.iterrows():
        for suf in ("current", "proj1", "proj2"):
            e, c, u = r[f"enrollment_{suf}"], r[f"perm_fish_{suf}"], r[f"util_perm_{suf}"]
            if e is not None and c and u is not None and not pd.isna(e) and not pd.isna(c) and not pd.isna(u):
                checks.append({"school_number": r.school_number, "period": suf, "printed": u, "recomputed": round(e / c * 100, 2), "ok": abs(e / c * 100 - u) <= 1.0})
    fails = [c for c in checks if not c["ok"]]
    df.insert(0, "source_id", source_id)
    OUT_DIR.mkdir(parents=True, exist_ok=True)
    df.to_csv(OUT_DIR / f"{source_id}.csv", index=False)
    summary = {"source_id": source_id, "file": str(path.relative_to(REPO_ROOT)), "schools": len(df),
               "arith_checks": len(checks), "arith_failures": fails, "pass": len(df) > 0 and not fails,
               "schools_list": df[["school_number", "school_name", "enrollment_current", "perm_fish_current", "util_perm_current"]].to_dict("records")}
    (OUT_DIR / f"{source_id}.gate.json").write_text(json.dumps(summary, indent=1, default=str))
    print(json.dumps({k: v for k, v in summary.items() if k != "schools_list"}, indent=1, default=str))
    print(df[["school_number", "school_name", "enrollment_current", "perm_fish_current", "util_perm_current", "enrollment_proj1", "enrollment_proj2"]].to_string(index=False))
    if not summary["pass"]:
        sys.exit(2)


if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "mdcps_abc_book_2026-27")