Blame for app.py

"""
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)

# Master bundle format:
#   "excel" -> Master_Dataset.xlsx, every result on its own sheet (full output
#              split across Output_..._1/_2/... if it exceeds Excel's row cap)
#   "csv"   -> Master_Dataset.zip, every result bundled as a CSV (faster: no Excel
#              writing -- use this for very large / state-sized runs)
# The individual per-category files are CSV in both modes.
MASTER_OUTPUT = "excel"

# How records are processed:
#   "stream" -> ONE COUNTY AT A TIME. The input is split into per-county temp
#               files (chunked read, never holding the whole state), each county
#               is processed and its results appended. Peak RAM ~ the largest
#               single county, so full states run on a small/cheap tier.
#   "memory" -> the whole dataset is loaded and processed at once (faster on
#               small files, but needs enough RAM to hold everything).
# Results are identical either way. Use "stream" for state-sized files.
PROCESS_MODE = "stream"
STREAM_CHUNK_ROWS = 400_000   # rows per chunk while splitting the input by county

# RawFile_FIPSValidated.csv is a full copy of the validated input (as big as the
# input itself). It's rarely used downstream and dominates the output/upload time
# on big states, so it's OFF by default. Set True if you actually need that file.
WRITE_RAWFILE = False

# ---- Dataset auto-cleanup (Run from dataset) -------------------------------
# HF datasets are git repos: deleting a file leaves its bytes in git HISTORY, so
# storage is only reclaimed by SQUASHING the history. These options keep the two
# datasets small automatically so you never hit the private-storage quota again.
# All of this runs ONLY after results have been safely uploaded, so nothing is
# lost if a run fails.
DELETE_INPUT_AFTER_RUN = True   # delete the processed input file(s) from the input dataset
KEEP_LATEST_OUTPUT_RUNS = 1     # in the output dataset, keep only the N newest runs/ folders (0 = keep all)
SQUASH_HISTORY = True           # after deleting, squash both datasets' history to reclaim the storage

# Threshold for the RURAL/URBAN label (records <= this -> RURAL, else URBAN).
RURAL_MAX_RECORDS = 10000

# State FIPS prefix (first 2 digits of a county FIPS) -> USPS abbreviation, so the
# Summary can fill the tracking file's "State" column with no manual lookup.
STATE_FIPS_ABBR = {
    "01": "AL", "02": "AK", "04": "AZ", "05": "AR", "06": "CA", "08": "CO",
    "09": "CT", "10": "DE", "11": "DC", "12": "FL", "13": "GA", "15": "HI",
    "16": "ID", "17": "IL", "18": "IN", "19": "IA", "20": "KS", "21": "KY",
    "22": "LA", "23": "ME", "24": "MD", "25": "MA", "26": "MI", "27": "MN",
    "28": "MS", "29": "MO", "30": "MT", "31": "NE", "32": "NV", "33": "NH",
    "34": "NJ", "35": "NM", "36": "NY", "37": "NC", "38": "ND", "39": "OH",
    "40": "OK", "41": "OR", "42": "PA", "44": "RI", "45": "SC", "46": "SD",
    "47": "TN", "48": "TX", "49": "UT", "50": "VT", "51": "VA", "53": "WA",
    "54": "WV", "55": "WI", "56": "WY",
}

# Exact column layout of the tracking file's Main sheet (+ RURAL/URBAN at the end),
# so a whole run can be pasted in one block with NO formulas left in Excel.
SUMMARY_COLUMNS = [
    "Date Processed", "State", "County", "FIPS",
    "Overall Total Dataset City Assignment", "Overall Total Undetermined City Assignment",
    "Overall Total Dataset Zip Assignment", "Overall Total Undetermined Zip Assignment",
    "Overall Total Dataset Census Tract Assignment", "Overall Total Undetermined Census Tract Assignment",
    "Total Records in County", "Total Subdivisions In County", "Total Clusters Made",
    "Total Undetermined City", "Total City Assignments", "Total Undetermined Zip",
    "Total Zip Assignments", "Total Blank Census", "Total Census Assignments",
    "Added By", "Processing Times/Notes", "Run", "RURAL/URBAN",
    "Total Blanks before run", "% of Total Blanks before run",
    "Total Blanks after run", "% of Total Blanks after run",
]


def _pct(num, den):
    """Percentage num/den*100, rounded to 2 dp; 0 when the denominator is 0
    (replaces the Excel formulas, which would show #DIV/0! in that case)."""
    return round(num / den * 100, 2) if den else 0


def _summary_row(name, fips, records, subdivisions, clusters,
                 undet_city, city_assign, undet_zip, zip_assign, blank_census, census_assign):
    """Build one fully-computed Summary row matching SUMMARY_COLUMNS. Percentages,
    run date, state, and RURAL/URBAN are all computed here so the tracking file
    needs no formulas. Used by BOTH the streaming and memory paths (no drift)."""
    blanks_before = undet_city + undet_zip + blank_census
    filled = city_assign + zip_assign + census_assign
    blanks_after = blanks_before - filled          # blanks the program couldn't resolve
    return {
        "Date Processed": time.strftime("%m/%d/%Y"),
        "State": STATE_FIPS_ABBR.get(str(fips).strip()[:2], ""),
        "County": name,
        "FIPS": fips,
        "Overall Total Dataset City Assignment": _pct(city_assign, records),
        "Overall Total Undetermined City Assignment": _pct(city_assign, undet_city),
        "Overall Total Dataset Zip Assignment": _pct(zip_assign, records),
        "Overall Total Undetermined Zip Assignment": _pct(zip_assign, undet_zip),
        "Overall Total Dataset Census Tract Assignment": _pct(census_assign, records),
        "Overall Total Undetermined Census Tract Assignment": _pct(census_assign, blank_census),
        "Total Records in County": records,
        "Total Subdivisions In County": subdivisions,
        "Total Clusters Made": clusters,
        "Total Undetermined City": undet_city,
        "Total City Assignments": city_assign,
        "Total Undetermined Zip": undet_zip,
        "Total Zip Assignments": zip_assign,
        "Total Blank Census": blank_census,
        "Total Census Assignments": census_assign,
        "Added By": "",
        "Processing Times/Notes": "",
        "Run": "",
        "RURAL/URBAN": "RURAL" if records <= RURAL_MAX_RECORDS else "URBAN",
        "Total Blanks before run": blanks_before,
        "% of Total Blanks before run": round(blanks_before / records, 2) if records else 0,
        "Total Blanks after run": blanks_after,
        "% of Total Blanks after run": round(blanks_after / records, 2) if records else 0,
    }


def _mem_mb():
    """Current process memory (RSS) in MB. psutil if available, else peak RSS."""
    try:
        import psutil
        return psutil.Process().memory_info().rss / (1024 * 1024)
    except Exception:
        try:
            import resource
            return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024  # KB -> MB
        except Exception:
            return float("nan")


def _step(log, msg):
    """Record a phase line with a timestamp + current RAM, and PRINT it live
    (flushed) so it reaches the Space's container log immediately. If the process
    is killed (e.g. out of memory), the last printed line pinpoints the phase."""
    line = f"[{time.strftime('%H:%M:%S')}] {msg} | RAM {_mem_mb():.0f} MB"
    print(line, flush=True)
    log.append(line)
    return line


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 detect_delimiter(path, candidates=(",", "|", "^", "\t", ";", "\x1f", "\x1e")):
    """Sniff a text file's field delimiter from its first lines. SQL exports here
    are sometimes '^' or '|' rather than ','. Also recognizes the ASCII control
    separators US ('\\x1f') and RS ('\\x1e') -- ideal for messy unquoted data since
    free-text fields never contain them. Returns the delimiter char (default ',').
    Detects once from a small header sample so the actual (possibly chunked) read
    can still use pandas' fast C parser with an explicit sep."""
    try:
        with open(path, "r", encoding="utf-8", errors="replace", newline="") as fh:
            lines = []
            for _ in range(8):
                ln = fh.readline()
                if not ln:
                    break
                ln = ln.rstrip("\r\n")
                if ln:
                    lines.append(ln)
    except Exception:
        return ","
    if not lines:
        return ","
    best, best_score = ",", -1
    for d in candidates:
        header_count = lines[0].count(d)
        if header_count == 0:
            continue
        # reward a high, consistent field count across the sampled rows
        consistent = sum(1 for ln in lines if ln.count(d) == header_count)
        score = header_count * 100 + consistent
        if score > best_score:
            best, best_score = d, score
    return best


_CTRL_DELETE = {c: None for c in range(32)}
for _keep in (9, 10, 13):            # keep tab, newline, carriage return
    _CTRL_DELETE.pop(_keep, None)
_CTRL_DELETE[127] = None             # DEL


def _strip_ctrl(s):
    """Remove C0 control bytes (NUL, SUB/'EOF', and the rest) except tab/LF/CR.
    A stray control byte embedded in a text field (e.g. an address or legal
    description from the source system) otherwise either aborts pandas' C parser
    or silently truncates the field at a fake end-of-file."""
    return s.translate(_CTRL_DELETE)


class _CleanReader:
    """Text file-like wrapper that strips control bytes as the file is read, so a
    stray control character in one field can't break the parse. Works with pandas'
    chunked and whole-file reads alike; opened with errors='replace' so stray
    non-UTF-8 bytes don't abort the read either."""
    def __init__(self, path, encoding="utf-8", errors="replace"):
        self._f = open(path, "r", encoding=encoding, errors=errors, newline="")

    def read(self, size=-1):
        return _strip_ctrl(self._f.read(size))

    def readline(self, size=-1):
        return _strip_ctrl(self._f.readline(size))

    def __iter__(self):
        for line in self._f:
            yield _strip_ctrl(line)

    def close(self):
        self._f.close()

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        self.close()


def _read_csv_auto(path, **kwargs):
    """pd.read_csv with the delimiter auto-detected (unless caller passed sep),
    tolerating malformed rows (a stray delimiter in one row won't abort the read)."""
    if "sep" not in kwargs:
        kwargs["sep"] = detect_delimiter(path)
    kwargs.setdefault("on_bad_lines", "skip")
    with _CleanReader(path) as fh:
        return pd.read_csv(fh, **kwargs)


def scan_bad_rows(path, sep, out_dir):
    """Find rows with MORE fields than the header -- a stray delimiter inside a
    field, which is exactly what pandas skips. Save them verbatim (with their line
    number) to BadRows_<name>.csv so the offending fields can be reviewed and fixed
    at the source. Returns (count, bad_path_or_None)."""
    bad_path = Path(out_dir) / f"BadRows_{Path(path).stem}.csv"
    count = 0
    bw = None
    try:
        with open(path, "r", encoding="utf-8", errors="replace", newline="") as fin:
            header = _strip_ctrl(fin.readline())
            if not header:
                return 0, None
            base = header.rstrip("\r\n").count(sep)   # header delimiter count
            for lineno, line in enumerate(fin, start=2):
                s = _strip_ctrl(line).rstrip("\r\n")
                if not s:
                    continue
                if s.count(sep) > base:               # too many fields -> skipped by pandas
                    if bw is None:
                        bw = open(bad_path, "w", encoding="utf-8", newline="")
                        bw.write("line_number" + sep + header.rstrip("\r\n") + "\n")
                    bw.write(str(lineno) + sep + s + "\n")
                    count += 1
    finally:
        if bw is not None:
            bw.close()
    return count, (str(bad_path) if count else None)


def scan_inputs_for_bad_rows(uploaded_files, out_dir, log):
    """Scan every CSV input for malformed rows; log the count and save them for
    review. Returns the list of BadRows_*.csv paths produced."""
    bad_files = []
    for f in uploaded_files:
        nm = f.name
        if not str(nm).lower().endswith(".csv"):
            continue
        count, bad_path = scan_bad_rows(nm, detect_delimiter(nm), out_dir)
        if count:
            bad_files.append(bad_path)
            _step(log, f"{Path(nm).name}: {count:,} malformed row(s) skipped "
                       f"(saved to {Path(bad_path).name} for review)")
        else:
            _step(log, f"{Path(nm).name}: no malformed rows detected")
    return bad_files


def build_whitelist_from_path(path):
    """Parse a safety-list file (xlsx or csv) into {pairs, cities, rows, fips_name}."""
    if str(path).lower().endswith(".csv"):
        safety_df = _read_csv_auto(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()
    # FIPS -> county name, from the positional pairing of "Acceptable counties"
    # (comma-separated) against Acceptable FIPS 1..N. Used to synthesize the
    # "<FIPS>-<NAME>" column when source data has only bare FIPS codes.
    fips_name_votes = defaultdict(lambda: defaultdict(int))
    has_names = "Acceptable counties" in safety_df.columns
    for _, r in safety_df.iterrows():
        c = _norm_city(r["City Name"])
        if c is not None:
            valid_cities.add(c)
            for col in acceptable_cols:
                f = _norm_fips(r[col])
                if f:
                    valid_pairs.add(c + "||" + f)
        if has_names:
            names = [n.strip().upper() for n in str(r["Acceptable counties"]).split(",") if n.strip()]
            codes = [_norm_fips(r[col]) for col in acceptable_cols]
            for nm, code in zip(names, codes):
                if code and nm:
                    fips_name_votes[code][nm] += 1

    fips_name = {code: max(v.items(), key=lambda kv: kv[1])[0]
                 for code, v in fips_name_votes.items()}

    return {"pairs": valid_pairs, "cities": valid_cities, "rows": len(safety_df),
            "fips_name": fips_name}


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"


def _is_combined_fips(series, sample=1000):
    """True if the column already holds '<FIPS>-<NAME>' values (vs bare codes)."""
    s = series.dropna().astype(str).str.strip().head(sample)
    if len(s) == 0:
        return False
    try:
        return float(s.str.match(FIPS_COUNTY_PATTERN).mean()) >= 0.5
    except Exception:
        return False


def ensure_combined_county(df, fips_col, reason, wl, log=None):
    """Guarantee a '<FIPS>-<NAME>' county column. If `fips_col` already holds
    combined values (PBIX-processed data), keep it. If it holds BARE FIPS codes
    (raw SQL, e.g. DocumentRecordingCountyFIPS = '48201'), synthesize a 'County'
    column '48201-HARRIS' using the FIPS->name map from the safety list, so the
    rest of the pipeline behaves exactly as it does for the combined-column states.
    Returns (column_name_to_use, reason)."""
    if fips_col is None:
        return fips_col, reason
    if _is_combined_fips(df[fips_col]):
        return fips_col, reason
    fips_name = wl.get("fips_name", {}) if isinstance(wl, dict) else {}
    if not fips_name:
        return fips_col, reason  # no map -> leave bare (still groups/validates by FIPS)
    codes = df[fips_col].map(_norm_fips)          # 5-digit string or None
    names = codes.map(lambda c: fips_name.get(c, "") if c else "")
    df["County"] = [f"{c}-{n}" if c else "" for c, n in zip(codes.fillna(""), names)]
    if log is not None:
        unresolved = int(((names == "") & codes.notna()).sum())
        log.append(f"Built county names from bare FIPS column '{fips_col}' via the "
                   f"safety list ({unresolved:,} rows had no name match).")
    return "County", f"synthesized '<FIPS>-<NAME>' from bare FIPS column '{fips_col}'"


def ensure_flag_columns(df, log=None):
    """The Backoffice upload keys on RecordKey + IsFlagged + ComesFromMtg2, so the
    final files must carry those flag columns. Data pulled straight from SQL lacks
    IsFlagged / ComesFromMtg2 (the PBIX added them), so create them as FALSE when
    missing. Columns already present (e.g. a PBIX pull) are left untouched."""
    added = []
    for col in ("IsFlagged", "ComesFromMtg2"):
        if col not in df.columns:
            df[col] = False
            added.append(col)
    if added and log is not None:
        log.append(f"Added missing flag column(s) {', '.join(added)} = FALSE "
                   f"(not in input; required for the Backoffice upload).")
    return df


# =====================================================================
# 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


def write_master_workbook(path, sheets_in_order, row_limit=1_048_575):
    """Write a multi-sheet .xlsx from [(sheet_name, df), ...]. Any sheet larger than
    Excel's 1,048,576-row cap is split into <name>_1, <name>_2, ... so nothing is
    dropped. Uses xlsxwriter in constant_memory mode, driven row-by-row (low RAM --
    important for the full-output sheet on state-sized files). We write the rows
    ourselves because the pandas + constant_memory path corrupts rows that contain
    NaN cells. Falls back to openpyxl if xlsxwriter isn't installed."""
    try:
        import xlsxwriter
    except Exception:
        xlsxwriter = None

    if xlsxwriter is not None:
        wb = xlsxwriter.Workbook(str(path), {"constant_memory": True, "nan_inf_to_errors": True})

        def _write(name, dfx):
            ws = wb.add_worksheet(str(name)[:31])
            ws.write_row(0, 0, [str(c) for c in dfx.columns])
            clean = dfx.where(pd.notna(dfx), None)   # NaN -> blank cell
            for r, rowvals in enumerate(clean.itertuples(index=False, name=None), start=1):
                ws.write_row(r, 0, rowvals)

        for name, dfx in sheets_in_order:
            if dfx is None:
                continue
            n = len(dfx)
            if n <= row_limit:
                _write(name, dfx)
            else:
                for i in range(math.ceil(n / row_limit)):
                    _write(f"{name}_{i + 1}", dfx.iloc[i * row_limit:(i + 1) * row_limit])
                    gc.collect()
        wb.close()
    else:
        with pd.ExcelWriter(path, engine="openpyxl") as writer:
            for name, dfx in sheets_in_order:
                if dfx is None:
                    continue
                n = len(dfx)
                if n <= row_limit:
                    dfx.to_excel(writer, sheet_name=str(name)[:31], index=False)
                else:
                    for i in range(math.ceil(n / row_limit)):
                        dfx.iloc[i * row_limit:(i + 1) * row_limit].to_excel(
                            writer, sheet_name=f"{name}_{i + 1}"[:31], index=False)
    return str(path)


# =====================================================================
# MAIN PIPELINE
# =====================================================================
def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
                 filter_dominant_county, cluster_threshold, replace_threshold,
                 build_master=True):
    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}", []

    # ---- Scan inputs for malformed rows (save them for review) -------
    bad_files = scan_inputs_for_bad_rows(uploaded_files, out_dir, log)
    outputs.extend(bad_files)

    # ---- Load & merge uploads (Excel or CSV) -------------------------
    dfs = []
    for f in uploaded_files:
        name = f.name
        try:
            if str(name).lower().endswith(".csv"):
                sep = detect_delimiter(name)
                with _CleanReader(name) as _fh:
                    dfs.append(pd.read_csv(_fh, low_memory=False, sep=sep,
                                           on_bad_lines="skip"))
                _step(log, f"Loaded {Path(name).name}  ({len(dfs[-1]):,} rows, "
                           f"delimiter {sep!r})")
            else:
                dfs.append(pd.read_excel(name))
                _step(log, f"Loaded {Path(name).name}  ({len(dfs[-1]):,} rows)")
        except Exception as e:
            _step(log, f"ERROR loading {Path(name).name}: {e}")
    if not dfs:
        return "\n".join(log) + "\nNo files could be read.", []
    df = pd.concat(dfs, ignore_index=True)
    del dfs
    gc.collect()
    _step(log, f"Merged shape: {df.shape}")
    ensure_flag_columns(df, log)

    # ---- 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."), []
    _step(log, f"FIPS-county column: '{fips_col}' ({fips_reason})")
    fips_col, fips_reason = ensure_combined_county(df, fips_col, fips_reason, wl, log)
    if "synthesized" in fips_reason:
        _step(log, f"County column now '{fips_col}' ({fips_reason})")

    # ---- FIPS validation (uses the reference list) -------------------
    _step(log, "Validating FIPS/city pairs...")
    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}", []
    _step(log, "Validation done")

    # ---- 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))

    if WRITE_RAWFILE:
        rp = out_dir / "RawFile_FIPSValidated.csv"
        df.to_csv(rp, index=False)
        outputs.append(str(rp))
        _step(log, "Wrote RawFile_FIPSValidated.csv")

    # ---- 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()
    _step(log, f"Source frames built (subdivisions: {len(original):,})")

    # ---- Cluster ------------------------------------------------------
    _step(log, "Clustering LegalSubDivision per county...")
    cluster_df = build_clusters(city_df, zip_df, census_tracts_df, cluster_threshold)
    _step(log, f"Clusters: {cluster_df['Cluster_ID'].nunique():,}")

    # ---- Replace-with (majority vote) --------------------------------
    _step(log, "Building majority-vote tables...")
    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"})
    _step(log, "Majority-vote tables done")

    # ---- Map back onto original --------------------------------------
    _step(log, "Mapping clusters + fills back onto records...")
    # 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"])))

    # Clean integer Cluster_ID for output (no "2.0"); consistent with streaming.
    original_final["Cluster_ID"] = original_final["Cluster_ID"].astype("Int64")

    # 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 is only READ from here on (never mutated), so we can alias
    # instead of copying -- this avoids a second full copy of the 7M-row frame.
    original_final_output = original_final
    _step(log, "Full output frame ready")

    # ---- 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))
    _step(log, "Wrote final_cities / final_zips / final_census_tracts / cluster_df")

    # ---- 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".
    _step(log, "Writing full output CSV...")
    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))
    _step(log, "Full output CSV written")

    # ---- 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(_summary_row(
                name, fips,
                records=len(gfull),
                subdivisions=int(gfull["LegalSubDivision"].count())
                             if "LegalSubDivision" in gfull.columns else 0,
                clusters=int(g["Cluster_ID"].nunique()) if "Cluster_ID" in g.columns else 0,
                undet_city=int(city_blank.sum()),
                city_assign=int((city_blank & city_fill).sum()),
                undet_zip=int(zip_blank.sum()),
                zip_assign=int((zip_blank & zip_fill).sum()),
                blank_census=int(tract_blank.sum()),
                census_assign=int((tract_blank & tract_fill).sum()),
            ))

    summary_df = pd.DataFrame(summary_rows, columns=SUMMARY_COLUMNS)
    _step(log, 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 (slow on huge states) -- optional / best-effort ------
    # Built from the CSVs already on disk (same as streaming), so it can be
    # skipped here and run separately after the individual files are secured.
    if build_master:
        _step(log, f"Building master ({MASTER_OUTPUT})...")
        outputs.append(build_master_bundle(out_dir, MASTER_OUTPUT))
        _step(log, "Master built")

    elapsed_min = (time.time() - t_start) / 60
    log.append(f"Done. {len(outputs)} files ready.")
    _step(log, f"DONE — {len(outputs)} files ready in {elapsed_min:.1f} min")
    log.append(f"Total processing time: {elapsed_min:.1f} minutes.")
    gc.collect()
    return "\n".join(log), outputs


# =====================================================================
# DATASET I/O  (for large inputs that time out on browser upload -> HTTP 408)
# ---------------------------------------------------------------------
# Instead of uploading a multi-GB file through the browser, put it in a private
# HF *dataset* repo and let the Space read it over HF's internal network. Outputs
# are written back to a dataset the same way (large downloads also 408 otherwise).
# Configure as Space Variables/Secrets:
#   INPUT_DATASET_REPO   e.g. "DDPLLeads/subdivision-inputs"
#   OUTPUT_DATASET_REPO  e.g. "DDPLLeads/subdivision-outputs"  (optional)
#   HF_TOKEN             a token with read on input + write on output
# =====================================================================
INPUT_DATASET_REPO = os.environ.get("INPUT_DATASET_REPO", "")
OUTPUT_DATASET_REPO = os.environ.get("OUTPUT_DATASET_REPO", "")


class _LocalFile:
    """Adapter so downloaded dataset files look like uploaded files to run_pipeline."""
    def __init__(self, path):
        self.name = str(path)


def list_dataset_inputs(repo_id, token=None):
    from huggingface_hub import list_repo_files
    token = token or os.environ.get("HF_TOKEN")
    files = list_repo_files(repo_id, repo_type="dataset", token=token)
    return sorted(f for f in files if f.lower().endswith((".csv", ".xlsx", ".xls")))


def _cleanup_datasets(api, input_repo, output_repo, processed_inputs, current_stamp):
    """Keep the datasets small so the private-storage quota is never hit. Runs
    ONLY after results are safely uploaded. Returns a short log string.
    Steps (each guarded by its config flag):
      1. delete the input file(s) we just processed from the input dataset
      2. keep only the newest KEEP_LATEST_OUTPUT_RUNS runs/ folders in the output dataset
      3. squash both datasets' git history -- REQUIRED to actually reclaim bytes,
         since a plain delete leaves the old versions in history."""
    from huggingface_hub import list_repo_files
    notes = []
    squashed_any = False

    # 1) delete processed inputs
    if DELETE_INPUT_AFTER_RUN and input_repo and processed_inputs:
        deleted = 0
        for fn in processed_inputs:
            try:
                api.delete_file(path_in_repo=fn, repo_id=input_repo,
                                repo_type="dataset",
                                commit_message=f"Auto-cleanup: remove processed input {fn}")
                deleted += 1
            except Exception as e:
                notes.append(f"  (could not delete input '{fn}': {e})")
        if deleted:
            notes.append(f"  Deleted {deleted} processed input file(s) from '{input_repo}'.")

    # 2) prune old run folders in the output dataset -- keep the newest
    #    KEEP_LATEST_OUTPUT_RUNS folders TOTAL, and the run we just wrote always
    #    counts as one of them (so it is never deleted).
    if KEEP_LATEST_OUTPUT_RUNS and output_repo:
        try:
            files = list_repo_files(output_repo, repo_type="dataset")
            stamps = sorted({p.split("/")[1] for p in files
                             if p.startswith("runs/") and len(p.split("/")) > 2})
            others = [s for s in stamps if s != current_stamp]   # oldest..newest
            keep_others = others[-(KEEP_LATEST_OUTPUT_RUNS - 1):] if KEEP_LATEST_OUTPUT_RUNS > 1 else []
            old = [s for s in others if s not in keep_others]
            for s in old:
                try:
                    api.delete_folder(path_in_repo=f"runs/{s}", repo_id=output_repo,
                                      repo_type="dataset",
                                      commit_message=f"Auto-cleanup: remove old run {s}")
                except Exception as e:
                    notes.append(f"  (could not delete old run '{s}': {e})")
            if old:
                notes.append(f"  Pruned {len(old)} older run folder(s) from '{output_repo}'.")
        except Exception as e:
            notes.append(f"  (could not list runs for pruning: {e})")

    # 3) squash history to actually reclaim the storage
    if SQUASH_HISTORY:
        for repo in filter(None, {input_repo, output_repo}):
            try:
                api.super_squash_history(repo_id=repo, repo_type="dataset")
                squashed_any = True
            except Exception as e:
                notes.append(f"  (could not squash history for '{repo}': {e})")
        if squashed_any:
            notes.append("  Squashed dataset history -> storage reclaimed.")

    return ("\n\nAuto-cleanup:\n" + "\n".join(notes)) if notes else ""


def run_from_dataset(input_repo, output_repo):
    """Download every supported file from input_repo, run the pipeline, and (if
    output_repo is set) upload the results back to that dataset. Returns
    (log_text, output_paths) just like run_pipeline, so it wires to the same UI."""
    token = os.environ.get("HF_TOKEN")
    input_repo = (input_repo or "").strip()
    output_repo = (output_repo or "").strip()
    if not input_repo:
        return "Set an input dataset repo (or the INPUT_DATASET_REPO variable).", []
    if not token:
        return "An HF_TOKEN secret is required to read/write datasets.", []

    from huggingface_hub import hf_hub_download, HfApi
    try:
        filenames = list_dataset_inputs(input_repo, token)
    except Exception as e:
        return f"Could not list dataset '{input_repo}': {e}", []
    if not filenames:
        return f"No .csv/.xlsx/.xls files found in dataset '{input_repo}'.", []

    local = []
    for fn in filenames:
        try:
            local.append(_LocalFile(
                hf_hub_download(repo_id=input_repo, filename=fn, repo_type="dataset", token=token)))
        except Exception as e:
            return f"Failed to download '{fn}' from '{input_repo}': {e}", []

    # Process WITHOUT the master first, so the individual files exist quickly.
    log, outputs = process(local, None, build_master=False)

    stamp = time.strftime("%Y%m%d_%H%M%S")
    dest = f"https://huggingface.co/datasets/{output_repo}/tree/main/runs/{stamp}" if output_repo else ""

    def _upload(paths, api):
        for p in paths:
            api.upload_file(path_or_fileobj=p, path_in_repo=f"runs/{stamp}/{Path(p).name}",
                            repo_id=output_repo, repo_type="dataset")

    api = None
    core_uploaded = False
    if output_repo and outputs:
        try:
            api = HfApi(token=token)
            _upload(outputs, api)                       # <-- SECURE the CSVs first
            core_uploaded = True
            log += (f"\nUploaded {len(outputs)} core file(s) to '{output_repo}' under "
                    f"runs/{stamp}/ BEFORE building the master. If the master step "
                    f"fails, these are already saved.\n{dest}")
        except Exception as e:
            log += f"\nWARNING: processed OK but uploading core files failed: {e}"

    # Master last / best-effort. If it dies on a huge state, the files above survive.
    if outputs:
        try:
            master = build_master_bundle(Path(outputs[0]).parent, MASTER_OUTPUT)
            outputs.append(master)
            if api is not None:
                _upload([master], api)
                log += f"\nMaster built and uploaded: {Path(master).name}"
            else:
                log += f"\nMaster built: {Path(master).name}"
        except Exception as e:
            log += (f"\nMaster build/upload FAILED -- the individual files are already "
                    f"saved in the dataset, so nothing is lost: {e}")

    # Auto-cleanup runs ONLY once results are safely uploaded, so nothing is lost.
    if api is not None and core_uploaded:
        try:
            log += _cleanup_datasets(api, input_repo, output_repo, filenames, stamp)
        except Exception as e:
            log += f"\nAuto-cleanup skipped (non-fatal): {e}"

    # A clean, clickable list of every file in the dataset, so nobody has to hunt.
    if output_repo and outputs:
        base = f"https://huggingface.co/datasets/{output_repo}/blob/main/runs/{stamp}"
        links = "\n".join(f"  - {Path(p).name}: {base}/{Path(p).name}" for p in outputs)
        log += (f"\n\n================ RESULTS (click to open) ================\n"
                f"Folder: {dest}\n{links}\n"
                f"=========================================================")
    return log, outputs


# =====================================================================
# PER-COUNTY STREAMING  (bounded memory for state-sized files)
# =====================================================================
_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",
]


