Upload app_simple_ui.py

app_simple_ui.py

@@ -0,0 +1,826 @@
+"""
+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 time
+import tempfile
+import zipfile
+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
+
+# Fixed pipeline settings (notebook values). Not exposed in the UI on purpose:
+# users should not be able to deviate from the notebook's behavior.
+STRICT_BLANK_CITIES = True    # blank cities that are not in the safety list
+CLUSTER_THRESHOLD = 95        # RapidFuzz similarity for LegalSubDivision clusters
+REPLACE_THRESHOLD = 70        # majority-vote % required to fill a blank
+FILTER_DOMINANT_COUNTY = False  # process ALL counties (multi-county / full-state files)
+
+
+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
+
+
+# ---------------------------------------------------------------------
+# FIPS-COUNTY COLUMN DETECTION (by CONTENT, not header name)
+# ---------------------------------------------------------------------
+# The source header has changed before ("** REC FIPS-COUNTY" ->
+# "** REC FIPS-COUNTY NAME"), so we identify the column by what's IN it:
+# values shaped like "05045-FAULKNER" (FIPS digits, a dash, a county name).
+# Header names are only used as a last-resort fallback.
+FIPS_COUNTY_PATTERN = r"^\s*\d{1,5}\s*-\s*[A-Za-z]"
+FALLBACK_FIPS_HEADERS = ["** REC FIPS-COUNTY NAME", "** REC FIPS-COUNTY", "County"]
+
+
+def detect_fips_county_column(df, sample_size=1000, min_score=0.5):
+    """Return (column_name, reason). Picks the column whose values best match
+    the 'FIPS-COUNTYNAME' shape; falls back to known headers, then to any
+    header mentioning FIPS."""
+    best_col, best_score = None, 0.0
+    for col in df.columns:
+        s = df[col].dropna()
+        if s.empty:
+            continue
+        sample = s.head(sample_size).astype(str)
+        try:
+            score = float(sample.str.match(FIPS_COUNTY_PATTERN).mean())
+        except Exception:
+            continue
+        if score > best_score:
+            best_col, best_score = col, score
+
+    if best_col is not None and best_score >= min_score:
+        return best_col, f"content match {best_score:.0%} of sampled rows"
+
+    for name in FALLBACK_FIPS_HEADERS:
+        if name in df.columns:
+            return name, "fallback: known header name"
+
+    for col in df.columns:
+        if "FIPS" in str(col).upper():
+            return col, "fallback: header mentions FIPS"
+
+    return None, "not found"
+
+
+# =====================================================================
+# 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,
+                  fips_col=None):
+    if wl is None:
+        wl = load_whitelist()
+    valid_pairs, valid_cities = wl["pairs"], wl["cities"]
+
+    if fips_col is None:
+        fips_col, _reason = detect_fips_county_column(df)
+    if fips_col is None:
+        raise ValueError(
+            "Could not find a FIPS-county column. Expected values like "
+            "'05045-FAULKNER' in one of the columns."
+        )
+    FIPS_COL = fips_col
+    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)
+
+    # Canonical order so the SAME set of names always clusters the same way,
+    # regardless of the order rows arrived in (single-county vs whole-state,
+    # app vs app_simple_ui, or any file-upload order). The greedy grouping below
+    # is order-sensitive, so we sort first to make it deterministic.
+    cleaned.sort(key=lambda x: (x.upper(), x))
+
+    visited, groups = set(), []
+    buckets = defaultdict(list)
+    for s in cleaned:
+        buckets[s.upper()[:prefix_len]].append(s)
+
+    for key in sorted(buckets.keys()):
+        bucket = buckets[key]
+        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 = []   # (county, [members]) so each cluster stays tied to its county
+    for county in sorted(all_legal["County"].unique()):
+        strings = all_legal[all_legal["County"] == county]["LegalSubDivision"].tolist()
+        for grp in group_similar_strings_optimized(
+                strings, threshold=cluster_threshold, prefix_len=3, max_len_diff=3):
+            groups.append((county, grp))
+
+    rows = []
+    for cid, (county, group) in enumerate(groups, start=1):
+        for sub in group:
+            rows.append({"Cluster_ID": cid, "County": county, "LegalSubDivision": sub})
+    return pd.DataFrame(rows, columns=["Cluster_ID", "County", "LegalSubDivision"])
+
+
+# =====================================================================
+# STEP 3 — WIDE TABLES + MAJORITY-VOTE "REPLACE WITH"
+# =====================================================================
+def build_cluster_wide_table(source_df, cluster_df, value_col, prefix):
+    # Map by (County, LegalSubDivision) -- NOT LegalSubDivision alone. The same
+    # subdivision name occurs in many counties, each with its own cluster; keying
+    # on the name only would collapse them and pool values across counties.
+    lk = cluster_df[["County", "LegalSubDivision", "Cluster_ID"]].drop_duplicates(
+        ["County", "LegalSubDivision"])
+    cluster_map = {(c, s): cid for c, s, cid in
+                   zip(lk["County"].astype(str).str.strip(),
+                       lk["LegalSubDivision"].astype(str).str.strip(), lk["Cluster_ID"])}
+
+    w = source_df[["County", "LegalSubDivision", value_col]].copy()
+    w["County"] = w["County"].astype(str).str.strip()
+    w["LegalSubDivision"] = w["LegalSubDivision"].astype(str).str.strip()
+    w[value_col] = w[value_col].replace("", np.nan)
+    w["Cluster_ID"] = [cluster_map.get((c, s))
+                       for c, s in zip(w["County"], w["LegalSubDivision"])]
+    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
+
+
+_BOOL_MAP = {
+    True: "TRUE", False: "FALSE", 1: "TRUE", 0: "FALSE",
+    "1": "TRUE", "0": "FALSE", "true": "TRUE", "false": "FALSE",
+    "t": "TRUE", "f": "FALSE", "yes": "TRUE", "no": "FALSE",
+}
+
+
+def boolify(series):
+    """Render a 0/1 (or True/False) column as literal 'TRUE'/'FALSE' text.
+    Recognized values are mapped; anything unrecognized is left as-is; NaN kept."""
+    def conv(v):
+        if pd.isna(v):
+            return v
+        if v in _BOOL_MAP:
+            return _BOOL_MAP[v]
+        return _BOOL_MAP.get(str(v).strip().lower(), v)
+    return series.map(conv)
+
+
+# =====================================================================
+# 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):
+    t_start = time.time()
+    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}")
+
+    # ---- Locate the FIPS-county column by CONTENT --------------------
+    fips_col, fips_reason = detect_fips_county_column(df)
+    if fips_col is None:
+        return ("\n".join(log) +
+                "\n\nCOLUMN ERROR: could not find a FIPS-county column. Expected "
+                "values like '05045-FAULKNER' in one of the columns."), []
+    log.append(f"FIPS-county column: '{fips_col}' ({fips_reason})")
+
+    # ---- FIPS validation (uses the reference list) -------------------
+    try:
+        df, audit = fips_validate(df, blank_cities_not_in_list, log, wl=wl, fips_col=fips_col)
+    except FileNotFoundError as e:
+        return "\n".join(log) + f"\n\nREFERENCE FILE ERROR:\n{e}", []
+
+    # ---- Verification files (audit + RawFile_FIPSValidated), CSV -----
+    if len(audit) > 0:
+        ap = out_dir / "FIPS_Validation_Audit_ChangedRecords.csv"
+        audit.to_csv(ap, index=False)
+        outputs.append(str(ap))
+
+    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 detected FIPS col -> County --------------------------
+    if fips_col != "County":
+        if "County" in df.columns:
+            df = df.drop(columns=["County"])
+        df = df.rename(columns={fips_col: "County"})
+
+    # ---- Filter to single most-dominant county -----------------------
+    # DISABLED: we now process big multi-county files (up to a full state) in
+    # one run, so we keep every county. Clustering already runs per-county.
+    # To re-enable single-county mode, uncomment the block below.
+    # 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 --------------------------------------
+    # Same fix: assign each row its county's cluster, keyed on (County, name).
+    lk = cluster_df[["County", "LegalSubDivision", "Cluster_ID"]].drop_duplicates(
+        ["County", "LegalSubDivision"])
+    cmap = {(c, s): cid for c, s, cid in
+            zip(lk["County"].astype(str).str.strip(),
+                lk["LegalSubDivision"].astype(str).str.strip(), lk["Cluster_ID"])}
+    original_final = original.copy()
+    _oc = original_final["County"].astype(str).str.strip()
+    _os = original_final["LegalSubDivision"].astype(str).str.strip()
+    original_final["Cluster_ID"] = [cmap.get((c, s)) for c, s in zip(_oc, _os)]
+    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}.csv"
+            d.to_csv(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()
+
+    # ---- Boolean columns as TRUE/FALSE (platform can't read 0/1) -----
+    for d in (final_cities, final_zips, final_ct):
+        for bc in ("ComesFromMtg2", "IsFlagged"):
+            if bc in d.columns:
+                d[bc] = boolify(d[bc])
+
+    # ---- Write the per-category + cluster outputs (CSV) --------------
+    for d, name in [(final_cities, "final_cities.csv"),
+                    (final_zips, "final_zips.csv"),
+                    (final_ct, "final_census_tracts.csv"),
+                    (cluster_df, "cluster_df.csv")]:
+        p = out_dir / name
+        d.to_csv(p, index=False)
+        outputs.append(str(p))
+
+    # ---- Full output CSV (chunked at 1,000,000 rows) -----------------
+    # NB: no '%' in the filename -- '%' is the URL-encoding character and makes
+    # the browser download fail with "unknown server error".
+    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))
+
+    # ---- Summary sheet: ONE ROW PER COUNTY ---------------------------
+    # Totals (records / subdivisions) come from the FULL post-validation frame
+    # (df -- every row, incl. those with no LegalSubDivision). The blank/fill/
+    # cluster metrics come from the clustered subset (output_df), which is the
+    # only set the pipeline can actually fill. Other output files are unchanged.
+    output_df = original_final_output     # rows WITH a LegalSubDivision + fills
+    full_df = df                          # ALL post-validation rows, has County
+    city_df, zips_df, census_df = final_cities, final_zips, final_ct
+
+    def _fips_name(label):
+        raw = str(label)
+        parts = raw.split("-")
+        fips = parts[0].strip()
+        name = parts[1].strip().title() if len(parts) > 1 else raw.strip().title()
+        return fips, name
+
+    RW_CITY, RW_ZIP, RW_TRACT = ("Replace with Cities", "Replace with Zip Codes",
+                                 "Replace with Census Tracts")
+    ORIG_CITY, ORIG_ZIP, ORIG_TRACT = ("DDPPropertyCityFinal", "DDPPropertyZIPFinal",
+                                       "DDPRevisedCensusTract")
+
+    def _blank(g, col):
+        return g[col].isna() if col in g.columns else pd.Series(False, index=g.index)
+
+    def _fill(g, col):
+        return g[col].notna() if col in g.columns else pd.Series(False, index=g.index)
+
+    def _out_subset(label):
+        if "County" not in output_df.columns:
+            return output_df.iloc[0:0]
+        m = output_df["County"].isna() if pd.isna(label) else (output_df["County"] == label)
+        return output_df[m]
+
+    summary_rows = []
+    if "County" in full_df.columns and len(full_df):
+        for county_label, gfull in full_df.groupby("County", dropna=False):
+            fips, name = _fips_name(county_label)
+            g = _out_subset(county_label)   # clustered rows for this county
+            city_blank, zip_blank, tract_blank = (_blank(g, ORIG_CITY),
+                                                  _blank(g, ORIG_ZIP), _blank(g, ORIG_TRACT))
+            city_fill, zip_fill, tract_fill = (_fill(g, RW_CITY),
+                                               _fill(g, RW_ZIP), _fill(g, RW_TRACT))
+            summary_rows.append({
+                "County": name,
+                "FIPS": fips,
+                # true totals from the full frame
+                "Total Records in County": len(gfull),
+                # how many of those records carry a LegalSubDivision (non-unique)
+                "Total Subdivisions": int(gfull["LegalSubDivision"].count())
+                                      if "LegalSubDivision" in gfull.columns else 0,
+                "Total Clusters Made": int(g["Cluster_ID"].nunique())
+                                       if "Cluster_ID" in g.columns else 0,
+                "Total Blank Cities": int(city_blank.sum()),
+                "Total Cities Assignment": int((city_blank & city_fill).sum()),
+                "Total Blank Zips": int(zip_blank.sum()),
+                "Total Zips Assignment": int((zip_blank & zip_fill).sum()),
+                "Total Blank Census": int(tract_blank.sum()),
+                "Total Census Assignment": int((tract_blank & tract_fill).sum()),
+            })
+
+    summary_cols = ["County", "FIPS", "Total Records in County", "Total Subdivisions",
+                    "Total Clusters Made", "Total Blank Cities", "Total Cities Assignment",
+                    "Total Blank Zips", "Total Zips Assignment", "Total Blank Census",
+                    "Total Census Assignment"]
+    summary_df = pd.DataFrame(summary_rows, columns=summary_cols)
+    log.append(f"Summary: {len(summary_df)} county row(s).")
+
+    summary_path = out_dir / "Summary.csv"
+    summary_df.to_csv(summary_path, index=False)
+    outputs.append(str(summary_path))
+
+    # ---- Master bundle: one ZIP of CSVs (no Excel row limit) ---------
+    # A single CSV can hold only one table, so the "master workbook" becomes a
+    # ZIP that bundles every result as its own CSV -- including the full output
+    # (chunked into parts), which no longer hits Excel's 1,048,576-row ceiling.
+    # CSV writing also avoids openpyxl's heavy memory use on huge files.
+    master_path = out_dir / "Master_Dataset.zip"
+    member_names = ["FIPS_Validation_Audit_ChangedRecords.csv", "cluster_df.csv",
+                    "final_cities.csv", "final_zips.csv", "final_census_tracts.csv",
+                    "Summary.csv"]
+    members = [out_dir / n for n in member_names if (out_dir / n).exists()]
+    members += sorted(out_dir.glob("OutputFile_*.csv"))   # full output (chunked)
+    with zipfile.ZipFile(master_path, "w", zipfile.ZIP_DEFLATED) as zf:
+        for p in members:
+            zf.write(p, arcname=p.name)
+    outputs.append(str(master_path))
+
+    elapsed_min = (time.time() - t_start) / 60
+    log.append(f"Done. {len(outputs)} files ready.")
+    log.append(f"Total processing time: {elapsed_min:.1f} minutes.")
+    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"
+        "*Runs with the notebook's fixed settings.*"
+    )
+    files_in = gr.File(file_count="multiple", file_types=[".xlsx", ".xls"],
+                       label="Excel files to process")
+    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):
+        # No reference upload in this layout -> always the server-side list.
+        # All settings are fixed (see constants at top). No user-facing options.
+        return run_pipeline(files, None, STRICT_BLANK_CITIES, FILTER_DOMINANT_COUNTY,
+                            CLUSTER_THRESHOLD, REPLACE_THRESHOLD)
+
+    run_btn.click(
+        _run_simple,
+        inputs=[files_in],
+        outputs=[status, files_out],
+    )
+
+if __name__ == "__main__":
+    demo.launch()