""" 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_.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 # "-" 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 '-' 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 '-' 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 '-' 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 _1, _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"") with gr.Row(): # -------- LEFT SIDEBAR: uploads --------------------------------- with gr.Column(scale=1, elem_id="sidebar", min_width=270): gr.Markdown("### Upload Files") gr.HTML('
1. Upload Files
') 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('
2. Upload Reference File (Optional)
') 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('
Large files: process from a HF Dataset ' '(no upload — avoids the upload timeout)
') 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('
Subdivision Cluster & FIPS Validation
') gr.HTML( '
Upload your file(s) on the sidebar and click ' 'Run pipeline. For files too large to upload (multi-GB / full-state), ' 'put them in a private HF Dataset and use Run from dataset instead — ' 'the results are written back to your output dataset.
' ) 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)