"""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/.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"(? 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"])