def process_county(cdf, wl, fips_col, cluster_offset,
                   blank_cities_not_in_list, cluster_threshold, replace_threshold):
    """Full per-county transform on ONE county's raw rows. Mirrors run_pipeline
    exactly, scoped to a single county. Returns a dict of that county's outputs
    plus how many clusters it produced (to continue global Cluster_ID numbering).
    Cluster_IDs are offset by `cluster_offset` so, processing counties in sorted
    order, the numbering matches the whole-dataset ("memory") path."""
    ignore = []
    ensure_flag_columns(cdf, None)
    fips_col, _r = ensure_combined_county(cdf, fips_col, "", wl, None)
    cdf, audit = fips_validate(cdf, blank_cities_not_in_list, ignore, wl=wl, fips_col=fips_col)

    for col in ["DDPPropertyCityFinal", "DDPPropertyZIPFinal",
                "DDPRevisedCensusTract", "LegalSubDivision"]:
        if col in cdf.columns:
            cdf[col] = cdf[col].replace(["UNDETERMINED CITY", "UNDETERMINED ZIP", "NULL"], np.nan)

    if fips_col != "County":
        if "County" in cdf.columns:
            cdf = cdf.drop(columns=["County"])
        cdf = cdf.rename(columns={fips_col: "County"})

    city_df = cdf.loc[cdf["DDPPropertyCityFinal"].notna()].copy()
    zip_df = cdf.loc[cdf["DDPPropertyZIPFinal"].notna()].copy()
    census_tracts_df = cdf.loc[cdf["DDPRevisedCensusTract"].notna()].copy()
    original = cdf.loc[cdf["LegalSubDivision"].notna()].copy()

    cluster_df = build_clusters(city_df, zip_df, census_tracts_df, cluster_threshold)
    if len(cluster_df):
        cluster_df["Cluster_ID"] = cluster_df["Cluster_ID"] + cluster_offset
    n_clusters = int(cluster_df["Cluster_ID"].nunique()) if len(cluster_df) else 0

    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"})

    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"])))
    original_final["Cluster_ID"] = original_final["Cluster_ID"].astype("Int64")
    original_final["Replace with Zip Codes"] = original_final["Replace with Zip Codes"].apply(clean_zip)

    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]

    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()
    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"]).copy()
    final_zips = final_zips.drop(columns=["DDPPropertyZIPFinalToRemove"]).copy()
    final_ct = final_ct.drop(columns=["DDPRevisedCensusTractToRemove"]).copy()
    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()
    for d in (final_cities, final_zips, final_ct):
        for bc in ("ComesFromMtg2", "IsFlagged"):
            if bc in d.columns:
                d[bc] = boolify(d[bc])

    # ---- summary row for this county (totals from full county frame) ----
    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)

    label = cdf["County"].dropna().iloc[0] if ("County" in cdf.columns and cdf["County"].notna().any()) else ""
    raw = str(label)
    parts = raw.split("-")
    fips = parts[0].strip()
    name = parts[1].strip().title() if len(parts) > 1 else raw.strip().title()
    g = original_final
    city_blank, zip_blank, tract_blank = (_blank(g, "DDPPropertyCityFinal"),
                                          _blank(g, "DDPPropertyZIPFinal"), _blank(g, "DDPRevisedCensusTract"))
    city_fill, zip_fill, tract_fill = (_fill(g, "Replace with Cities"),
                                       _fill(g, "Replace with Zip Codes"), _fill(g, "Replace with Census Tracts"))
    summary_row = _summary_row(
        name, fips,
        records=len(cdf),
        subdivisions=int(cdf["LegalSubDivision"].count()) if "LegalSubDivision" in cdf.columns else 0,
        clusters=n_clusters,
        undet_city=int(city_blank.sum()),
        city_assign=int((city_blank & city_fill).sum()),
        undet_zip=int(zip_blank.sum()),
        zip_assign=int((zip_blank & zip_fill).sum()),
        blank_census=int(tract_blank.sum()),
        census_assign=int((tract_blank & tract_fill).sum()),
    )

    return {
        "validated": cdf, "audit": audit, "cluster_df": cluster_df,
        "output_df": original_final, "final_cities": final_cities,
        "final_zips": final_zips, "final_ct": final_ct,
        "summary_row": summary_row, "n_clusters": n_clusters,
    }


