pipeline_miami/discover_srs.py

"""Discover FLDOE State Report of Schools (SRS) assessment files for a results
year and register the spring administrations in docs/cities/miami/sources.yaml.

The per-year results page (fldoe.org/accountability/assessments/k-12-student-
assessment/results/<year>.stml) links one .xls per subject × grade ×
administration; the file-name prefixes drift by year (findings.md F.5), so
the registry is built from the page rather than templated.

Usage:  python -m pipeline_miami.discover_srs 2025 [2024 ...]
Then:   python -m pipeline_miami.discover --all
"""
from __future__ import annotations

import re
import sys

import requests
import yaml

from pipeline_miami.discover import SOURCES_YAML, UA, fetch_url_for

PAGE = "https://www.fldoe.org/accountability/assessments/k-12-student-assessment/results/{year}.stml"
# file-name fragment → (metric family, grade or subject)
PATTERNS = [
    (re.compile(r"ELA(\d{2})SRS", re.I), lambda m: ("fast_ela", f"g{int(m.group(1))}")),
    (re.compile(r"Math(\d{2})SRS", re.I), lambda m: ("fast_math", f"g{int(m.group(1))}")),
    (re.compile(r"Alg1SRS", re.I), lambda m: ("eoc", "alg1")),
    (re.compile(r"(?<![A-Za-z])G(\d)SRS", re.I), lambda m: ("science", f"g{m.group(1)}")),
    (re.compile(r"GeoSRS", re.I), lambda m: ("eoc", "geometry")),
    (re.compile(r"Bio1SRS", re.I), lambda m: ("eoc", "biology")),
    (re.compile(r"CivSRS", re.I), lambda m: ("eoc", "civics")),
    (re.compile(r"USHSRS", re.I), lambda m: ("eoc", "ushistory")),
]


def classify(fname: str):
    if not re.search(r"spring", fname, re.I):
        return None  # fall / winter / summer EOC administrations and non-SRS files
    for pat, fn in PATTERNS:
        m = pat.search(fname)
        if m:
            return fn(m)
    return None


def main(years: list[str]) -> None:
    sources = yaml.safe_load(open(SOURCES_YAML))
    existing = {s["source_id"] for s in sources}
    urls_known = {s.get("url") for s in sources}
    new = []
    for y in years:
        url = PAGE.format(year=y)
        r = requests.get(fetch_url_for(url), headers={"User-Agent": UA}, timeout=120)
        r.raise_for_status()
        links = set(re.findall(r'href="([^"]+\.xls[x]?)"', r.text, re.I))
        sy = f"{int(y) - 1}-{y[2:]}"
        for href in sorted(links):
            fname = href.rsplit("/", 1)[-1]
            c = classify(fname)
            if not c:
                continue
            fam, part = c
            sid = f"fldoe_srs_{fam}_{part}_spring{y}"
            full = href if href.startswith("http") else "https://www.fldoe.org" + href
            if sid in existing or full in urls_known:
                continue
            new.append({
                "source_id": sid, "role": "ingested", "publisher": "Florida Department of Education, Bureau of K-12 Student Assessment",
                "url": full, "format": "xls", "year_format": "spring_yy", "year": sy, "header_row": 5,
                "join_key": ["District Number", "School Number"],
                "metrics_backed": [f"{fam}_{part}_proficiency" if fam != "eoc" else f"eoc_{part}_proficiency"],
                "notes": f"State Report of Schools, spring {y} administration; discovered from {url} by pipeline_miami/discover_srs.py. All students only; '*' n<10.",
            })
            existing.add(sid)
    if not new:
        print("nothing new")
        return
    text = open(SOURCES_YAML).read().rstrip("\n") + "\n\n# ─────────────────────────────────────────────────────────────────────────────\n# Assessment SRS files discovered from the FLDOE results pages (discover_srs.py)\n# ─────────────────────────────────────────────────────────────────────────────\n\n" + yaml.safe_dump(new, sort_keys=False, allow_unicode=True, width=120)
    open(SOURCES_YAML, "w").write(text)
    for s in new:
        print(f"  + {s['source_id']:40s} {s['url']}")
    print(f"registered {len(new)} SRS sources")


if __name__ == "__main__":
    main(sys.argv[1:] or ["2025"])