"""First-fetch / discovery for Miami-Dade sources. Reads docs/cities/miami/sources.yaml, fetches each source whose URL has been resolved past TBD, records provenance (sha256 + retrieved_at + bytes + url + fetched_from) to data/cities/miami/sources///, and emits a per-source findings markdown documenting the sheets / columns / sample rows. Miami-specific rules (see README.md): www.fldoe.org is rewritten to origin.fldoe.org for the actual request; the MSID download primes a session cookie first. Otherwise identical in spirit to pipeline_philly/discover.py. Usage: python -m pipeline_miami.discover # list python -m pipeline_miami.discover --source python -m pipeline_miami.discover --all python -m pipeline_miami.discover --source --force # re-fetch today even if present """ from __future__ import annotations import argparse import datetime as dt import hashlib import json import pathlib import re import sys from typing import Any import requests import yaml REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent SOURCES_YAML = REPO_ROOT / "docs" / "cities" / "miami" / "sources.yaml" DATA_DIR = REPO_ROOT / "data" / "cities" / "miami" / "sources" FINDINGS_DIR = REPO_ROOT / "pipeline_miami" / "findings" UA = "Mozilla/5.0 (X11; Linux x86_64) edu-data/miami-discover" FLDOE_WWW = "https://www.fldoe.org/" FLDOE_ORIGIN = "https://origin.fldoe.org/" MSID_INDEX = "https://eds.fldoe.org/EDS/MasterSchoolID/index.cfm" def load_sources() -> list[dict[str, Any]]: with open(SOURCES_YAML) as f: return yaml.safe_load(f) def is_resolved(src: dict[str, Any]) -> bool: url = src.get("url") return bool(url and url != "TBD") def fetch_url_for(url: str) -> str: """The URL actually requested: the FLDOE origin host, else the URL itself.""" if url.startswith(FLDOE_WWW): return FLDOE_ORIGIN + url[len(FLDOE_WWW):] return url def filename_for(src: dict[str, Any]) -> str: url = src["url"] fname = url.rsplit("/", 1)[-1].split("?")[0] # MSID download and REST queries have no useful last segment. if not fname or "." not in fname or fname.endswith(".cfm"): ext = {"txt": "txt", "json": "json", "csv": "csv"}.get(src.get("format", ""), src.get("format", "bin")) fname = f"{src['source_id']}.{ext}" return fname def fetch(src: dict[str, Any], force: bool = False) -> pathlib.Path: sid = src["source_id"] url = src["url"] today = dt.date.today().isoformat() out_dir = DATA_DIR / sid / today out_dir.mkdir(parents=True, exist_ok=True) path = out_dir / filename_for(src) prov_path = out_dir / "provenance.json" if prov_path.exists() and not force: # Already fetched today (or placed here with provenance by hand): # re-use it, but still (re)inspect so findings stay current. existing = [f for f in out_dir.iterdir() if f.name != "provenance.json"] if existing: print(f"[reuse] {sid}: {existing[0].name} (fetched earlier; --force to re-fetch)", file=sys.stderr) return existing[0] real_url = fetch_url_for(url) print(f"[fetch] {sid} -> {real_url}", file=sys.stderr) session = requests.Session() session.headers["User-Agent"] = UA if "MasterSchoolID" in url: session.get(MSID_INDEX, timeout=60) # primes the ColdFusion session cookie r = session.get(real_url, timeout=300, stream=True) r.raise_for_status() h = hashlib.sha256() n_bytes = 0 with open(path, "wb") as f: for chunk in r.iter_content(chunk_size=1 << 16): if not chunk: continue f.write(chunk) h.update(chunk) n_bytes += len(chunk) sha = h.hexdigest() print(f"[fetch] wrote {path} ({n_bytes:,} bytes, sha256={sha[:12]}...)", file=sys.stderr) provenance: dict[str, Any] = { "source_id": sid, "url": url, "retrieved_at": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z"), "sha256": sha, "bytes": n_bytes, "publisher": src.get("publisher"), "format": src.get("format"), } if real_url != url: provenance["fetched_from"] = real_url if src.get("notes"): provenance["notes"] = str(src["notes"]).strip() prov_path.write_text(json.dumps(provenance, indent=2)) return path # ── inspectors ─────────────────────────────────────────────────────────────── def _find_header_row(raw, must: tuple[str, ...] = ("School",)) -> int | None: for i in range(min(12, len(raw))): row = [str(v) for v in raw.iloc[i].tolist()] if any(any(m.lower() in c.lower() for c in row) for m in must): return i return None def inspect_excel(path: pathlib.Path) -> dict[str, Any]: """Sheets + detected header row + columns + first data row; xls via xlrd.""" import pandas as pd out: dict[str, Any] = {"format": "excel", "sheets": {}} xf = pd.ExcelFile(path) out["sheet_names"] = list(xf.sheet_names) for sheet in xf.sheet_names[:8]: try: raw = pd.read_excel(xf, sheet_name=sheet, header=None, nrows=15) except Exception as e: # noqa: BLE001 out["sheets"][sheet] = {"error": str(e)} continue h = _find_header_row(raw) entry: dict[str, Any] = {"header_row_1based": None if h is None else h + 1} if h is None: entry["first_rows"] = [[None if pd.isna(v) else str(v)[:60] for v in raw.iloc[i].tolist()[:12]] for i in range(min(4, len(raw)))] else: df = pd.read_excel(xf, sheet_name=sheet, header=h, nrows=5) entry["n_columns"] = len(df.columns) entry["columns"] = [str(c) for c in df.columns] entry["sample_row"] = ( {str(k): (None if pd.isna(v) else str(v)) for k, v in df.iloc[0].items()} if len(df) > 0 else None ) out["sheets"][sheet] = entry return out def inspect_csv(path: pathlib.Path, sep: str = ",") -> dict[str, Any]: import pandas as pd out: dict[str, Any] = {"format": "csv"} df = pd.read_csv(path, nrows=5, low_memory=False, sep=sep, dtype=str, encoding="latin1") with open(path, "rb") as f: out["n_rows"] = sum(1 for _ in f) - 1 out["n_columns"] = len(df.columns) out["columns"] = [str(c) for c in df.columns] out["sample_row"] = ( {str(k): (None if pd.isna(v) else str(v)) for k, v in df.iloc[0].items()} if len(df) > 0 else None ) return out def inspect_pdf(path: pathlib.Path) -> dict[str, Any]: import pdfplumber out: dict[str, Any] = {"format": "pdf"} with pdfplumber.open(path) as pdf: out["n_pages"] = len(pdf.pages) out["page1_text"] = (pdf.pages[0].extract_text() or "")[:1500] return out def inspect_json(path: pathlib.Path) -> dict[str, Any]: d = json.loads(path.read_text()) out: dict[str, Any] = {"format": "json"} if isinstance(d, dict) and "features" in d: # ArcGIS feature set feats = d["features"] out["n_features"] = len(feats) out["fields"] = [f["name"] for f in d.get("fields", [])] out["sample_attributes"] = feats[0]["attributes"] if feats else None else: out["top_level_keys"] = list(d.keys())[:40] if isinstance(d, dict) else f"list[{len(d)}]" return out def inspect_zip(path: pathlib.Path) -> dict[str, Any]: import zipfile out: dict[str, Any] = {"format": "zip", "entries": []} with zipfile.ZipFile(path) as z: for info in z.infolist()[:30]: out["entries"].append({"name": info.filename, "bytes": info.file_size}) out["entry_count"] = len(z.namelist()) return out def inspect(path: pathlib.Path, fmt: str | None) -> dict[str, Any]: suffix = path.suffix.lower() if fmt in ("xlsx", "xls") or suffix in (".xlsx", ".xls", ".xlsm"): return inspect_excel(path) if fmt == "zip" or suffix == ".zip": return inspect_zip(path) if fmt == "csv" or suffix == ".csv": return inspect_csv(path) if fmt == "txt" or suffix == ".txt": return inspect_csv(path, sep="\t") if fmt == "pdf" or suffix == ".pdf": return inspect_pdf(path) if fmt == "json" or suffix == ".json": return inspect_json(path) return {"format": fmt or suffix, "note": "no inspector for format"} def render_findings(src: dict[str, Any], path: pathlib.Path, ins: dict[str, Any]) -> str: sid = src["source_id"] out: list[str] = [f"# {sid} — first-fetch findings\n"] out += [f"- publisher: {src.get('publisher')}", f"- url: {src.get('url')}", f"- format: {src.get('format')}", f"- local path: `{path.relative_to(REPO_ROOT)}`", ""] fmt = ins.get("format") if fmt == "excel": out.append(f"## Sheets ({len(ins.get('sheet_names', []))})") out += [f"- `{n}`" for n in ins.get("sheet_names", [])] out.append("") for name, sheet in ins.get("sheets", {}).items(): out.append(f"### `{name}` (header row {sheet.get('header_row_1based')})") if sheet.get("columns"): out += [f"- `{c}`" for c in sheet["columns"]] if sheet.get("first_rows"): out.append("No header detected in the first 15 rows; first rows:") out.append("```json\n" + json.dumps(sheet["first_rows"], indent=1) + "\n```") if sheet.get("sample_row"): out.append("\nSample first row:\n\n```json\n" + json.dumps(sheet["sample_row"], indent=2) + "\n```") out.append("") elif fmt == "csv": out.append(f"## Table — {ins.get('n_rows', '?'):,} rows × {ins.get('n_columns', '?')} columns\n") out += [f"- `{c}`" for c in ins.get("columns", [])] if ins.get("sample_row"): out.append("\nSample first row:\n\n```json\n" + json.dumps(ins["sample_row"], indent=2) + "\n```") elif fmt == "pdf": out.append(f"## PDF — {ins.get('n_pages')} pages\n\nPage 1 text:\n\n```\n{ins.get('page1_text')}\n```") elif fmt == "json": out.append("## JSON\n\n```json\n" + json.dumps({k: v for k, v in ins.items() if k != 'format'}, indent=2)[:4000] + "\n```") elif fmt == "zip": out.append(f"## Archive contents ({ins.get('entry_count')} entries)") out += [f"- `{e['name']}` ({e['bytes']:,} bytes)" for e in ins.get("entries", [])] return "\n".join(out) def discover_one(src: dict[str, Any], force: bool = False) -> None: sid = src["source_id"] if not is_resolved(src): print(f"[skip] {sid}: url is TBD", file=sys.stderr) return try: path = fetch(src, force=force) ins = inspect(path, src.get("format")) FINDINGS_DIR.mkdir(parents=True, exist_ok=True) (FINDINGS_DIR / f"{sid}.md").write_text(render_findings(src, path, ins)) print(f"[findings] -> pipeline_miami/findings/{sid}.md", file=sys.stderr) except Exception as e: # noqa: BLE001 print(f"[error] {sid}: {type(e).__name__}: {e}", file=sys.stderr) def main() -> None: p = argparse.ArgumentParser() p.add_argument("--source", default=None) p.add_argument("--all", action="store_true") p.add_argument("--force", action="store_true") args = p.parse_args() sources = load_sources() if args.source: match = [s for s in sources if s["source_id"] == args.source] if not match: print(f"no such source: {args.source}", file=sys.stderr) sys.exit(1) for s in match: discover_one(s, force=args.force) elif args.all: for s in sources: discover_one(s, force=args.force) else: for s in sources: fetched = any((DATA_DIR / s["source_id"]).glob("*/provenance.json")) if (DATA_DIR / s["source_id"]).exists() else False mark = "✓" if fetched else ("·" if is_resolved(s) else "?") print(f" {mark} {s['source_id']:48s} {s.get('publisher', '-')}") if __name__ == "__main__": main()