#!/usr/bin/env python3 # column_stats_to_json.py (updated Defendant handling: case-insensitive + wider suffix strip) import csv import json import argparse import re from pathlib import Path from collections import Counter, defaultdict from typing import List, Tuple, Dict, Any, Optional # NEW: broader, case-insensitive suffix matcher # Strips: ", Jr", ", Jr.", ", SR", ", Sr.", ", II", ", III", ..., ", X" (period optional, case-insensitive) SUFFIX_RE = re.compile( r",\s*(?:jr\.?|sr\.?|ii|iii|iv|v|vi|vii|viii|ix|x|[A-Za-z]{2}\.?)$", re.IGNORECASE ) def parse_ranges(spec: Optional[str]) -> List[Tuple[int, int, str]]: if not spec: return [] out: List[Tuple[int, int, str]] = [] for chunk in spec.split(","): chunk = chunk.strip() if not chunk: continue if "-" not in chunk: try: k = int(chunk) out.append((k, k, f"{k}-{k}")) continue except ValueError: raise ValueError(f"Invalid range token: '{chunk}'") a, b = chunk.split("-", 1) try: start = int(a.strip()); end = int(b.strip()) except ValueError: raise ValueError(f"Invalid range token: '{chunk}'") if end < start: start, end = end, start out.append((start, end, f"{start}-{end}")) return out def normalize(val: Any, case_insensitive: bool) -> str: s = "" if val is None else str(val) s = s.strip() if case_insensitive: s = s.lower() return s def load_csv(path: Path) -> Tuple[List[str], List[Dict[str, str]]]: with path.open("r", encoding="utf-8-sig", newline="") as f: reader = csv.DictReader(f) headers = reader.fieldnames or [] rows = [dict(r) for r in reader] return headers, rows def build_index_set(rows: List[Dict[str, str]], start: int, end: int) -> set: idx = set() for r in rows: try: i = int(str(r.get("Index", "")).strip()) except ValueError: continue if start <= i <= end: idx.add(i) return idx def filter_rows_by_index(rows: List[Dict[str, str]], idxset: set) -> List[Dict[str, str]]: out = [] for r in rows: try: i = int(str(r.get("Index", "")).strip()) except ValueError: continue if i in idxset: out.append(r) return out def normalize_defendant(raw: str) -> Tuple[str, Tuple[str, ...]]: """ Returns (representative_original, grouping_key). - Strip suffix like ", Jr." / ", II" (case-insensitive, period optional). - Tokenize; group by (first, last) if >=2 tokens, else by the single token. - Grouping key is ALWAYS case-insensitive. - Representative string is the cleaned original (first seen for that key). """ cleaned = raw.strip() cleaned = SUFFIX_RE.sub("", cleaned) # remove suffix cleaned = re.sub(r"\s+", " ", cleaned) # collapse spaces toks = cleaned.split() if cleaned else [] toks_key = [t.lower() for t in toks] # <- always ignore case if len(toks_key) >= 2: key = (toks_key[0], toks_key[-1]) # (first, last) elif len(toks_key) == 1: key = (toks_key[0],) else: key = tuple() return cleaned, key def compute_column_stats( headers: List[str], rows: List[Dict[str, str]], drop_empty: bool, case_insensitive_generic: bool ) -> Dict[str, Dict[str, Any]]: if not headers or headers[0] != "Index": raise ValueError('The first column must be named exactly "Index"') results: Dict[str, Dict[str, Any]] = {} other_cols = headers[1:] for col in other_cols: if col == "Defendant": counts: Dict[Tuple[str, ...], int] = defaultdict(int) representative: Dict[Tuple[str, ...], str] = {} total_considered = 0 for r in rows: raw = r.get(col, "") if raw is None: raw = "" raw = raw.strip() if raw == "" and drop_empty: continue total_considered += 1 if raw != "" or not drop_empty else 0 rep_str, key = normalize_defendant(raw) if key not in representative: representative[key] = rep_str counts[key] += 1 items = sorted(counts.items(), key=lambda kv: (-kv[1], representative.get(kv[0], ""))) values = [{"String": representative[k], "Count": v} for k, v in items] results[col] = { "Values": values, "Unique": len(counts), "TotalRowsConsidered": total_considered } else: counter: Counter = Counter() total_considered = 0 for r in rows: raw = r.get(col, "") s = normalize(raw, case_insensitive_generic) if s == "" and drop_empty: continue total_considered += 1 if s != "" or not drop_empty else 0 counter[s] += 1 items = sorted(counter.items(), key=lambda kv: (-kv[1], kv[0])) values = [{"String": k, "Count": v} for k, v in items] results[col] = { "Values": values, "Unique": len(counter), "TotalRowsConsidered": total_considered } return results def main(): ap = argparse.ArgumentParser(description="Per-column value counts over Index ranges → JSON (with robust 'Defendant' grouping).") ap.add_argument("input_csv", help="Path to input CSV (must have 'Index' as first header).") ap.add_argument("output_json", help="Path to output JSON.") ap.add_argument("--ranges", help='Comma-separated inclusive ranges, e.g. "1-2,4-78,80-136". If omitted, processes ALL rows.') ap.add_argument("--keep-empty", action="store_true", help="Include empty strings as values (default: drop empties).") ap.add_argument("--case-insensitive", action="store_true", help="Fold to lowercase for NON-Defendant columns.") args = ap.parse_args() in_path = Path(args.input_csv) out_path = Path(args.output_json) drop_empty = not args.keep_empty case_insensitive_generic = bool(args.case_insensitive) if not in_path.is_file(): raise SystemExit(f"Input CSV not found: {in_path}") headers, all_rows = load_csv(in_path) if not headers: raise SystemExit("Input CSV has no header row.") if headers[0] != "Index": raise SystemExit('Input CSV must have "Index" as the first column.') ranges = parse_ranges(args.ranges) output: Dict[str, Any] = { "source_csv": str(in_path.name), "params": { "ranges": args.ranges or "ALL", "drop_empty": drop_empty, "case_insensitive_generic_columns": case_insensitive_generic, "defendant_grouping": "strip ', Jr./Sr./II..X' suffix; key=(first,last) case-insensitive" }, "by_range": [] } if not ranges: stats = compute_column_stats(headers, all_rows, drop_empty, case_insensitive_generic) output["by_range"].append({ "range": "ALL", "row_count": len(all_rows), "columns": stats }) else: for (start, end, label) in ranges: idxset = build_index_set(all_rows, start, end) subset = filter_rows_by_index(all_rows, idxset) stats = compute_column_stats(headers, subset, drop_empty, case_insensitive_generic) output["by_range"].append({ "range": label, "row_count": len(subset), "columns": stats }) out_path.write_text(json.dumps(output, indent=2, ensure_ascii=False), encoding="utf-8") print(f"Wrote {out_path} with {sum(section['row_count'] for section in output['by_range'])} rows considered across {len(output['by_range'])} range(s).") if __name__ == "__main__": main()