class _RollingCsv:
    """Append rows to CSV part files, rolling to a new part every row_limit rows."""
    def __init__(self, out_dir, base, row_limit=1_000_000):
        self.out_dir, self.base, self.row_limit = out_dir, base, row_limit
        self.part, self.count, self.header = 0, 0, False
        self.paths = []
        self._new()

    def _new(self):
        self.part += 1
        self.count, self.header = 0, False
        self.path = self.out_dir / f"{self.base}_part_{self.part}.csv"
        self.paths.append(self.path)

    def write(self, dfx):
        i, n = 0, len(dfx)
        while i < n:
            room = self.row_limit - self.count
            take = dfx.iloc[i:i + room]
            take.to_csv(self.path, mode="a", header=not self.header, index=False)
            self.header = True
            self.count += len(take)
            i += len(take)
            if self.count >= self.row_limit and i < n:
                self._new()

    def finalize(self):
        # If everything fit in one part, drop the "_part_1" suffix to match memory mode.
        if self.part == 1 and self.paths[0].exists():
            final = self.out_dir / f"{self.base}.csv"
            os.replace(self.paths[0], final)
            return [str(final)]
        return [str(p) for p in self.paths if p.exists()]


def _split_by_county(uploaded_files, out_dir, log, chunksize=STREAM_CHUNK_ROWS):
    """Stream the input(s) in chunks and route rows to one temp CSV per county.
    Never holds more than one chunk in memory. Returns (fips_col, reason,
    [(county_label, path), ...] sorted by county)."""
    cdir = out_dir / "_by_county"
    cdir.mkdir(parents=True, exist_ok=True)

    # detect the FIPS-county column from a small sample of the first file
    first = uploaded_files[0].name
    if str(first).lower().endswith(".csv"):
        with _CleanReader(first) as _fh:
            sample = pd.read_csv(_fh, nrows=50_000, low_memory=False,
                                 sep=detect_delimiter(first), on_bad_lines="skip")
    else:
        sample = pd.read_excel(first)
    fips_col, reason = detect_fips_county_column(sample)
    sample_cols = list(sample.columns)
    del sample
    if fips_col is None:
        return None, reason, [], sample_cols

    county_paths = {}       # county label -> temp path
    counter = [0]

    def path_for(county):
        if county not in county_paths:
            counter[0] += 1
            county_paths[county] = cdir / f"county_{counter[0]:05d}.csv"
        return county_paths[county]

    def route(chunk):
        # Key on the NORMALIZED 5-digit FIPS, not the raw text: chunked reading can
        # render the same code as '48001' in one chunk and '48001.0' in another
        # (float, from a chunk containing a blank), which would otherwise split a
        # county in two and cluster the halves separately.
        keys = chunk[fips_col].map(_norm_fips).fillna("UNKNOWN")
        for county, idx in chunk.groupby(keys).groups.items():
            p = path_for(county)
            chunk.loc[idx].to_csv(p, mode="a", header=not p.exists(), index=False)

    total = 0
    for f in uploaded_files:
        nm = f.name
        if str(nm).lower().endswith(".csv"):
            sep = detect_delimiter(nm)
            _step(log, f"Reading {Path(nm).name} (delimiter {sep!r})")
            with _CleanReader(nm) as _fh:
                for chunk in pd.read_csv(_fh, chunksize=chunksize, low_memory=False,
                                         sep=sep, on_bad_lines="skip"):
                    route(chunk)
                    total += len(chunk)
                    _step(log, f"Split: routed {total:,} rows into {len(county_paths)} counties")
        else:
            chunk = pd.read_excel(nm)
            route(chunk)
            total += len(chunk)
            _step(log, f"Split: routed {total:,} rows into {len(county_paths)} counties")

    ordered = sorted(county_paths.items(), key=lambda kv: str(kv[0]))
    return fips_col, reason, ordered, sample_cols


