Delete app_simple_ui.py

app_simple_ui.py

@@ -1,819 +0,0 @@
-"""
-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)
-
-    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 = []   # (county, [members]) so each cluster stays tied to its county
-    for county in 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()