pipeline_miami/discover.py

"""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/<source_id>/<date>/, 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 <source_id>
  python -m pipeline_miami.discover --all
  python -m pipeline_miami.discover --source <id> --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()