Added logic to not generate RAW file and limit Excel Master File to certain super large files

app.py

@@ -69,6 +69,11 @@ MASTER_OUTPUT = "excel"
 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
+
 
 def _mem_mb():
     """Current process memory (RSS) in MB. psutil if available, else peak RSS."""
@@ -743,10 +748,11 @@ def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
         audit.to_csv(ap, index=False)
         outputs.append(str(ap))
 
-    rp = out_dir / "RawFile_FIPSValidated.csv"
-    df.to_csv(rp, index=False)
-    outputs.append(str(rp))
-    _step(log, "Wrote RawFile_FIPSValidated.csv")
+    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",
@@ -1085,6 +1091,14 @@ def run_from_dataset(input_repo, output_repo):
         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}")
+
+    # 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
 
 
@@ -1298,7 +1312,11 @@ def _split_by_county(uploaded_files, out_dir, log, chunksize=STREAM_CHUNK_ROWS):
         return county_paths[county]
 
     def route(chunk):
-        keys = chunk[fips_col].astype(str).str.strip()
+        # 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)
@@ -1378,7 +1396,8 @@ def run_streaming(uploaded_files, reference_file, blank_cities_not_in_list,
                              blank_cities_not_in_list, cluster_threshold, replace_threshold)
         cluster_offset += res["n_clusters"]
 
-        _append(res["validated"], raw_path)
+        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)
@@ -1426,8 +1445,11 @@ def run_streaming(uploaded_files, reference_file, blank_cities_not_in_list,
 
 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 in low memory: stream the Output sheet from the
-    combined output CSV(s) in chunks; load the (small) side sheets fully."""
+    """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:
@@ -1436,57 +1458,66 @@ def _write_master_streaming(path, output_parts, audit_path, clu_path,
     def _read(p):
         return pd.read_csv(p, low_memory=False) if Path(p).exists() else pd.DataFrame()
 
-    if xlsxwriter is None:
-        # openpyxl fallback (higher memory): load Output fully.
-        frames = []
-        for p in output_parts:
-            frames.append(pd.read_csv(p, low_memory=False))
-        output_df = pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()
-        write_master_workbook(path, [
-            ("Output_70_95_RapidFuzz", output_df),
-            ("ValidationAuditResults", _read(audit_path)),
-            ("cluster_df", _read(clu_path)),
-            ("final_cities", _read(fc_path)),
-            ("final_zips", _read(fz_path)),
-            ("final_census_tracts", _read(fct_path)),
-            ("Summary", summary_df),
-        ], row_limit=row_limit)
-        return
+    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
 
-    wb = xlsxwriter.Workbook(str(path), {"constant_memory": True, "nan_inf_to_errors": True})
+    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):
-        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)
-
-    # Output sheet(s): stream from the combined CSV, split at row_limit
-    sheet_idx, row_in_sheet, ws, header = 0, 0, None, None
-    for p in output_parts:
-        for chunk in pd.read_csv(p, chunksize=200_000, low_memory=False):
-            if header is None:
-                header = [str(c) for c in chunk.columns]
-            clean = chunk.where(pd.notna(chunk), None)
-            for rowvals in clean.itertuples(index=False, name=None):
-                if ws is None or row_in_sheet >= row_limit:
-                    sheet_idx += 1
-                    ws = wb.add_worksheet(f"Output_70_95_RapidFuzz_{sheet_idx}"[:31])
-                    ws.write_row(0, 0, header)
-                    row_in_sheet = 0
-                row_in_sheet += 1
-                ws.write_row(row_in_sheet, 0, rowvals)
-    if ws is None:  # no output rows at all
-        wb.add_worksheet("Output_70_95_RapidFuzz").write_row(0, 0, header or ["(empty)"])
-
-    _write_df("ValidationAuditResults", _read(audit_path))
-    _write_df("cluster_df", _read(clu_path))
-    _write_df("final_cities", _read(fc_path))
-    _write_df("final_zips", _read(fz_path))
-    _write_df("final_census_tracts", _read(fct_path))
+        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)
-    wb.close()
+
+    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):