Upload 3 files

app.py

@@ -0,0 +1,649 @@
+"""
+Subdivision Cluster / FIPS-Validation pipeline — Hugging Face Space (Gradio).
+
+Faithful port of the Colab notebook
+`LatestUndeterminedZipCityCensusTract70_Threshold_95_Clusters.ipynb`.
+
+Colab-specific I/O replaced:
+  * Drive input folder      -> file upload (multiple .xlsx/.xls)
+  * google.colab.files      -> Gradio file outputs (written to a temp dir)
+  * Drive safety-list path  -> get_safety_list_path() (repo file OR private dataset)
+
+All processing logic (FIPS validation, RawFile snapshot, RapidFuzz clustering,
+wide tables, majority-vote replacement, column reorder, OnlyBlanks warnings,
+chunked CSV export, self-test) matches the notebook.
+"""
+
+import os
+import re
+import gc
+import math
+import tempfile
+from pathlib import Path
+from collections import defaultdict
+
+import numpy as np
+import pandas as pd
+import gradio as gr
+from rapidfuzz import fuzz
+
+
+# =====================================================================
+# REFERENCE FILE LOADER  (the Subdivision Program Safety List)
+# ---------------------------------------------------------------------
+# 1. A file committed in the Space repo (override name via SAFETY_LIST_FILE).
+#    Use ONLY if the Space is PRIVATE.
+# 2. A private HF *dataset* repo, downloaded with the HF_TOKEN secret. Set:
+#       SAFETY_LIST_REPO        e.g. "your-username/subdivision-safety-list"
+#       SAFETY_LIST_REPO_FILE   filename inside that dataset (optional)
+#       HF_TOKEN                read token, stored as a Space Secret
+# =====================================================================
+DEFAULT_SAFETY_LIST_NAME = "Subdivision_Program_Safety_List.xlsx"
+EXCEL_ROW_LIMIT = 1_048_575   # matches the notebook
+
+
+def get_safety_list_path() -> str:
+    local_name = os.environ.get("SAFETY_LIST_FILE", DEFAULT_SAFETY_LIST_NAME)
+    if Path(local_name).exists():
+        return local_name
+
+    repo_id = os.environ.get("SAFETY_LIST_REPO")
+    token = os.environ.get("HF_TOKEN")
+    if repo_id and token:
+        from huggingface_hub import hf_hub_download
+        filename = os.environ.get("SAFETY_LIST_REPO_FILE", local_name)
+        return hf_hub_download(
+            repo_id=repo_id, filename=filename, repo_type="dataset", token=token
+        )
+
+    raise FileNotFoundError(
+        "Safety list not found.\n"
+        f"  - Commit '{local_name}' into this Space (keep the Space private), OR\n"
+        "  - Set SAFETY_LIST_REPO + HF_TOKEN to pull it from a private dataset."
+    )
+
+
+_SAFETY_CACHE: dict | None = None
+
+
+def build_whitelist_from_path(path):
+    """Parse a safety-list file (xlsx or csv) into {pairs, cities, rows}."""
+    if str(path).lower().endswith(".csv"):
+        safety_df = pd.read_csv(path)
+    else:
+        safety_df = pd.read_excel(path)
+    acceptable_cols = [c for c in safety_df.columns if str(c).startswith("Acceptable FIPS")]
+
+    valid_pairs, valid_cities = set(), set()
+    for _, r in safety_df.iterrows():
+        c = _norm_city(r["City Name"])
+        if c is None:
+            continue
+        valid_cities.add(c)
+        for col in acceptable_cols:
+            f = _norm_fips(r[col])
+            if f:
+                valid_pairs.add(c + "||" + f)
+
+    return {"pairs": valid_pairs, "cities": valid_cities, "rows": len(safety_df)}
+
+
+def load_whitelist():
+    """Server-side safety list (repo file or private dataset). Cached per process."""
+    global _SAFETY_CACHE
+    if _SAFETY_CACHE is None:
+        _SAFETY_CACHE = build_whitelist_from_path(get_safety_list_path())
+    return _SAFETY_CACHE
+
+
+# =====================================================================
+# NORMALIZATION HELPERS  (identical to the notebook)
+# =====================================================================
+PLACEHOLDERS = {"UNDETERMINED CITY", "UNDETERMINED ZIP", "NULL", "NAN", "NONE", ""}
+
+
+def _norm_city(x):
+    if pd.isna(x):
+        return None
+    return str(x).strip().upper()
+
+
+def _norm_fips(x):
+    """5-digit FIPS string, or None. Handles 28071, 28071.0, '01067', '28071 - LAFAYETTE'."""
+    if pd.isna(x):
+        return None
+    m = re.search(r"\d+", str(x))
+    return str(int(m.group(0))).zfill(5) if m else None
+
+
+# =====================================================================
+# SELF-TEST  (startup health check — same logic as the notebook's cell)
+# =====================================================================
+def run_self_test() -> str:
+    path = get_safety_list_path()
+    st = pd.read_excel(path)
+    fc = [c for c in st.columns if str(c).startswith("Acceptable FIPS")]
+    pairs = set()
+    for _, r in st.iterrows():
+        c = _norm_city(r["City Name"])
+        if c is None:
+            continue
+        for col in fc:
+            f = _norm_fips(r[col])
+            if f:
+                pairs.add(c + "||" + f)
+
+    good_city = st.iloc[0]["City Name"]
+    good_fips = st.iloc[0]["Acceptable FIPS 1"]
+    test = pd.DataFrame({
+        "DDPPropertyCityFinal": [good_city, good_city],
+        "** REC FIPS-COUNTY": [good_fips, 99999],   # row 2 = wrong FIPS on purpose
+    })
+    cn = test["DDPPropertyCityFinal"].map(_norm_city)
+    fn = test["** REC FIPS-COUNTY"].map(_norm_fips)
+    key = cn.fillna("") + "||" + fn.fillna("")
+    ok = key.isin(pairs)
+    assert bool(ok.iloc[0]) and not bool(ok.iloc[1]), \
+        "SELF-TEST FAILED -- matching logic is not working as expected."
+    return f"SELF-TEST PASSED: known-good ({good_city}) kept, wrong FIPS flagged."
+
+
+# =====================================================================
+# STEP 1 — FIPS SAFETY-LIST VALIDATION
+# =====================================================================
+def fips_validate(df: pd.DataFrame, blank_cities_not_in_list: bool, log, wl=None):
+    if wl is None:
+        wl = load_whitelist()
+    valid_pairs, valid_cities = wl["pairs"], wl["cities"]
+
+    FIPS_COL = "** REC FIPS-COUNTY" if "** REC FIPS-COUNTY" in df.columns else "County"
+    CITY_COL, ZIP_COL, TRACT_COL = (
+        "DDPPropertyCityFinal", "DDPPropertyZIPFinal", "DDPRevisedCensusTract",
+    )
+
+    city_norm = df[CITY_COL].map(_norm_city)
+    fips_norm = df[FIPS_COL].map(_norm_fips)
+    row_key = city_norm.fillna("") + "||" + fips_norm.fillna("")
+
+    is_real_city = df[CITY_COL].notna() & (~city_norm.fillna("").isin(PLACEHOLDERS))
+    city_in_list = city_norm.isin(valid_cities)
+    pair_ok = row_key.isin(valid_pairs)
+
+    if blank_cities_not_in_list:
+        fails_mask = is_real_city & (~pair_ok)
+    else:
+        fails_mask = is_real_city & city_in_list & (~pair_ok)
+
+    n_real = int(is_real_city.sum())
+    n_inlist = int((is_real_city & city_in_list).sum())
+    n_valid = int((is_real_city & pair_ok).sum())
+    n_fail = int(fails_mask.sum())
+    log.append(
+        f"FIPS validation — list: {wl['rows']:,} rows | {len(valid_cities):,} cities | "
+        f"{len(valid_pairs):,} pairs"
+    )
+    log.append(
+        f"  real:{n_real:,} in-list:{n_inlist:,} valid-kept:{n_valid:,} blanked:{n_fail:,}"
+    )
+    if n_real > 0 and n_inlist == 0:
+        log.append("  !! 0 cities matched the list — likely a format/spelling mismatch.")
+
+    # Audit built BEFORE overwrite (insert position matches the notebook).
+    key_col = "RecordKey" if "RecordKey" in df.columns else None
+    base_cols = [c for c in [key_col, CITY_COL, FIPS_COL, ZIP_COL, TRACT_COL] if c]
+    audit = df.loc[fails_mask, base_cols].copy()
+    audit.insert(
+        len(base_cols) if key_col is None else 2,
+        "NormalizedFIPS", fips_norm[fails_mask].values,
+    )
+    audit["Reason"] = np.where(
+        city_in_list[fails_mask].values, "FIPS does not match city", "City not in safety list"
+    )
+    audit = audit.rename(
+        columns={CITY_COL: "OLD_City", ZIP_COL: "OLD_ZIP", TRACT_COL: "OLD_CensusTract"}
+    )
+
+    # Recent pandas reads all-numeric Excel columns as float64 and refuses to
+    # accept a string into them; Colab's older pandas silently upcast to object.
+    # Cast first so the placeholder write reproduces the notebook's result.
+    df[CITY_COL] = df[CITY_COL].astype(object)
+    df[ZIP_COL] = df[ZIP_COL].astype(object)
+    df.loc[fails_mask, CITY_COL] = "UNDETERMINED CITY"
+    df.loc[fails_mask, ZIP_COL] = "UNDETERMINED ZIP"
+    df.loc[fails_mask, TRACT_COL] = np.nan
+    return df, audit
+
+
+# =====================================================================
+# STEP 2 — RAPIDFUZZ CLUSTERING  (identical to the notebook)
+# =====================================================================
+def group_similar_strings_optimized(strings, threshold=95, prefix_len=3, max_len_diff=3):
+    cleaned, seen = [], set()
+    for s in strings:
+        if pd.isna(s):
+            continue
+        s = str(s).strip()
+        if s and s not in seen:
+            cleaned.append(s)
+            seen.add(s)
+
+    visited, groups = set(), []
+    buckets = defaultdict(list)
+    for s in cleaned:
+        buckets[s.upper()[:prefix_len]].append(s)
+
+    for bucket in buckets.values():
+        for i, base in enumerate(bucket):
+            if base in visited:
+                continue
+            group = [base]
+            visited.add(base)
+            base_upper = base.upper()
+            base_len = len(base_upper)
+            for j in range(i + 1, len(bucket)):
+                compare = bucket[j]
+                if compare in visited:
+                    continue
+                comp_upper = compare.upper()
+                if abs(base_len - len(comp_upper)) > max_len_diff:
+                    continue
+                if fuzz.ratio(base_upper, comp_upper) >= threshold:
+                    group.append(compare)
+                    visited.add(compare)
+            groups.append(group)
+    return groups
+
+
+def build_clusters(city_df, zip_df, census_tracts_df, cluster_threshold):
+    all_legal = pd.concat(
+        [
+            city_df[["County", "LegalSubDivision"]],
+            zip_df[["County", "LegalSubDivision"]],
+            census_tracts_df[["County", "LegalSubDivision"]],
+        ],
+        ignore_index=True,
+    ).dropna(subset=["County", "LegalSubDivision"])
+    all_legal["County"] = all_legal["County"].astype(str).str.strip()
+    all_legal["LegalSubDivision"] = all_legal["LegalSubDivision"].astype(str).str.strip()
+    all_legal = all_legal[(all_legal["County"] != "") & (all_legal["LegalSubDivision"] != "")]
+    all_legal = all_legal[["County", "LegalSubDivision"]].drop_duplicates()
+
+    groups = []
+    for county in all_legal["County"].unique():
+        strings = all_legal[all_legal["County"] == county]["LegalSubDivision"].tolist()
+        groups.extend(
+            group_similar_strings_optimized(
+                strings, threshold=cluster_threshold, prefix_len=3, max_len_diff=3
+            )
+        )
+
+    rows = []
+    for cid, group in enumerate(groups, start=1):
+        for sub in group:
+            rows.append({"Cluster_ID": cid, "LegalSubDivision": sub})
+    return pd.DataFrame(rows, columns=["Cluster_ID", "LegalSubDivision"])
+
+
+# =====================================================================
+# STEP 3 — WIDE TABLES + MAJORITY-VOTE "REPLACE WITH"
+# =====================================================================
+def build_cluster_wide_table(source_df, cluster_df, value_col, prefix):
+    lk = cluster_df[["LegalSubDivision", "Cluster_ID"]].drop_duplicates("LegalSubDivision")
+    cluster_map = dict(zip(lk["LegalSubDivision"], lk["Cluster_ID"]))
+
+    w = source_df[["LegalSubDivision", value_col]].copy()
+    w["LegalSubDivision"] = w["LegalSubDivision"].astype(str).str.strip()
+    w[value_col] = w[value_col].replace("", np.nan)
+    w["Cluster_ID"] = w["LegalSubDivision"].map(cluster_map)
+    w = w.dropna(subset=["Cluster_ID", value_col])
+    if w.empty:
+        return pd.DataFrame(columns=["Cluster_ID"])
+    w["Cluster_ID"] = w["Cluster_ID"].astype(int)
+    w[value_col] = w[value_col].astype(str).str.strip()
+    w = w[w[value_col] != ""]
+
+    counts = w.groupby(["Cluster_ID", value_col], observed=True).size().unstack(fill_value=0)
+    totals = counts.sum(axis=1)
+    pct = counts.div(totals.replace(0, np.nan), axis=0).mul(100).fillna(0)
+    counts.columns = [f"{prefix}{c}_Count" for c in counts.columns]
+    pct.columns = [f"{prefix}{c}_Percentage" for c in pct.columns]
+    wide = pd.concat([counts, pct], axis=1).reset_index()
+
+    # Match the notebook: Cluster_ID first, all other columns sorted.
+    other = sorted([c for c in wide.columns if c != "Cluster_ID"])
+    return wide[["Cluster_ID"] + other]
+
+
+def build_replace_with(wide_df, prefix_to_remove, threshold):
+    pct_cols = [c for c in wide_df.columns if c.endswith("_Percentage")]
+    if not pct_cols:
+        res = wide_df[["Cluster_ID"]].copy()
+        res["Replace with"] = None
+        return res.drop_duplicates(subset=["Cluster_ID"])
+    labels = np.array(
+        [c.replace(prefix_to_remove, "").replace("_Percentage", "") for c in pct_cols],
+        dtype=object,
+    )
+    values = wide_df[pct_cols].to_numpy(copy=False)
+    out = []
+    for row in values:
+        matched = labels[row >= threshold]
+        out.append(", ".join(matched) if len(matched) else None)
+    res = wide_df[["Cluster_ID"]].copy()
+    res["Replace with"] = out
+    return res.drop_duplicates(subset=["Cluster_ID"])
+
+
+def clean_census_tract_suffix(value):
+    if pd.isna(value):
+        return np.nan
+    try:
+        f = float(value)
+        return str(int(f)) if f == int(f) else str(f)
+    except ValueError:
+        return str(value)
+
+
+def clean_zip(value):
+    """Normalize ZIPs read from Excel as numbers: drop the float '.0' artifact and
+    zero-pad to 5 digits (recovering leading zeros lost on numeric import, e.g.
+    7221.0 -> 07221). Handles comma-joined majority-vote results. ZIP+4 left as-is."""
+    if pd.isna(value):
+        return np.nan
+    out = []
+    for part in str(value).split(","):
+        p = part.strip()
+        if p.endswith(".0"):
+            p = p[:-2]
+        digits = re.sub(r"\D", "", p)
+        if not digits:
+            continue
+        out.append(digits.zfill(5) if len(digits) <= 5 else digits)
+    return ", ".join(out) if out else np.nan
+
+
+# =====================================================================
+# CHUNKED CSV WRITER  (faithful port of save_csv_in_parts)
+# =====================================================================
+def save_csv_in_parts(df, base_filename, out_dir, row_limit=1_000_000):
+    base_name = base_filename.replace(".csv", "")
+    created = []
+    total = len(df)
+    num_parts = math.ceil(total / row_limit)
+    for i in range(num_parts):
+        start = i * row_limit
+        end = min((i + 1) * row_limit, total)
+        part = f"{base_name}_part_{i + 1}.csv" if num_parts > 1 else base_filename
+        p = out_dir / part
+        df.iloc[start:end].to_csv(p, index=False)
+        created.append(str(p))
+        gc.collect()
+    return created
+
+
+# =====================================================================
+# MAIN PIPELINE
+# =====================================================================
+def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
+                 filter_dominant_county, cluster_threshold, replace_threshold):
+    log = []
+    if not uploaded_files:
+        return "Please upload at least one Excel file.", []
+
+    out_dir = Path(tempfile.mkdtemp(prefix="subdiv_"))
+    outputs = []
+
+    # ---- Resolve the reference (safety) list -------------------------
+    # Default: server-side list (repo file or private dataset). If the user
+    # uploads an override on the sidebar, use it for THIS run only.
+    try:
+        if reference_file is not None:
+            wl = build_whitelist_from_path(reference_file.name)
+            log.append(f"Using UPLOADED reference list: {Path(reference_file.name).name}")
+        else:
+            wl = load_whitelist()
+            log.append("Using server-side safety list.")
+    except FileNotFoundError as e:
+        return "\n".join(log) + f"\n\nREFERENCE FILE ERROR:\n{e}", []
+    except Exception as e:
+        return "\n".join(log) + f"\n\nCould not parse reference list: {e}", []
+
+    # ---- Load & merge uploads ----------------------------------------
+    dfs = []
+    for f in uploaded_files:
+        try:
+            dfs.append(pd.read_excel(f.name))
+            log.append(f"Loaded {Path(f.name).name}")
+        except Exception as e:
+            log.append(f"ERROR loading {Path(f.name).name}: {e}")
+    if not dfs:
+        return "\n".join(log) + "\nNo files could be read.", []
+    df = pd.concat(dfs, ignore_index=True)
+    log.append(f"Merged shape: {df.shape}")
+
+    # ---- FIPS validation (uses the reference list) -------------------
+    try:
+        df, audit = fips_validate(df, blank_cities_not_in_list, log, wl=wl)
+    except FileNotFoundError as e:
+        return "\n".join(log) + f"\n\nREFERENCE FILE ERROR:\n{e}", []
+
+    # ---- Download verification files (audit + RawFile_FIPSValidated) -
+    if len(audit) > 0:
+        ap = out_dir / "FIPS_Validation_Audit_ChangedRecords.xlsx"
+        audit.to_excel(ap, index=False)
+        outputs.append(str(ap))
+
+    if len(df) <= EXCEL_ROW_LIMIT:
+        rp = out_dir / "RawFile_FIPSValidated.xlsx"
+        df.to_excel(rp, index=False)
+    else:
+        rp = out_dir / "RawFile_FIPSValidated.csv"
+        df.to_csv(rp, index=False)
+    outputs.append(str(rp))
+
+    # ---- UNDETERMINED/NULL -> NaN ------------------------------------
+    for col in ["DDPPropertyCityFinal", "DDPPropertyZIPFinal",
+                "DDPRevisedCensusTract", "LegalSubDivision"]:
+        if col in df.columns:
+            df[col] = df[col].replace(["UNDETERMINED CITY", "UNDETERMINED ZIP", "NULL"], np.nan)
+
+    # ---- Rename FIPS col -> County -----------------------------------
+    if "** REC FIPS-COUNTY" in df.columns:
+        df = df.rename(columns={"** REC FIPS-COUNTY": "County"})
+
+    # ---- Optional: filter to single most-dominant county -------------
+    if filter_dominant_county:
+        dominant = df["County"].value_counts().idxmax()
+        df = df[df["County"] == dominant].copy()
+        log.append(f"Filtered to dominant county: {dominant} -> {df.shape}")
+
+    # ---- Source frames -----------------------------------------------
+    city_df = df.loc[df["DDPPropertyCityFinal"].notna()].copy()
+    zip_df = df.loc[df["DDPPropertyZIPFinal"].notna()].copy()
+    census_tracts_df = df.loc[df["DDPRevisedCensusTract"].notna()].copy()
+    original = df.loc[df["LegalSubDivision"].notna()].copy()
+
+    # ---- Cluster ------------------------------------------------------
+    cluster_df = build_clusters(city_df, zip_df, census_tracts_df, cluster_threshold)
+    log.append(f"Clusters: {cluster_df['Cluster_ID'].nunique():,}")
+
+    # ---- Replace-with (majority vote) --------------------------------
+    zip_w = build_replace_with(
+        build_cluster_wide_table(zip_df, cluster_df, "DDPPropertyZIPFinal", "ZIP_"),
+        "ZIP_", replace_threshold).rename(columns={"Replace with": "Replace with Zip Codes"})
+    city_w = build_replace_with(
+        build_cluster_wide_table(city_df, cluster_df, "DDPPropertyCityFinal", "City_"),
+        "City_", replace_threshold).rename(columns={"Replace with": "Replace with Cities"})
+    cen_w = build_replace_with(
+        build_cluster_wide_table(census_tracts_df, cluster_df, "DDPRevisedCensusTract", "CensusTract_"),
+        "CensusTract_", replace_threshold).rename(columns={"Replace with": "Replace with Census Tracts"})
+
+    # ---- Map back onto original --------------------------------------
+    lk = cluster_df[["LegalSubDivision", "Cluster_ID"]].drop_duplicates("LegalSubDivision")
+    cmap = dict(zip(lk["LegalSubDivision"], lk["Cluster_ID"]))
+    original_final = original.copy()
+    original_final["Cluster_ID"] = original_final["LegalSubDivision"].map(cmap)
+    original_final["Replace with Zip Codes"] = original_final["Cluster_ID"].map(
+        dict(zip(zip_w["Cluster_ID"], zip_w["Replace with Zip Codes"])))
+    original_final["Replace with Cities"] = original_final["Cluster_ID"].map(
+        dict(zip(city_w["Cluster_ID"], city_w["Replace with Cities"])))
+    original_final["Replace with Census Tracts"] = original_final["Cluster_ID"].map(
+        dict(zip(cen_w["Cluster_ID"], cen_w["Replace with Census Tracts"])))
+
+    # Normalize ZIPs (drop float '.0', restore 5-digit zero-padding).
+    original_final["Replace with Zip Codes"] = (
+        original_final["Replace with Zip Codes"].apply(clean_zip))
+
+    # ---- Reorder columns (desired_leading_columns) -------------------
+    desired_leading = [
+        "RecordKey", "ComesFromMtg2", "IsFlagged", "Cluster_ID", "LegalSubDivision",
+        "County", "Replace with Zip Codes", "Replace with Cities", "Replace with Census Tracts",
+        "DDPPropertyZIPFinal", "DDPPropertyCityFinal", "DDPRevisedCensusTract",
+        "LotBlockSubdivisionTract", "DDPRevisedLatitude", "DDPRevisedLongitude",
+    ]
+    existing = original_final.columns.tolist()
+    lead = [c for c in desired_leading if c in existing]
+    other = [c for c in existing if c not in lead]
+    original_final = original_final[lead + other]
+
+    original_final_output = original_final.copy()   # full output snapshot
+
+    # ---- Split into final cities / zips / census tracts --------------
+    keep = ["RecordKey", "ComesFromMtg2", "IsFlagged",
+            "Replace with Zip Codes", "Replace with Cities", "Replace with Census Tracts",
+            "DDPPropertyZIPFinal", "DDPPropertyCityFinal", "DDPRevisedCensusTract"]
+    keep = [c for c in keep if c in original_final.columns]
+    of = original_final[keep].rename(columns={
+        "DDPPropertyZIPFinal": "DDPPropertyZIPFinalToRemove",
+        "DDPPropertyCityFinal": "DDPPropertyCityFinalToRemove",
+        "DDPRevisedCensusTract": "DDPRevisedCensusTractToRemove",
+        "Replace with Zip Codes": "DDPPropertyZIPFinal",
+        "Replace with Cities": "DDPPropertyCityFinal",
+        "Replace with Census Tracts": "DDPRevisedCensusTract",
+    })
+    of["DDPRevisedCensusTract"] = of["DDPRevisedCensusTract"].apply(clean_census_tract_suffix)
+
+    idcols = [c for c in ["RecordKey", "ComesFromMtg2", "IsFlagged"] if c in of.columns]
+    final_cities = of[idcols + ["DDPPropertyCityFinalToRemove", "DDPPropertyCityFinal"]].copy()
+    final_zips = of[idcols + ["DDPPropertyZIPFinalToRemove", "DDPPropertyZIPFinal"]].copy()
+    final_ct = of[idcols + ["DDPRevisedCensusTractToRemove", "DDPRevisedCensusTract"]].copy()
+
+    # keep only rows whose ORIGINAL value was blank (the ones we filled)
+    final_cities = final_cities[final_cities["DDPPropertyCityFinalToRemove"].isna()].copy()
+    final_zips = final_zips[final_zips["DDPPropertyZIPFinalToRemove"].isna()].copy()
+    final_ct = final_ct[final_ct["DDPRevisedCensusTractToRemove"].isna()].copy()
+
+    final_cities = final_cities.drop(columns=["DDPPropertyCityFinalToRemove"])
+    final_zips = final_zips.drop(columns=["DDPPropertyZIPFinalToRemove"])
+    final_ct = final_ct.drop(columns=["DDPRevisedCensusTractToRemove"])
+
+    # ---- OnlyBlanks warnings (column entirely null) ------------------
+    def only_blanks_warning(d, col, prefix):
+        if col in d.columns and d[col].isnull().all():
+            p = out_dir / f"OnlyBlanks_{prefix}.xlsx"
+            d.to_excel(p, index=False)
+            outputs.append(str(p))
+            log.append(f"WARNING: {prefix} has only blanks in '{col}' -> {p.name}")
+
+    only_blanks_warning(final_cities, "DDPPropertyCityFinal", "final_cities")
+    only_blanks_warning(final_zips, "DDPPropertyZIPFinal", "final_zips")
+    only_blanks_warning(final_ct, "DDPRevisedCensusTract", "final_census_tracts")
+
+    # ---- Drop NaNs so only program assignments remain ----------------
+    final_cities = final_cities[final_cities["DDPPropertyCityFinal"].notna()].copy()
+    final_zips = final_zips[final_zips["DDPPropertyZIPFinal"].notna()].copy()
+    final_ct = final_ct[final_ct["DDPRevisedCensusTract"].notna()].copy()
+
+    # ---- Write the per-category + cluster outputs --------------------
+    for d, name in [(final_cities, "final_cities.xlsx"),
+                    (final_zips, "final_zips.xlsx"),
+                    (final_ct, "final_census_tracts.xlsx"),
+                    (cluster_df, "cluster_df.xlsx")]:
+        p = out_dir / name
+        d.to_excel(p, index=False)
+        outputs.append(str(p))
+
+    # ---- Full output CSV (chunked at 1,000,000 rows) -----------------
+    csv_name = f"OutputFile_{replace_threshold}%Threshold_{cluster_threshold}%ClustersRapidFuzz.csv"
+    outputs.extend(save_csv_in_parts(original_final_output, csv_name, out_dir, row_limit=1_000_000))
+
+    log.append(f"Done. {len(outputs)} files ready.")
+    gc.collect()
+    return "\n".join(log), outputs
+
+
+# =====================================================================
+# STARTUP HEALTH CHECK
+# =====================================================================
+def _startup_health_check():
+    if os.environ.get("SELF_TEST_ON_STARTUP", "1") != "1":
+        return
+    try:
+        print("[startup] " + run_self_test())
+    except FileNotFoundError as e:
+        # Reference file not configured yet — let the UI load so the user can fix it.
+        print(f"[startup] Self-test skipped — safety list unavailable:\n{e}")
+    # AssertionError intentionally propagates: a failed self-test fails the boot.
+
+
+_startup_health_check()
+
+
+# =====================================================================
+# GRADIO UI  (sidebar "Upload Files" layout)
+# =====================================================================
+CSS = """
+#sidebar {background: var(--block-background-fill); border-right: 1px solid var(--border-color-primary);
+          padding: 18px 16px; min-width: 270px;}
+#sidebar h3 {margin: 0 0 14px 0; font-size: 1.15rem; font-weight: 700;}
+.step-label {font-size: .82rem; color: var(--body-text-color-subdued); margin: 14px 0 6px 0;}
+#main-title {font-size: 2.2rem; font-weight: 800; margin: 6px 0 18px 0;}
+.info-banner {background: #e8f1fe; color: #1d4ed8; border-radius: 10px;
+              padding: 16px 18px; font-size: 1rem; margin-bottom: 18px;}
+"""
+
+with gr.Blocks() as demo:
+    gr.HTML(f"<style>{CSS}</style>")
+    with gr.Row():
+        # -------- LEFT SIDEBAR: uploads ---------------------------------
+        with gr.Column(scale=1, elem_id="sidebar", min_width=270):
+            gr.Markdown("### Upload Files")
+
+            gr.HTML('<div class="step-label">1. Upload Excel Files</div>')
+            files_in = gr.File(file_count="multiple", file_types=[".xlsx", ".xls"],
+                               label="200MB per file • XLSX", show_label=True)
+
+            gr.HTML('<div class="step-label">2. Upload Reference File (Optional)</div>')
+            reference_in = gr.File(file_count="single", file_types=[".xlsx", ".csv"],
+                                   label="200MB per file • XLSX, CSV", show_label=True)
+
+            with gr.Accordion("Options", open=False):
+                blank_flag = gr.Checkbox(
+                    value=True, label="Strict: blank cities not in the safety list")
+                dominant_flag = gr.Checkbox(
+                    value=True, label="Filter to single most-dominant county")
+                clu_thr = gr.Slider(70, 100, value=95, step=1, label="Cluster threshold (%)")
+                rep_thr = gr.Slider(50, 100, value=70, step=1, label="Replace-with threshold (%)")
+
+            run_btn = gr.Button("Run pipeline", variant="primary")
+
+        # -------- MAIN AREA: title, info, results -----------------------
+        with gr.Column(scale=4):
+            gr.HTML('<div id="main-title">Subdivision Cluster &amp; FIPS Validation</div>')
+            gr.HTML(
+                '<div class="info-banner">Please upload your Excel file(s) using the '
+                'sidebar, then click <b>Run pipeline</b>. The reference safety list is '
+                'applied automatically; upload one on the sidebar only to override it for '
+                'a single run.</div>'
+            )
+            status = gr.Textbox(label="Log", lines=14)
+            files_out = gr.File(label="Output files", file_count="multiple")
+
+    run_btn.click(
+        run_pipeline,
+        inputs=[files_in, reference_in, blank_flag, dominant_flag, clu_thr, rep_thr],
+        outputs=[status, files_out],
+    )
+
+if __name__ == "__main__":
+    demo.launch()

