FIPS vs Contiguous County List logic added for FIPS only column

app.py

@@ -56,7 +56,7 @@ FILTER_DOMINANT_COUNTY = False  # process ALL counties (multi-county / full-stat
 #   "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 = "csv"
+MASTER_OUTPUT = "excel"
 
 # How records are processed:
 #   "stream" -> ONE COUNTY AT A TIME. The input is split into per-county temp
@@ -118,7 +118,7 @@ _SAFETY_CACHE: dict | None = None
 
 
 def build_whitelist_from_path(path):
-    """Parse a safety-list file (xlsx or csv) into {pairs, cities, rows}."""
+    """Parse a safety-list file (xlsx or csv) into {pairs, cities, rows, fips_name}."""
     if str(path).lower().endswith(".csv"):
         safety_df = pd.read_csv(path)
     else:
@@ -126,17 +126,31 @@ def build_whitelist_from_path(path):
     acceptable_cols = [c for c in safety_df.columns if str(c).startswith("Acceptable FIPS")]
 
     valid_pairs, valid_cities = set(), set()
+    # FIPS -> county name, from the positional pairing of "Acceptable counties"
+    # (comma-separated) against Acceptable FIPS 1..N. Used to synthesize the
+    # "<FIPS>-<NAME>" column when source data has only bare FIPS codes.
+    fips_name_votes = defaultdict(lambda: defaultdict(int))
+    has_names = "Acceptable counties" in safety_df.columns
     for _, r in safety_df.iterrows():
         c = _norm_city(r["City Name"])
-        if c is None:
-            continue
-        valid_cities.add(c)
-        for col in acceptable_cols:
-            f = _norm_fips(r[col])
-            if f:
-                valid_pairs.add(c + "||" + f)
-
-    return {"pairs": valid_pairs, "cities": valid_cities, "rows": len(safety_df)}
+        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():
@@ -175,7 +189,7 @@ def _norm_fips(x):
 # 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", "DocumentRecordingCountyFIPS"]
+FALLBACK_FIPS_HEADERS = ["** REC FIPS-COUNTY NAME", "** REC FIPS-COUNTY", "County"]
 
 
 def detect_fips_county_column(df, sample_size=1000, min_score=0.5):
@@ -209,6 +223,41 @@ def detect_fips_county_column(df, sample_size=1000, min_score=0.5):
     return None, "not found"
 
 
+def _is_combined_fips(series, sample=1000):
+    """True if the column already holds '<FIPS>-<NAME>' values (vs bare codes)."""
+    s = series.dropna().astype(str).str.strip().head(sample)
+    if len(s) == 0:
+        return False
+    try:
+        return float(s.str.match(FIPS_COUNTY_PATTERN).mean()) >= 0.5
+    except Exception:
+        return False
+
+
+def ensure_combined_county(df, fips_col, reason, wl, log=None):
+    """Guarantee a '<FIPS>-<NAME>' county column. If `fips_col` already holds
+    combined values (PBIX-processed data), keep it. If it holds BARE FIPS codes
+    (raw SQL, e.g. DocumentRecordingCountyFIPS = '48201'), synthesize a 'County'
+    column '48201-HARRIS' using the FIPS->name map from the safety list, so the
+    rest of the pipeline behaves exactly as it does for the combined-column states.
+    Returns (column_name_to_use, reason)."""
+    if fips_col is None:
+        return fips_col, reason
+    if _is_combined_fips(df[fips_col]):
+        return fips_col, reason
+    fips_name = wl.get("fips_name", {}) if isinstance(wl, dict) else {}
+    if not fips_name:
+        return fips_col, reason  # no map -> leave bare (still groups/validates by FIPS)
+    codes = df[fips_col].map(_norm_fips)          # 5-digit string or None
+    names = codes.map(lambda c: fips_name.get(c, "") if c else "")
+    df["County"] = [f"{c}-{n}" if c else "" for c, n in zip(codes.fillna(""), names)]
+    if log is not None:
+        unresolved = int(((names == "") & codes.notna()).sum())
+        log.append(f"Built county names from bare FIPS column '{fips_col}' via the "
+                   f"safety list ({unresolved:,} rows had no name match).")
+    return "County", f"synthesized '<FIPS>-<NAME>' from bare FIPS column '{fips_col}'"
+
+
 # =====================================================================
 # SELF-TEST  (startup health check — same logic as the notebook's cell)
 # =====================================================================
@@ -616,6 +665,9 @@ def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
                 "\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...")
@@ -994,6 +1046,7 @@ def process_county(cdf, wl, fips_col, cluster_offset,
     Cluster_IDs are offset by `cluster_offset` so, processing counties in sorted
     order, the numbering matches the whole-dataset ("memory") path."""
     ignore = []
+    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",