def run_streaming(uploaded_files, reference_file, blank_cities_not_in_list,
                  filter_dominant_county, cluster_threshold, replace_threshold,
                  build_master=True):
    """Per-county streaming pipeline: bounded memory, identical results."""
    t_start = time.time()
    log = []
    if not uploaded_files:
        return "Please upload at least one file.", []
    out_dir = Path(tempfile.mkdtemp(prefix="subdiv_"))
    outputs = []

    try:
        wl = build_whitelist_from_path(reference_file.name) if reference_file is not None else load_whitelist()
        _step(log, "Loaded safety list" + (" (uploaded override)" if reference_file is not None else ""))
    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}", []

    _step(log, "Scanning inputs for malformed rows...")
    bad_files = scan_inputs_for_bad_rows(uploaded_files, out_dir, log)

    _step(log, "Splitting input by county (streaming)...")
    fips_col, reason, counties, sample_cols = _split_by_county(uploaded_files, out_dir, log)
    if fips_col is None:
        return ("\n".join(log) + "\n\nCOLUMN ERROR: could not find a FIPS-county column. "
                "Expected values like '05045-FAULKNER'."), []
    _step(log, f"FIPS-county column: '{fips_col}' ({reason}) | {len(counties)} counties")
    _missing_flags = [c for c in ("IsFlagged", "ComesFromMtg2") if c not in sample_cols]
    if _missing_flags:
        _step(log, f"Adding missing flag column(s) {', '.join(_missing_flags)} = FALSE "
                   f"(required for the Backoffice upload)")

    # combined output writers
    audit_path = out_dir / "FIPS_Validation_Audit_ChangedRecords.csv"
    raw_path = out_dir / "RawFile_FIPSValidated.csv"
    fc_path = out_dir / "final_cities.csv"
    fz_path = out_dir / "final_zips.csv"
    fct_path = out_dir / "final_census_tracts.csv"
    clu_path = out_dir / "cluster_df.csv"
    out_roll = _RollingCsv(out_dir,
                           f"OutputFile_{replace_threshold}Threshold_{cluster_threshold}ClustersRapidFuzz",
                           row_limit=1_000_000)

    def _append(dfx, path):
        if len(dfx) == 0 and Path(path).exists():
            return
        dfx.to_csv(path, mode="a", header=not Path(path).exists(), index=False)

    summary_rows = []
    cluster_offset = 0
    any_audit = False
    for i, (label, cpath) in enumerate(counties, start=1):
        cdf = pd.read_csv(cpath, low_memory=False)
        res = process_county(cdf, wl, fips_col, cluster_offset,
                             blank_cities_not_in_list, cluster_threshold, replace_threshold)
        cluster_offset += res["n_clusters"]

        if WRITE_RAWFILE:
            _append(res["validated"], raw_path)
        if len(res["audit"]) > 0:
            _append(res["audit"], audit_path); any_audit = True
        _append(res["final_cities"], fc_path)
        _append(res["final_zips"], fz_path)
        _append(res["final_ct"], fct_path)
        if len(res["cluster_df"]):
            _append(res["cluster_df"], clu_path)
        out_roll.write(res["output_df"])
        summary_rows.append(res["summary_row"])

        del cdf, res
        gc.collect()
        _step(log, f"County {i}/{len(counties)} done ({label})")

    # finalize combined files
    for p in [audit_path, raw_path, fc_path, fz_path, fct_path, clu_path]:
        if p.exists():
            outputs.append(str(p))
    output_parts = out_roll.finalize()
    outputs.extend(output_parts)
    outputs.extend(bad_files)   # malformed-row files for review

    summary_df = pd.DataFrame(summary_rows, columns=SUMMARY_COLUMNS)
    summary_path = out_dir / "Summary.csv"
    summary_df.to_csv(summary_path, index=False)
    outputs.append(str(summary_path))
    _step(log, f"Summary: {len(summary_df)} county row(s)")

    # ---- Master (slow on huge states) -- optional / best-effort ------
    if build_master:
        _step(log, f"Building master ({MASTER_OUTPUT})...")
        outputs.append(build_master_bundle(out_dir, MASTER_OUTPUT))
        _step(log, "Master built")

    elapsed_min = (time.time() - t_start) / 60
    log.append(f"Done. {len(outputs)} files ready.")
    _step(log, f"DONE — {len(outputs)} files ready in {elapsed_min:.1f} min")
    log.append(f"Total processing time: {elapsed_min:.1f} minutes.")
    gc.collect()
    return "\n".join(log), outputs