app_simple_ui.py

@@ -0,0 +1,629 @@
+"""
+Subdivision Cluster / FIPS-Validation pipeline — Hugging Face Space (Gradio).
+
+Faithful port of the Colab notebook
+`LatestUndeterminedZipCityCensusTract70_Threshold_95_Clusters.ipynb`.
+
+Colab-specific I/O replaced:
+  * Drive input folder      -> file upload (multiple .xlsx/.xls)
+  * google.colab.files      -> Gradio file outputs (written to a temp dir)
+  * Drive safety-list path  -> get_safety_list_path() (repo file OR private dataset)
+
+All processing logic (FIPS validation, RawFile snapshot, RapidFuzz clustering,
+wide tables, majority-vote replacement, column reorder, OnlyBlanks warnings,
+chunked CSV export, self-test) matches the notebook.
+"""
+
+import os
+import re
+import gc
+import math
+import tempfile
+from pathlib import Path
+from collections import defaultdict
+
+import numpy as np
+import pandas as pd
+import gradio as gr
+from rapidfuzz import fuzz
+
+
+# =====================================================================
+# REFERENCE FILE LOADER  (the Subdivision Program Safety List)
+# ---------------------------------------------------------------------
+# 1. A file committed in the Space repo (override name via SAFETY_LIST_FILE).
+#    Use ONLY if the Space is PRIVATE.
+# 2. A private HF *dataset* repo, downloaded with the HF_TOKEN secret. Set:
+#       SAFETY_LIST_REPO        e.g. "your-username/subdivision-safety-list"
+#       SAFETY_LIST_REPO_FILE   filename inside that dataset (optional)
+#       HF_TOKEN                read token, stored as a Space Secret
+# =====================================================================
+DEFAULT_SAFETY_LIST_NAME = "Subdivision_Program_Safety_List.xlsx"
+EXCEL_ROW_LIMIT = 1_048_575   # matches the notebook
+
+
+def get_safety_list_path() -> str:
+    local_name = os.environ.get("SAFETY_LIST_FILE", DEFAULT_SAFETY_LIST_NAME)
+    if Path(local_name).exists():
+        return local_name
+
+    repo_id = os.environ.get("SAFETY_LIST_REPO")
+    token = os.environ.get("HF_TOKEN")
+    if repo_id and token:
+        from huggingface_hub import hf_hub_download
+        filename = os.environ.get("SAFETY_LIST_REPO_FILE", local_name)
+        return hf_hub_download(
+            repo_id=repo_id, filename=filename, repo_type="dataset", token=token
+        )
+
+    raise FileNotFoundError(
+        "Safety list not found.\n"
+        f"  - Commit '{local_name}' into this Space (keep the Space private), OR\n"
+        "  - Set SAFETY_LIST_REPO + HF_TOKEN to pull it from a private dataset."
+    )
+
+
+_SAFETY_CACHE: dict | None = None
+
+
+def build_whitelist_from_path(path):
+    """Parse a safety-list file (xlsx or csv) into {pairs, cities, rows}."""
+    if str(path).lower().endswith(".csv"):
+        safety_df = pd.read_csv(path)
+    else:
+        safety_df = pd.read_excel(path)
+    acceptable_cols = [c for c in safety_df.columns if str(c).startswith("Acceptable FIPS")]
+
+    valid_pairs, valid_cities = set(), set()
+    for _, r in safety_df.iterrows():
+        c = _norm_city(r["City Name"])
+        if c is None:
+            continue
+        valid_cities.add(c)
+        for col in acceptable_cols:
+            f = _norm_fips(r[col])
+            if f:
+                valid_pairs.add(c + "||" + f)
+
+    return {"pairs": valid_pairs, "cities": valid_cities, "rows": len(safety_df)}
+
+
+def load_whitelist():
+    """Server-side safety list (repo file or private dataset). Cached per process."""
+    global _SAFETY_CACHE
+    if _SAFETY_CACHE is None:
+        _SAFETY_CACHE = build_whitelist_from_path(get_safety_list_path())
+    return _SAFETY_CACHE
+
+
+# =====================================================================
+# NORMALIZATION HELPERS  (identical to the notebook)
+# =====================================================================
+PLACEHOLDERS = {"UNDETERMINED CITY", "UNDETERMINED ZIP", "NULL", "NAN", "NONE", ""}
+
+
+def _norm_city(x):
+    if pd.isna(x):
+        return None
+    return str(x).strip().upper()
+
+
+def _norm_fips(x):
+    """5-digit FIPS string, or None. Handles 28071, 28071.0, '01067', '28071 - LAFAYETTE'."""
+    if pd.isna(x):
+        return None
+    m = re.search(r"\d+", str(x))
+    return str(int(m.group(0))).zfill(5) if m else None
+
+
+# =====================================================================
+# SELF-TEST  (startup health check — same logic as the notebook's cell)
+# =====================================================================
+def run_self_test() -> str:
+    path = get_safety_list_path()
+    st = pd.read_excel(path)
+    fc = [c for c in st.columns if str(c).startswith("Acceptable FIPS")]
+    pairs = set()
+    for _, r in st.iterrows():
+        c = _norm_city(r["City Name"])
+        if c is None:
+            continue
+        for col in fc:
+            f = _norm_fips(r[col])
+            if f:
+                pairs.add(c + "||" + f)
+
+    good_city = st.iloc[0]["City Name"]
+    good_fips = st.iloc[0]["Acceptable FIPS 1"]
+    test = pd.DataFrame({
+        "DDPPropertyCityFinal": [good_city, good_city],
+        "** REC FIPS-COUNTY": [good_fips, 99999],   # row 2 = wrong FIPS on purpose
+    })
+    cn = test["DDPPropertyCityFinal"].map(_norm_city)
+    fn = test["** REC FIPS-COUNTY"].map(_norm_fips)
+    key = cn.fillna("") + "||" + fn.fillna("")
+    ok = key.isin(pairs)
+    assert bool(ok.iloc[0]) and not bool(ok.iloc[1]), \
+        "SELF-TEST FAILED -- matching logic is not working as expected."
+    return f"SELF-TEST PASSED: known-good ({good_city}) kept, wrong FIPS flagged."
+
+
+# =====================================================================
+# STEP 1 — FIPS SAFETY-LIST VALIDATION
+# =====================================================================
+def fips_validate(df: pd.DataFrame, blank_cities_not_in_list: bool, log, wl=None):
+    if wl is None:
+        wl = load_whitelist()
+    valid_pairs, valid_cities = wl["pairs"], wl["cities"]
+
+    FIPS_COL = "** REC FIPS-COUNTY" if "** REC FIPS-COUNTY" in df.columns else "County"
+    CITY_COL, ZIP_COL, TRACT_COL = (
+        "DDPPropertyCityFinal", "DDPPropertyZIPFinal", "DDPRevisedCensusTract",
+    )
+
+    city_norm = df[CITY_COL].map(_norm_city)
+    fips_norm = df[FIPS_COL].map(_norm_fips)
+    row_key = city_norm.fillna("") + "||" + fips_norm.fillna("")
+
+    is_real_city = df[CITY_COL].notna() & (~city_norm.fillna("").isin(PLACEHOLDERS))
+    city_in_list = city_norm.isin(valid_cities)
+    pair_ok = row_key.isin(valid_pairs)
+
+    if blank_cities_not_in_list:
+        fails_mask = is_real_city & (~pair_ok)
+    else:
+        fails_mask = is_real_city & city_in_list & (~pair_ok)
+
+    n_real = int(is_real_city.sum())
+    n_inlist = int((is_real_city & city_in_list).sum())
+    n_valid = int((is_real_city & pair_ok).sum())
+    n_fail = int(fails_mask.sum())
+    log.append(
+        f"FIPS validation — list: {wl['rows']:,} rows | {len(valid_cities):,} cities | "
+        f"{len(valid_pairs):,} pairs"
+    )
+    log.append(
+        f"  real:{n_real:,} in-list:{n_inlist:,} valid-kept:{n_valid:,} blanked:{n_fail:,}"
+    )
+    if n_real > 0 and n_inlist == 0:
+        log.append("  !! 0 cities matched the list — likely a format/spelling mismatch.")
+
+    # Audit built BEFORE overwrite (insert position matches the notebook).
+    key_col = "RecordKey" if "RecordKey" in df.columns else None
+    base_cols = [c for c in [key_col, CITY_COL, FIPS_COL, ZIP_COL, TRACT_COL] if c]
+    audit = df.loc[fails_mask, base_cols].copy()
+    audit.insert(
+        len(base_cols) if key_col is None else 2,
+        "NormalizedFIPS", fips_norm[fails_mask].values,
+    )
+    audit["Reason"] = np.where(
+        city_in_list[fails_mask].values, "FIPS does not match city", "City not in safety list"
+    )
+    audit = audit.rename(
+        columns={CITY_COL: "OLD_City", ZIP_COL: "OLD_ZIP", TRACT_COL: "OLD_CensusTract"}
+    )
+
+    # Recent pandas reads all-numeric Excel columns as float64 and refuses to
+    # accept a string into them; Colab's older pandas silently upcast to object.
+    # Cast first so the placeholder write reproduces the notebook's result.
+    df[CITY_COL] = df[CITY_COL].astype(object)
+    df[ZIP_COL] = df[ZIP_COL].astype(object)
+    df.loc[fails_mask, CITY_COL] = "UNDETERMINED CITY"
+    df.loc[fails_mask, ZIP_COL] = "UNDETERMINED ZIP"
+    df.loc[fails_mask, TRACT_COL] = np.nan
+    return df, audit
+
+
+# =====================================================================
+# STEP 2 — RAPIDFUZZ CLUSTERING  (identical to the notebook)
+# =====================================================================
+def group_similar_strings_optimized(strings, threshold=95, prefix_len=3, max_len_diff=3):
+    cleaned, seen = [], set()
+    for s in strings:
+        if pd.isna(s):
+            continue
+        s = str(s).strip()
+        if s and s not in seen:
+            cleaned.append(s)
+            seen.add(s)
+
+    visited, groups = set(), []
+    buckets = defaultdict(list)
+    for s in cleaned:
+        buckets[s.upper()[:prefix_len]].append(s)
+
+    for bucket in buckets.values():
+        for i, base in enumerate(bucket):
+            if base in visited:
+                continue
+            group = [base]
+            visited.add(base)
+            base_upper = base.upper()
+            base_len = len(base_upper)
+            for j in range(i + 1, len(bucket)):
+                compare = bucket[j]
+                if compare in visited:
+                    continue
+                comp_upper = compare.upper()
+                if abs(base_len - len(comp_upper)) > max_len_diff:
+                    continue
+                if fuzz.ratio(base_upper, comp_upper) >= threshold:
+                    group.append(compare)
+                    visited.add(compare)
+            groups.append(group)
+    return groups
+
+
+def build_clusters(city_df, zip_df, census_tracts_df, cluster_threshold):
+    all_legal = pd.concat(
+        [
+            city_df[["County", "LegalSubDivision"]],
+            zip_df[["County", "LegalSubDivision"]],
+            census_tracts_df[["County", "LegalSubDivision"]],
+        ],
+        ignore_index=True,
+    ).dropna(subset=["County", "LegalSubDivision"])
+    all_legal["County"] = all_legal["County"].astype(str).str.strip()
+    all_legal["LegalSubDivision"] = all_legal["LegalSubDivision"].astype(str).str.strip()
+    all_legal = all_legal[(all_legal["County"] != "") & (all_legal["LegalSubDivision"] != "")]
+    all_legal = all_legal[["County", "LegalSubDivision"]].drop_duplicates()
+
+    groups = []
+    for county in all_legal["County"].unique():
+        strings = all_legal[all_legal["County"] == county]["LegalSubDivision"].tolist()
+        groups.extend(
+            group_similar_strings_optimized(
+                strings, threshold=cluster_threshold, prefix_len=3, max_len_diff=3
+            )
+        )
+
+    rows = []
+    for cid, group in enumerate(groups, start=1):
+        for sub in group:
+            rows.append({"Cluster_ID": cid, "LegalSubDivision": sub})
+    return pd.DataFrame(rows, columns=["Cluster_ID", "LegalSubDivision"])
+
+
+# =====================================================================
+# STEP 3 — WIDE TABLES + MAJORITY-VOTE "REPLACE WITH"
+# =====================================================================
+def build_cluster_wide_table(source_df, cluster_df, value_col, prefix):
+    lk = cluster_df[["LegalSubDivision", "Cluster_ID"]].drop_duplicates("LegalSubDivision")
+    cluster_map = dict(zip(lk["LegalSubDivision"], lk["Cluster_ID"]))
+
+    w = source_df[["LegalSubDivision", value_col]].copy()
+    w["LegalSubDivision"] = w["LegalSubDivision"].astype(str).str.strip()
+    w[value_col] = w[value_col].replace("", np.nan)
+    w["Cluster_ID"] = w["LegalSubDivision"].map(cluster_map)
+    w = w.dropna(subset=["Cluster_ID", value_col])
+    if w.empty:
+        return pd.DataFrame(columns=["Cluster_ID"])
+    w["Cluster_ID"] = w["Cluster_ID"].astype(int)
+    w[value_col] = w[value_col].astype(str).str.strip()
+    w = w[w[value_col] != ""]
+
+    counts = w.groupby(["Cluster_ID", value_col], observed=True).size().unstack(fill_value=0)
+    totals = counts.sum(axis=1)
+    pct = counts.div(totals.replace(0, np.nan), axis=0).mul(100).fillna(0)
+    counts.columns = [f"{prefix}{c}_Count" for c in counts.columns]
+    pct.columns = [f"{prefix}{c}_Percentage" for c in pct.columns]
+    wide = pd.concat([counts, pct], axis=1).reset_index()
+
+    # Match the notebook: Cluster_ID first, all other columns sorted.
+    other = sorted([c for c in wide.columns if c != "Cluster_ID"])
+    return wide[["Cluster_ID"] + other]
+
+
+def build_replace_with(wide_df, prefix_to_remove, threshold):
+    pct_cols = [c for c in wide_df.columns if c.endswith("_Percentage")]
+    if not pct_cols:
+        res = wide_df[["Cluster_ID"]].copy()
+        res["Replace with"] = None
+        return res.drop_duplicates(subset=["Cluster_ID"])
+    labels = np.array(
+        [c.replace(prefix_to_remove, "").replace("_Percentage", "") for c in pct_cols],
+        dtype=object,
+    )
+    values = wide_df[pct_cols].to_numpy(copy=False)
+    out = []
+    for row in values:
+        matched = labels[row >= threshold]
+        out.append(", ".join(matched) if len(matched) else None)
+    res = wide_df[["Cluster_ID"]].copy()
+    res["Replace with"] = out
+    return res.drop_duplicates(subset=["Cluster_ID"])
+
+
+def clean_census_tract_suffix(value):
+    if pd.isna(value):
+        return np.nan
+    try:
+        f = float(value)
+        return str(int(f)) if f == int(f) else str(f)
+    except ValueError:
+        return str(value)
+
+
+def clean_zip(value):
+    """Normalize ZIPs read from Excel as numbers: drop the float '.0' artifact and
+    zero-pad to 5 digits (recovering leading zeros lost on numeric import, e.g.
+    7221.0 -> 07221). Handles comma-joined majority-vote results. ZIP+4 left as-is."""
+    if pd.isna(value):
+        return np.nan
+    out = []
+    for part in str(value).split(","):
+        p = part.strip()
+        if p.endswith(".0"):
+            p = p[:-2]
+        digits = re.sub(r"\D", "", p)
+        if not digits:
+            continue
+        out.append(digits.zfill(5) if len(digits) <= 5 else digits)
+    return ", ".join(out) if out else np.nan
+
+
+# =====================================================================
+# CHUNKED CSV WRITER  (faithful port of save_csv_in_parts)
+# =====================================================================
+def save_csv_in_parts(df, base_filename, out_dir, row_limit=1_000_000):
+    base_name = base_filename.replace(".csv", "")
+    created = []
+    total = len(df)
+    num_parts = math.ceil(total / row_limit)
+    for i in range(num_parts):
+        start = i * row_limit
+        end = min((i + 1) * row_limit, total)
+        part = f"{base_name}_part_{i + 1}.csv" if num_parts > 1 else base_filename
+        p = out_dir / part
+        df.iloc[start:end].to_csv(p, index=False)
+        created.append(str(p))
+        gc.collect()
+    return created
+
+
+# =====================================================================
+# MAIN PIPELINE
+# =====================================================================
+def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
+                 filter_dominant_county, cluster_threshold, replace_threshold):
+    log = []
+    if not uploaded_files:
+        return "Please upload at least one Excel file.", []
+
+    out_dir = Path(tempfile.mkdtemp(prefix="subdiv_"))
+    outputs = []
+
+    # ---- Resolve the reference (safety) list -------------------------
+    # Default: server-side list (repo file or private dataset). If the user
+    # uploads an override on the sidebar, use it for THIS run only.
+    try:
+        if reference_file is not None:
+            wl = build_whitelist_from_path(reference_file.name)
+            log.append(f"Using UPLOADED reference list: {Path(reference_file.name).name}")
+        else:
+            wl = load_whitelist()
+            log.append("Using server-side safety list.")
+    except FileNotFoundError as e:
+        return "\n".join(log) + f"\n\nREFERENCE FILE ERROR:\n{e}", []
+    except Exception as e:
+        return "\n".join(log) + f"\n\nCould not parse reference list: {e}", []
+
+    # ---- Load & merge uploads ----------------------------------------
+    dfs = []
+    for f in uploaded_files:
+        try:
+            dfs.append(pd.read_excel(f.name))
+            log.append(f"Loaded {Path(f.name).name}")
+        except Exception as e:
+            log.append(f"ERROR loading {Path(f.name).name}: {e}")
+    if not dfs:
+        return "\n".join(log) + "\nNo files could be read.", []
+    df = pd.concat(dfs, ignore_index=True)
+    log.append(f"Merged shape: {df.shape}")
+
+    # ---- FIPS validation (uses the reference list) -------------------
+    try:
+        df, audit = fips_validate(df, blank_cities_not_in_list, log, wl=wl)
+    except FileNotFoundError as e:
+        return "\n".join(log) + f"\n\nREFERENCE FILE ERROR:\n{e}", []
+
+    # ---- Download verification files (audit + RawFile_FIPSValidated) -
+    if len(audit) > 0:
+        ap = out_dir / "FIPS_Validation_Audit_ChangedRecords.xlsx"
+        audit.to_excel(ap, index=False)
+        outputs.append(str(ap))
+
+    if len(df) <= EXCEL_ROW_LIMIT:
+        rp = out_dir / "RawFile_FIPSValidated.xlsx"
+        df.to_excel(rp, index=False)
+    else:
+        rp = out_dir / "RawFile_FIPSValidated.csv"
+        df.to_csv(rp, index=False)
+    outputs.append(str(rp))
+
+    # ---- UNDETERMINED/NULL -> NaN ------------------------------------
+    for col in ["DDPPropertyCityFinal", "DDPPropertyZIPFinal",
+                "DDPRevisedCensusTract", "LegalSubDivision"]:
+        if col in df.columns:
+            df[col] = df[col].replace(["UNDETERMINED CITY", "UNDETERMINED ZIP", "NULL"], np.nan)
+
+    # ---- Rename FIPS col -> County -----------------------------------
+    if "** REC FIPS-COUNTY" in df.columns:
+        df = df.rename(columns={"** REC FIPS-COUNTY": "County"})
+
+    # ---- Optional: filter to single most-dominant county -------------
+    if filter_dominant_county:
+        dominant = df["County"].value_counts().idxmax()
+        df = df[df["County"] == dominant].copy()
+        log.append(f"Filtered to dominant county: {dominant} -> {df.shape}")
+
+    # ---- Source frames -----------------------------------------------
+    city_df = df.loc[df["DDPPropertyCityFinal"].notna()].copy()
+    zip_df = df.loc[df["DDPPropertyZIPFinal"].notna()].copy()
+    census_tracts_df = df.loc[df["DDPRevisedCensusTract"].notna()].copy()
+    original = df.loc[df["LegalSubDivision"].notna()].copy()
+
+    # ---- Cluster ------------------------------------------------------
+    cluster_df = build_clusters(city_df, zip_df, census_tracts_df, cluster_threshold)
+    log.append(f"Clusters: {cluster_df['Cluster_ID'].nunique():,}")
+
+    # ---- Replace-with (majority vote) --------------------------------
+    zip_w = build_replace_with(
+        build_cluster_wide_table(zip_df, cluster_df, "DDPPropertyZIPFinal", "ZIP_"),
+        "ZIP_", replace_threshold).rename(columns={"Replace with": "Replace with Zip Codes"})
+    city_w = build_replace_with(
+        build_cluster_wide_table(city_df, cluster_df, "DDPPropertyCityFinal", "City_"),
+        "City_", replace_threshold).rename(columns={"Replace with": "Replace with Cities"})
+    cen_w = build_replace_with(
+        build_cluster_wide_table(census_tracts_df, cluster_df, "DDPRevisedCensusTract", "CensusTract_"),
+        "CensusTract_", replace_threshold).rename(columns={"Replace with": "Replace with Census Tracts"})
+
+    # ---- Map back onto original --------------------------------------
+    lk = cluster_df[["LegalSubDivision", "Cluster_ID"]].drop_duplicates("LegalSubDivision")
+    cmap = dict(zip(lk["LegalSubDivision"], lk["Cluster_ID"]))
+    original_final = original.copy()
+    original_final["Cluster_ID"] = original_final["LegalSubDivision"].map(cmap)
+    original_final["Replace with Zip Codes"] = original_final["Cluster_ID"].map(
+        dict(zip(zip_w["Cluster_ID"], zip_w["Replace with Zip Codes"])))
+    original_final["Replace with Cities"] = original_final["Cluster_ID"].map(
+        dict(zip(city_w["Cluster_ID"], city_w["Replace with Cities"])))
+    original_final["Replace with Census Tracts"] = original_final["Cluster_ID"].map(
+        dict(zip(cen_w["Cluster_ID"], cen_w["Replace with Census Tracts"])))
+
+    # Normalize ZIPs (drop float '.0', restore 5-digit zero-padding).
+    original_final["Replace with Zip Codes"] = (
+        original_final["Replace with Zip Codes"].apply(clean_zip))
+
+    # ---- Reorder columns (desired_leading_columns) -------------------
+    desired_leading = [
+        "RecordKey", "ComesFromMtg2", "IsFlagged", "Cluster_ID", "LegalSubDivision",
+        "County", "Replace with Zip Codes", "Replace with Cities", "Replace with Census Tracts",
+        "DDPPropertyZIPFinal", "DDPPropertyCityFinal", "DDPRevisedCensusTract",
+        "LotBlockSubdivisionTract", "DDPRevisedLatitude", "DDPRevisedLongitude",
+    ]
+    existing = original_final.columns.tolist()
+    lead = [c for c in desired_leading if c in existing]
+    other = [c for c in existing if c not in lead]
+    original_final = original_final[lead + other]
+
+    original_final_output = original_final.copy()   # full output snapshot
+
+    # ---- Split into final cities / zips / census tracts --------------
+    keep = ["RecordKey", "ComesFromMtg2", "IsFlagged",
+            "Replace with Zip Codes", "Replace with Cities", "Replace with Census Tracts",
+            "DDPPropertyZIPFinal", "DDPPropertyCityFinal", "DDPRevisedCensusTract"]
+    keep = [c for c in keep if c in original_final.columns]
+    of = original_final[keep].rename(columns={
+        "DDPPropertyZIPFinal": "DDPPropertyZIPFinalToRemove",
+        "DDPPropertyCityFinal": "DDPPropertyCityFinalToRemove",
+        "DDPRevisedCensusTract": "DDPRevisedCensusTractToRemove",
+        "Replace with Zip Codes": "DDPPropertyZIPFinal",
+        "Replace with Cities": "DDPPropertyCityFinal",
+        "Replace with Census Tracts": "DDPRevisedCensusTract",
+    })
+    of["DDPRevisedCensusTract"] = of["DDPRevisedCensusTract"].apply(clean_census_tract_suffix)
+
+    idcols = [c for c in ["RecordKey", "ComesFromMtg2", "IsFlagged"] if c in of.columns]
+    final_cities = of[idcols + ["DDPPropertyCityFinalToRemove", "DDPPropertyCityFinal"]].copy()
+    final_zips = of[idcols + ["DDPPropertyZIPFinalToRemove", "DDPPropertyZIPFinal"]].copy()
+    final_ct = of[idcols + ["DDPRevisedCensusTractToRemove", "DDPRevisedCensusTract"]].copy()
+
+    # keep only rows whose ORIGINAL value was blank (the ones we filled)
+    final_cities = final_cities[final_cities["DDPPropertyCityFinalToRemove"].isna()].copy()
+    final_zips = final_zips[final_zips["DDPPropertyZIPFinalToRemove"].isna()].copy()
+    final_ct = final_ct[final_ct["DDPRevisedCensusTractToRemove"].isna()].copy()
+
+    final_cities = final_cities.drop(columns=["DDPPropertyCityFinalToRemove"])
+    final_zips = final_zips.drop(columns=["DDPPropertyZIPFinalToRemove"])
+    final_ct = final_ct.drop(columns=["DDPRevisedCensusTractToRemove"])
+
+    # ---- OnlyBlanks warnings (column entirely null) ------------------
+    def only_blanks_warning(d, col, prefix):
+        if col in d.columns and d[col].isnull().all():
+            p = out_dir / f"OnlyBlanks_{prefix}.xlsx"
+            d.to_excel(p, index=False)
+            outputs.append(str(p))
+            log.append(f"WARNING: {prefix} has only blanks in '{col}' -> {p.name}")
+
+    only_blanks_warning(final_cities, "DDPPropertyCityFinal", "final_cities")
+    only_blanks_warning(final_zips, "DDPPropertyZIPFinal", "final_zips")
+    only_blanks_warning(final_ct, "DDPRevisedCensusTract", "final_census_tracts")
+
+    # ---- Drop NaNs so only program assignments remain ----------------
+    final_cities = final_cities[final_cities["DDPPropertyCityFinal"].notna()].copy()
+    final_zips = final_zips[final_zips["DDPPropertyZIPFinal"].notna()].copy()
+    final_ct = final_ct[final_ct["DDPRevisedCensusTract"].notna()].copy()
+
+    # ---- Write the per-category + cluster outputs --------------------
+    for d, name in [(final_cities, "final_cities.xlsx"),
+                    (final_zips, "final_zips.xlsx"),
+                    (final_ct, "final_census_tracts.xlsx"),
+                    (cluster_df, "cluster_df.xlsx")]:
+        p = out_dir / name
+        d.to_excel(p, index=False)
+        outputs.append(str(p))
+
+    # ---- Full output CSV (chunked at 1,000,000 rows) -----------------
+    csv_name = f"OutputFile_{replace_threshold}%Threshold_{cluster_threshold}%ClustersRapidFuzz.csv"
+    outputs.extend(save_csv_in_parts(original_final_output, csv_name, out_dir, row_limit=1_000_000))
+
+    log.append(f"Done. {len(outputs)} files ready.")
+    gc.collect()
+    return "\n".join(log), outputs
+
+
+# =====================================================================
+# STARTUP HEALTH CHECK
+# =====================================================================
+def _startup_health_check():
+    if os.environ.get("SELF_TEST_ON_STARTUP", "1") != "1":
+        return
+    try:
+        print("[startup] " + run_self_test())
+    except FileNotFoundError as e:
+        # Reference file not configured yet — let the UI load so the user can fix it.
+        print(f"[startup] Self-test skipped — safety list unavailable:\n{e}")
+    # AssertionError intentionally propagates: a failed self-test fails the boot.
+
+
+_startup_health_check()
+
+
+# =====================================================================
+# GRADIO UI  (simple vertical layout — previous design)
+# =====================================================================
+with gr.Blocks() as demo:
+    gr.Markdown(
+        "## Subdivision Cluster + FIPS Validation\n"
+        "Upload the Excel files to process. Records are validated against the "
+        "Subdivision Program Safety List, then `LegalSubDivision` values are "
+        "fuzzy-clustered and city/ZIP/census-tract are filled by majority vote.\n\n"
+        "*Defaults reproduce the notebook exactly. Change the controls only if you "
+        "intend to deviate from it.*"
+    )
+    files_in = gr.File(file_count="multiple", file_types=[".xlsx", ".xls"],
+                       label="Excel files to process")
+    with gr.Row():
+        blank_flag = gr.Checkbox(
+            value=True, label="Strict: blank cities not in the safety list")
+        dominant_flag = gr.Checkbox(
+            value=True, label="Filter to single most-dominant county")
+    with gr.Row():
+        clu_thr = gr.Slider(70, 100, value=95, step=1, label="Cluster threshold (%)")
+        rep_thr = gr.Slider(50, 100, value=70, step=1, label="Replace-with threshold (%)")
+    run_btn = gr.Button("Run pipeline", variant="primary")
+    status = gr.Textbox(label="Log", lines=14)
+    files_out = gr.File(label="Output files", file_count="multiple")
+
+    def _run_simple(files, blank, dominant, clu, rep):
+        # This layout has no reference upload, so always use the server-side list.
+        return run_pipeline(files, None, blank, dominant, clu, rep)
+
+    run_btn.click(
+        _run_simple,
+        inputs=[files_in, blank_flag, dominant_flag, clu_thr, rep_thr],
+        outputs=[status, files_out],
+    )
+
+if __name__ == "__main__":
+    demo.launch()

requirements.txt

@@ -0,0 +1,10 @@
+gradio>=4.0
+pandas>=2.0
+numpy>=1.24
+openpyxl>=3.1
+rapidfuzz>=3.0
+huggingface_hub>=0.23,<1.0
+pydantic>=2.0,<2.11
+fastapi<0.116
+starlette<1.0
+audioop-lts; python_version >= "3.13"