Added 2 more delimiters to delimiters sniffing code, and logic to skip, count and add to a csv bad rows with more than allowed columns

app.py

@@ -122,11 +122,13 @@ def get_safety_list_path() -> str:
 _SAFETY_CACHE: dict | None = None
 
 
-def detect_delimiter(path, candidates=(",", "|", "^", "\t", ";")):
+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 ','. 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."""
+    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 = []
@@ -155,12 +157,62 @@ def detect_delimiter(path, candidates=(",", "|", "^", "\t", ";")):
 
 
 def _read_csv_auto(path, **kwargs):
-    """pd.read_csv with the delimiter auto-detected (unless caller passed sep)."""
+    """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")
     return pd.read_csv(path, **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 = 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 = 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"):
@@ -700,6 +752,10 @@ def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
     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:
@@ -707,7 +763,8 @@ def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
         try:
             if str(name).lower().endswith(".csv"):
                 sep = detect_delimiter(name)
-                dfs.append(pd.read_csv(name, low_memory=False, sep=sep))
+                dfs.append(pd.read_csv(name, 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:
@@ -1293,7 +1350,8 @@ def _split_by_county(uploaded_files, out_dir, log, chunksize=STREAM_CHUNK_ROWS):
     # detect the FIPS-county column from a small sample of the first file
     first = uploaded_files[0].name
     if str(first).lower().endswith(".csv"):
-        sample = pd.read_csv(first, nrows=50_000, low_memory=False, sep=detect_delimiter(first))
+        sample = pd.read_csv(first, 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)
@@ -1327,7 +1385,8 @@ def _split_by_county(uploaded_files, out_dir, log, chunksize=STREAM_CHUNK_ROWS):
         if str(nm).lower().endswith(".csv"):
             sep = detect_delimiter(nm)
             _step(log, f"Reading {Path(nm).name} (delimiter {sep!r})")
-            for chunk in pd.read_csv(nm, chunksize=chunksize, low_memory=False, sep=sep):
+            for chunk in pd.read_csv(nm, 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")
@@ -1360,6 +1419,9 @@ def run_streaming(uploaded_files, reference_file, blank_cities_not_in_list,
     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:
@@ -1418,6 +1480,7 @@ def run_streaming(uploaded_files, reference_file, blank_cities_not_in_list,
             outputs.append(str(p))
     output_parts = out_roll.finalize()
     outputs.extend(output_parts)
+    outputs.extend(bad_files)   # malformed-row files for review
 
     summary_cols = ["County", "FIPS", "Total Records in County", "Total Subdivisions",
                     "Total Clusters Made", "Total Blank Cities", "Total Cities Assignment",