CSV Cleaning skill for Claude Code

#!/usr/bin/env python3
"""Profile a messy CSV/TSV before cleaning it (stdlib only, streams large files).

Usage:
  python3 scripts/csv_profile.py data.csv
  python3 scripts/csv_profile.py data.csv --delimiter ';' --encoding cp1252 --max-rows 500000

Reports: encoding guess and BOM, line endings, delimiter, header issues, ragged rows
(with line numbers), exact duplicate rows, and per column: empty share, distinct count,
inferred type, ambiguous date formats, whitespace problems, and sample values.
Exit code 1 if structural problems were found (undecodable bytes, ragged rows,
duplicate/empty headers), 0 otherwise, 2 on usage errors.
"""
import argparse
import csv
import hashlib
import re
import sys
from collections import Counter

csv.field_size_limit(min(sys.maxsize, 2**31 - 1))

NULL_TOKENS = {"", "na", "n/a", "null", "none", "nan", "-", "--", "?"}
PATTERNS = [
    ("int", re.compile(r"^[+-]?\d+$")),
    ("decimal", re.compile(r"^[+-]?(\d+\.\d*|\.\d+|\d+)([eE][+-]?\d+)?$")),
    ("decimal_comma", re.compile(r"^[+-]?\d{1,3}(\.\d{3})*,\d+$|^[+-]?\d+,\d+$")),
    ("thousands_sep", re.compile(r"^[+-]?\d{1,3}(,\d{3})+(\.\d+)?$")),
    ("date_iso", re.compile(r"^\d{4}-\d{2}-\d{2}$")),
    ("datetime_iso", re.compile(r"^\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}(:\d{2}(\.\d+)?)?(Z|[+-]\d{2}:?\d{2})?$")),
    ("date_slash", re.compile(r"^\d{1,2}/\d{1,2}/\d{2,4}$")),
    ("date_dot", re.compile(r"^\d{1,2}\.\d{1,2}\.\d{2,4}$")),
    ("bool", re.compile(r"^(true|false|yes|no|y|n|t|f)$", re.I)),
    ("currency", re.compile(r"^[$€£¥]\s?[\d.,]+$|^[\d.,]+\s?[$€£¥]$")),
    ("percent", re.compile(r"^[+-]?[\d.,]+\s?%$")),
]


def detect_encoding(path, sample_bytes=2_000_000):
    with open(path, "rb") as fh:
        raw = fh.read(sample_bytes)
    bom = None
    if raw.startswith(b"\xef\xbb\xbf"):
        bom = "UTF-8 BOM"
    elif raw.startswith((b"\xff\xfe", b"\xfe\xff")):
        bom = "UTF-16 BOM"
        return "utf-16", bom, raw
    for enc in ("utf-8-sig", "cp1252", "latin-1"):
        try:
            raw.decode(enc)
            return enc, bom, raw
        except UnicodeDecodeError:
            continue
    return "latin-1", bom, raw


def line_endings(raw):
    crlf = raw.count(b"\r\n")
    lf = raw.count(b"\n") - crlf
    cr = raw.count(b"\r") - crlf
    kinds = [k for k, n in (("CRLF", crlf), ("LF", lf), ("CR", cr)) if n]
    return "+".join(kinds) or "none", len(kinds) > 1


def guess_delimiter(text, candidates=",;\t|", max_lines=200):
    """Pick the delimiter that yields the most consistent field count (>1) across lines."""
    lines = text.splitlines()[:max_lines]
    if len(lines) > 1:
        lines = lines[:-1]  # last sampled line may be truncated
    best, best_score = ",", (-1.0, 0)
    for d in candidates:
        counts = Counter(len(r) for r in csv.reader(lines, delimiter=d) if r)
        if not counts:
            continue
        mode, freq = counts.most_common(1)[0]
        if mode < 2:
            continue
        score = (freq / sum(counts.values()), mode)
        if score > best_score:
            best, best_score = d, score
    return best


