pipeline_miami/parse/names.py

"""School-name matching for sources that carry no school number (the FISH
level-of-service report). Matches a facility name to the 4-digit M-DCPS
school number using the MSID directory (active + closed) and the county GIS
layer, with exact normalized matching first and difflib similarity second.
Every match records its method and score; unmatched names are returned, not
guessed.
"""
from __future__ import annotations

import difflib
import json
import pathlib
import re

import pandas as pd

REPO_ROOT = pathlib.Path(__file__).resolve().parents[2]
SOURCES = REPO_ROOT / "data" / "cities" / "miami" / "sources"

_STOP = {"SCHOOL", "THE", "OF", "AND", "&", "CENTER", "CENTRE", "ACADEMY", "ACADEMIES"}
_REPL = [
    (r"\bSR\b", "SENIOR"), (r"\bJR\b", "JUNIOR"), (r"\bELEM\b", "ELEMENTARY"), (r"\bMID\b", "MIDDLE"),
    (r"\bK ?- ?8\b", "K8"), (r"\bK ?- ?12\b", "K12"), (r"\bPK ?- ?8\b", "K8"), (r"\bSHS\b", "SENIOR HIGH"),
    (r"\bMS\b", "MIDDLE"), (r"\bES\b", "ELEMENTARY"), (r"\bHS\b", "HIGH"), (r"\bCTR\b", "CENTER"),
    (r"\bINTL\b", "INTERNATIONAL"), (r"\bEDUC\b", "EDUCATION"), (r"\bPREP\b", "PREPARATORY"),
]


def _latest(source_id: str) -> pathlib.Path:
    d = SOURCES / source_id
    dated = sorted([p for p in d.iterdir() if p.is_dir()], reverse=True)
    for day in dated:
        for f in day.iterdir():
            if f.name != "provenance.json":
                return f
    raise FileNotFoundError(source_id)


def normalize(name: str) -> str:
    s = name.upper()
    s = s.replace("/", " ").replace("-", " ").replace(".", " ").replace(",", " ").replace("'", "")
    for pat, rep in _REPL:
        s = re.sub(pat, rep, s)
    s = re.sub(r"[^A-Z0-9 ]", " ", s)
    toks = [t for t in s.split() if t not in _STOP]
    return " ".join(toks)


def load_candidates() -> pd.DataFrame:
    """(school_number, name, source) from MSID (all statuses) and the GIS layer."""
    rows: list[dict] = []
    msid = pd.read_csv(_latest("fldoe_msid_all_schools"), sep="\t", dtype=str, encoding="latin1", low_memory=False)
    msid = msid[msid.DISTRICT.astype(str).str.strip() == "13"]
    for _, r in msid.iterrows():
        num = str(r.SCHOOL).strip().zfill(4)
        for col in ("SCHOOL_NAME_LONG", "SCHOOL_NAME_SHORT"):
            if isinstance(r[col], str) and r[col].strip():
                rows.append({"school_number": num, "name": r[col].strip(), "source": f"msid:{col}", "active": r.ACTIVITY_CODE == "A"})
    gis = json.loads(_latest("mdc_gis_schoolsite").read_text())
    for f in gis["features"]:
        a = f["attributes"]
        if a.get("ID") and a.get("NAME"):
            rows.append({"school_number": str(a["ID"]).strip().zfill(4), "name": str(a["NAME"]).strip(), "source": "gis", "active": True})
    df = pd.DataFrame(rows)
    df["norm"] = df.name.map(normalize)
    return df


class NameMatcher:
    def __init__(self, overrides: dict[str, str] | None = None):
        self.cands = load_candidates()
        self.exact = {}
        for _, r in self.cands.iterrows():
            self.exact.setdefault(r.norm, set()).add(r.school_number)
        self.norms = list(self.exact.keys())
        self.overrides = {normalize(k): v for k, v in (overrides or {}).items()}

    def match(self, facility_name: str) -> tuple[str | None, str, float]:
        n = normalize(facility_name)
        if n in self.overrides:
            return self.overrides[n], "override", 1.0
        if n in self.exact and len(self.exact[n]) == 1:
            return next(iter(self.exact[n])), "exact", 1.0
        if n in self.exact and len(self.exact[n]) > 1:
            # Same normalized name for several numbers (e.g. a school and its
            # annex). Prefer an active MSID row.
            act = self.cands[(self.cands.norm == n) & (self.cands.active)]
            if act.school_number.nunique() == 1:
                return act.school_number.iloc[0], "exact-active", 1.0
            return None, "ambiguous", 1.0
        best = difflib.get_close_matches(n, self.norms, n=3, cutoff=0.80)
        if not best:
            return None, "none", 0.0
        score = difflib.SequenceMatcher(None, n, best[0]).ratio()
        nums = self.exact[best[0]]
        if len(nums) == 1 and score >= 0.86:
            return next(iter(nums)), "fuzzy", round(score, 3)
        return None, "fuzzy-weak" if len(nums) == 1 else "fuzzy-ambiguous", round(score, 3)