def _write_master_streaming(path, output_parts, audit_path, clu_path,
                            fc_path, fz_path, fct_path, summary_df, row_limit=1_048_575):
    """Build Master_Dataset.xlsx. Any result too big for a single Excel sheet
    (the full output on a big state, sometimes cluster_df) is NOT written into the
    workbook -- writing millions of Excel cells row-by-row is extremely slow and
    such a sheet isn't usable in Excel anyway. Those are left as their CSV files
    and listed on a READ_ME sheet. Small results are written normally."""
    try:
        import xlsxwriter
    except Exception:
        xlsxwriter = None

    def _read(p):
        return pd.read_csv(p, low_memory=False) if Path(p).exists() else pd.DataFrame()

    def _count_rows(p):
        if not Path(p).exists():
            return 0
        with open(p, "r", encoding="utf-8", errors="replace") as fh:
            return max(sum(1 for _ in fh) - 1, 0)   # minus header

    notes = []
    if xlsxwriter is None:
        engine_kwargs = {}
        wb = None
        writer = pd.ExcelWriter(path, engine="openpyxl")
    else:
        wb = xlsxwriter.Workbook(str(path), {"constant_memory": True, "nan_inf_to_errors": True})
        writer = None

    def _write_df(name, dfx):
        if wb is not None:
            ws = wb.add_worksheet(str(name)[:31])
            ws.write_row(0, 0, [str(c) for c in dfx.columns])
            clean = dfx.where(pd.notna(dfx), None)
            for r, rowvals in enumerate(clean.itertuples(index=False, name=None), start=1):
                ws.write_row(r, 0, rowvals)
        else:
            dfx.to_excel(writer, sheet_name=str(name)[:31], index=False)

    def _sheet_or_note(name, path_, csv_label):
        n = _count_rows(path_)
        if n == 0:
            _write_df(name, _read(path_))
        elif n <= row_limit:
            _write_df(name, _read(path_))
        else:
            notes.append(f"{name}: {n:,} rows -- too many for one Excel sheet. See {csv_label}.")

    # Full output: skip if it spans more than one 1M-row part (huge states).
    total_output = sum(_count_rows(p) for p in output_parts)
    if len(output_parts) == 1 and total_output <= row_limit:
        _write_df("Output_70_95_RapidFuzz", _read(output_parts[0]))
    elif output_parts:
        notes.append(f"Output_70_95_RapidFuzz: {total_output:,} rows across "
                     f"{len(output_parts)} files -- too many for Excel. See the "
                     f"OutputFile_...part_N.csv files.")

    _sheet_or_note("cluster_df", clu_path, "cluster_df.csv")
    _sheet_or_note("ValidationAuditResults", audit_path, "FIPS_Validation_Audit_ChangedRecords.csv")
    _sheet_or_note("final_cities", fc_path, "final_cities.csv")
    _sheet_or_note("final_zips", fz_path, "final_zips.csv")
    _sheet_or_note("final_census_tracts", fct_path, "final_census_tracts.csv")
    _write_df("Summary", summary_df)

    if notes:
        header = ("Some results are too large for an Excel sheet (Excel caps a sheet "
                  "at 1,048,576 rows) and are provided as CSV files alongside this "
                  "workbook:")
        _write_df("READ_ME", pd.DataFrame({"Note": [header] + notes}))

    if wb is not None:
        wb.close()
    else:
        writer.close()


