pipeline_miami/export_metric_index.py

"""Export the committed metric index the Miami methodology pages read:
docs/cities/miami/sources.yaml (which files back which metric, with their
provenance) and the served coverage of every metric (years, cells,
suppression, subgroups, percentiles). Read-only.

Run:  python -m pipeline_miami.export_metric_index
Writes data/cities/miami/reference/metric_index.json (commit it).
"""
from __future__ import annotations

import datetime as dt
import json
import re

import yaml

from pipeline_miami.verify import common

OUT = common.REPO_ROOT / "data" / "cities" / "miami" / "reference" / "metric_index.json"
FINDINGS = common.REPO_ROOT / "pipeline_miami" / "findings"


def _pattern(spec: str) -> re.Pattern[str]:
    """"enrollment_oct_g{PK..12}" → enrollment_oct_g\w+; "{a,b}" → (?:a|b); "*" → \w*; plain keys match exactly."""
    out = []
    for part in re.split(r"(\{[^}]*\}|\*)", spec):
        if not part:
            continue
        if part == "*":
            out.append(r"\w*")
        elif part.startswith("{"):
            inner = part[1:-1]
            alts = [a.strip() for a in inner.split(",")] if "," in inner and ".." not in inner else None
            out.append("(?:" + "|".join(re.escape(a) for a in alts) + ")" if alts else r"\w+")
        else:
            out.append(re.escape(part))
    return re.compile("^" + "".join(out) + "$")


def sources_index() -> tuple[dict, dict]:
    reg = yaml.safe_load(open(common.SOURCES_YAML))
    sources: dict[str, dict] = {}
    backed: list[tuple[re.Pattern[str], str]] = []
    for e in reg:
        sid = e["source_id"]
        prov = None
        d = common.SOURCES_DIR / sid
        if d.exists():
            days = sorted([p for p in d.iterdir() if p.is_dir() and p.name[:4].isdigit()], reverse=True)
            for day in days:
                pj = day / "provenance.json"
                if pj.exists():
                    prov = json.load(open(pj))
                    break
        years = e.get("years")
        sources[sid] = {
            "publisher": e.get("publisher"), "url": e.get("url"), "format": e.get("format"), "role": e.get("role"),
            "year": str(e["year"]) if e.get("year") is not None else None, "years": [str(y) for y in years] if years else None,
            "fetched": str(e.get("fetched")) if e.get("fetched") else None, "metrics_backed": [str(m) for m in (e.get("metrics_backed") or [])],
            "notes": (str(e.get("notes")).strip() or None) if e.get("notes") else None,
            "sha256": prov.get("sha256") if prov else None, "bytes": prov.get("bytes") if prov else None,
            "fetched_from": prov.get("fetched_from") if prov else None, "retrieved_at": prov.get("retrieved_at") if prov else None,
            "findings_md": f"pipeline_miami/findings/{sid}.md" if (FINDINGS / f"{sid}.md").exists() else None,
        }
        for m in sources[sid]["metrics_backed"]:
            backed.append((_pattern(m), sid))
    return sources, {"patterns": backed}


def coverage() -> dict[str, dict]:
    import psycopg

    out: dict[str, dict] = {}
    with psycopg.connect(common.served_db_dsn()) as conn:
        conn.read_only = True
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT metric_key,
                       array_agg(DISTINCT year ORDER BY year),
                       count(*), count(*) FILTER (WHERE suppressed), count(DISTINCT school_number),
                       array_agg(DISTINCT subgroup::text ORDER BY subgroup::text),
                       count(*) FILTER (WHERE comparison_group_percentile IS NOT NULL),
                       min(year) FILTER (WHERE value IS NOT NULL), max(year) FILTER (WHERE value IS NOT NULL)
                  FROM miami_school_year_metrics GROUP BY metric_key
                """
            )
            for key, years, cells, sup, schools, subgroups, pct, y0, y1 in cur:
                out[key] = {"years": years, "cells": cells, "suppressed": sup, "schools": schools, "subgroups": subgroups,
                            "with_peer_percentile": pct, "first_year": y0, "last_year": y1}
    return out


def main() -> None:
    sources, backed = sources_index()
    cov = coverage()
    keys = sorted(cov)
    metric_sources = {k: sorted({sid for pat, sid in backed["patterns"] if pat.match(k)}) for k in keys}
    doc = {"generated_at": dt.datetime.now(dt.timezone.utc).isoformat(), "sources": sources, "metric_sources": metric_sources, "coverage": cov}
    OUT.parent.mkdir(parents=True, exist_ok=True)
    OUT.write_text(json.dumps(doc, indent=1, default=str))
    unbacked = [k for k in keys if not metric_sources[k]]
    print(f"[metric_index] {len(sources)} sources, {len(keys)} served metrics, {sum(1 for k in keys if metric_sources[k])} with a registered source; unbacked: {unbacked}")


if __name__ == "__main__":
    main()