def classify(value):
    for name, pat in PATTERNS:
        if pat.match(value):
            return name
    return "text"


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("path")
    ap.add_argument("--delimiter")
    ap.add_argument("--encoding")
    ap.add_argument("--max-rows", type=int, default=0, help="stop after N data rows (0 = all)")
    ap.add_argument("--samples", type=int, default=5)
    a = ap.parse_args()

    try:
        enc_guess, bom, raw = detect_encoding(a.path)
    except OSError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    encoding = a.encoding or enc_guess
    endings, mixed_endings = line_endings(raw)

    sample_text = raw.decode(encoding, errors="replace")[:200_000]
    delimiter = a.delimiter or guess_delimiter(sample_text)
    problems = []

    print(f"file: {a.path}")
    print(f"encoding: {encoding}{' (forced)' if a.encoding else ' (guessed)'}" + (f", {bom}" if bom else ""))
    print(f"line endings: {endings}" + ("  <-- MIXED" if mixed_endings else ""))
    print(f"delimiter: {delimiter!r}")

    decode_errors = 0
    with open(a.path, "r", encoding=encoding, errors="strict" if encoding != "latin-1" else "replace", newline="") as fh:
        def safe_lines(f):
            nonlocal decode_errors
            while True:
                try:
                    line = f.readline()
                except UnicodeDecodeError:
                    decode_errors += 1
                    return
                if not line:
                    return
                yield line

        reader = csv.reader(safe_lines(fh), delimiter=delimiter)
        try:
            header = next(reader)
        except StopIteration:
            print("error: empty file", file=sys.stderr)
            return 1
        header = [h.strip() for h in header]
        dupes = [h for h, n in Counter(header).items() if n > 1]
        empties = [i for i, h in enumerate(header) if not h]
        print(f"columns ({len(header)}): {header}")
        if dupes:
            problems.append(f"duplicate header names: {dupes}")
        if empties:
            problems.append(f"empty header names at positions {empties}")
        if any(re.match(r"^\d+(\.\d+)?$", h) for h in header if h):
            problems.append("numeric header names: file may have no header row")

        ncol = len(header)
        stats = [
            {"empty": 0, "types": Counter(), "distinct": set(), "overflow": False, "ws": 0,
             "maxlen": 0, "samples": [], "dmy_hint": 0, "mdy_hint": 0}
            for _ in header
        ]
        ragged = Counter()
        ragged_lines = []
        seen_hashes = set()
        dup_rows = 0
        rows = 0
        for row in reader:
            if a.max_rows and rows >= a.max_rows:
                break
            rows += 1
            if len(row) != ncol:
                ragged[len(row)] += 1
                if len(ragged_lines) < 10:
                    ragged_lines.append((reader.line_num, len(row)))
            h = hashlib.blake2b("\x1f".join(row).encode("utf-8", "replace"), digest_size=12).digest()
            if h in seen_hashes:
                dup_rows += 1
            else:
                seen_hashes.add(h)
            for i, v in enumerate(row[:ncol]):
                s = stats[i]
                if v != v.strip():
                    s["ws"] += 1
                vs = v.strip()
                s["maxlen"] = max(s["maxlen"], len(vs))
                if vs.lower() in NULL_TOKENS:
                    s["empty"] += 1
                    continue
                t = classify(vs)
                s["types"][t] += 1
                if t in ("date_slash", "date_dot"):
                    first, second = re.split(r"[/.]", vs)[:2]
                    if int(first) > 12:
                        s["dmy_hint"] += 1
                    elif int(second) > 12:
                        s["mdy_hint"] += 1
                if not s["overflow"]:
                    s["distinct"].add(vs)
                    if len(s["distinct"]) > 100_000:
                        s["overflow"] = True
                if len(s["samples"]) < a.samples and vs not in s["samples"]:
                    s["samples"].append(vs)

    print(f"data rows: {rows:,}" + (" (truncated by --max-rows)" if a.max_rows and rows >= a.max_rows else ""))
    if decode_errors:
        problems.append(f"undecodable bytes with encoding {encoding}; try --encoding cp1252 or latin-1")
    if ragged:
        detail = ", ".join(f"{n:,} rows with {k} fields" for k, n in sorted(ragged.items()))
        problems.append(f"ragged rows (expected {ncol} fields): {detail}; first at lines {ragged_lines}")
    print(f"exact duplicate rows: {dup_rows:,}")
    print()

    for name, s in zip(header, stats):
        filled = rows - s["empty"]
        total_typed = sum(s["types"].values())
        top = s["types"].most_common(3)
        type_str = ", ".join(f"{t} {n / total_typed:.0%}" for t, n in top) if total_typed else "all empty"
        distinct = ">100k" if s["overflow"] else f"{len(s['distinct']):,}"
        print(f"[{name}]")
        print(f"  empty/null: {s['empty']:,} ({(s['empty'] / rows if rows else 0):.1%})  distinct: {distinct}  max length: {s['maxlen']}")
        print(f"  types: {type_str}")
        notes = []
        if len(s["types"]) > 1 and top and top[0][1] / max(total_typed, 1) < 0.99:
            notes.append("mixed types")
        if s["ws"]:
            notes.append(f"{s['ws']:,} values with leading/trailing whitespace")
        if s["dmy_hint"] and s["mdy_hint"]:
            notes.append("dates match both DD/MM and MM/DD: inconsistent formats")
        elif s["dmy_hint"]:
            notes.append("dates look DD/MM/YYYY")
        elif s["mdy_hint"]:
            notes.append("dates look MM/DD/YYYY")
        elif s["types"].get("date_slash") or s["types"].get("date_dot"):
            notes.append("slash/dot dates AMBIGUOUS (no day > 12 seen): confirm format with the source")
        if s["types"].get("decimal_comma"):
            notes.append("decimal commas present (European number format)")
        if s["types"].get("thousands_sep"):
            notes.append("thousands separators present")
        if not s["overflow"] and filled and len(s["distinct"]) == filled and filled == rows:
            notes.append("unique and complete: candidate key")
        if notes:
            print(f"  notes: {'; '.join(notes)}")
        print(f"  samples: {s['samples']}")

    print()
    if problems:
        print("STRUCTURAL PROBLEMS:")
        for p in problems:
            print(f"  - {p}")
        return 1
    print("No structural problems found.")
    return 0


if __name__ == "__main__":
    sys.exit(main())