def build_master_bundle(out_dir, master_output):
    """Build the master (zip or xlsx) from the CSV outputs already on disk in
    out_dir. Kept SEPARATE from the pipeline so callers can secure the fast CSVs
    first and run this slow step last / best-effort -- if it times out on a huge
    state, the individual files are already saved. Returns the master path."""
    out_dir = Path(out_dir)
    output_parts = [str(p) for p in sorted(out_dir.glob("OutputFile_*.csv"))]
    audit_path = out_dir / "FIPS_Validation_Audit_ChangedRecords.csv"
    clu_path = out_dir / "cluster_df.csv"
    fc_path = out_dir / "final_cities.csv"
    fz_path = out_dir / "final_zips.csv"
    fct_path = out_dir / "final_census_tracts.csv"
    summary_path = out_dir / "Summary.csv"

    if master_output == "csv":
        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 += [Path(p) for p in output_parts]
        with zipfile.ZipFile(master_path, "w", zipfile.ZIP_DEFLATED) as zf:
            for p in members:
                zf.write(p, arcname=Path(p).name)
    else:
        master_path = out_dir / "Master_Dataset.xlsx"
        summary_df = (pd.read_csv(summary_path, low_memory=False)
                      if summary_path.exists() else pd.DataFrame())
        _write_master_streaming(master_path, output_parts, audit_path, clu_path,
                                fc_path, fz_path, fct_path, summary_df)
    return str(master_path)


def process(uploaded_files, reference_file, build_master=True):
    """Dispatcher used by the UI: streaming or in-memory per PROCESS_MODE.
    build_master=False returns just the individual files (no master), so a caller
    can save/upload those first and build the slow master separately."""
    if PROCESS_MODE == "stream":
        return run_streaming(uploaded_files, reference_file, STRICT_BLANK_CITIES,
                             FILTER_DOMINANT_COUNTY, CLUSTER_THRESHOLD, REPLACE_THRESHOLD,
                             build_master=build_master)
    return run_pipeline(uploaded_files, reference_file, STRICT_BLANK_CITIES,
                        FILTER_DOMINANT_COUNTY, CLUSTER_THRESHOLD, REPLACE_THRESHOLD,
                        build_master=build_master)


# =====================================================================
# 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 Files</div>')
            files_in = gr.File(file_count="multiple", file_types=[".xlsx", ".xls", ".csv"],
                               label="XLSX, XLS, or CSV — upload several and they're merged",
                               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="XLSX or CSV", show_label=True)

            run_btn = gr.Button("Run pipeline", variant="primary")

            gr.HTML('<div class="step-label">Large files: process from a HF Dataset '
                    '(no upload — avoids the upload timeout)</div>')
            ds_in = gr.Textbox(value=INPUT_DATASET_REPO, label="Input dataset repo",
                               placeholder="user/dataset-name")
            ds_out = gr.Textbox(value=OUTPUT_DATASET_REPO, label="Output dataset repo (optional)",
                                placeholder="user/dataset-name")
            ds_btn = gr.Button("Run from dataset")

        # -------- 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">Upload your file(s) on the sidebar and click '
                '<b>Run pipeline</b>. For files too large to upload (multi-GB / full-state), '
                'put them in a private HF Dataset and use <b>Run from dataset</b> instead — '
                'the results are written back to your output dataset.</div>'
            )
            status = gr.Textbox(label="Log", lines=14)
            files_out = gr.File(label="Output files", file_count="multiple")

    def _run(files, reference):
        # Produce the individual files first and show them for download, THEN build
        # the (slow-on-huge-files) master. If the master step is interrupted, the
        # files below are already downloadable.
        log, outputs = process(files, reference, build_master=False)
        yield log + "\n\nFiles below are ready to download. Building master...", outputs
        if outputs:
            try:
                master = build_master_bundle(Path(outputs[0]).parent, MASTER_OUTPUT)
                outputs = outputs + [master]
                yield log + f"\n\nMaster ready: {Path(master).name}", outputs
            except Exception as e:
                yield log + f"\n\nFiles are ready. Master build failed: {e}", outputs
        else:
            yield log, outputs

    run_btn.click(
        _run,
        inputs=[files_in, reference_in],
        outputs=[status, files_out],
    )
    ds_btn.click(
        run_from_dataset,
        inputs=[ds_in, ds_out],
        outputs=[status, files_out],
    )

if __name__ == "__main__":
    # max_file_size=None -> no per-file upload cap imposed by the app itself.
    # (Very large uploads are still limited by HF's proxy/RAM -- prefer splitting
    # the input into several files, which this app merges, or run big jobs in Colab.)
    demo.launch(max_file_size=None)