CSV Processing skill for Codex

Gives the Codex app a display name, a one-line description and a starter prompt for this skill.

interface:
  display_name: "CSV Processing"
  short_description: "Big, messy CSVs cleaned without losing a digit"
  default_prompt: "$csv-processing This 8 GB CSV crashes pandas - filter it to EU orders and save as Parquet"

#!/usr/bin/env python3
"""Profile a CSV/TSV file with the standard library only, streaming row by row.

Detects encoding (BOM, UTF-8, falls back to cp1252), delimiter, quoting and line
endings, then reports per column: inferred type, null-ish count, distinct estimate,
max length and sample values. Flags ragged rows, duplicate or blank headers, IDs with
leading zeros, long digit strings Excel would corrupt, decimal commas, mixed date
formats, and values starting with = + - @ (spreadsheet formula injection).

Usage:
    python3 csv_profile.py data.csv
    python3 csv_profile.py data.csv --limit 200000     # profile only the first N rows
    python3 csv_profile.py data.csv --delimiter ';' --encoding latin-1

Exit codes: 0 clean, 1 structural problems (ragged rows, bad header), 2 unreadable.
"""
from __future__ import annotations

import argparse
import codecs
import csv
import re
import sys
from collections import Counter

NULLS = {"", "na", "n/a", "null", "none", "nan", "-", "--", "#n/a", "nil"}
INT = re.compile(r"^[+-]?\d+$")
FLOAT = re.compile(r"^[+-]?(\d+\.\d*|\.\d+|\d+)([eE][+-]?\d+)?$")
THOUSANDS = re.compile(r"^[+-]?\d{1,3}(,\d{3})+(\.\d+)?$")
DEC_COMMA = re.compile(r"^[+-]?\d+,\d+$")
DATE_PATTERNS = {
    "iso": re.compile(r"^\d{4}-\d{2}-\d{2}([ T]\d{2}:\d{2}(:\d{2}(\.\d+)?)?(Z|[+-]\d{2}:?\d{2})?)?$"),
    "slash": re.compile(r"^\d{1,2}/\d{1,2}/\d{4}$"),
    "dotted": re.compile(r"^\d{1,2}\.\d{1,2}\.\d{4}$"),
}
BOOL = {"true", "false", "yes", "no"}
csv.field_size_limit(min(sys.maxsize, 2**31 - 1))


def detect_encoding(path: str, forced: str | None) -> str:
    if forced:
        return forced
    with open(path, "rb") as fh:
        head = fh.read(4)
        if head.startswith(codecs.BOM_UTF8):
            return "utf-8-sig"
        if head.startswith((codecs.BOM_UTF16_LE, codecs.BOM_UTF16_BE)):
            return "utf-16"
        fh.seek(0)
        dec = codecs.getincrementaldecoder("utf-8")()
        try:
            while chunk := fh.read(1 << 20):
                dec.decode(chunk)
            dec.decode(b"", final=True)
            return "utf-8"
        except UnicodeDecodeError:
            return "cp1252"


def classify(v: str) -> str:
    s = v.strip()
    if s.lower() in NULLS:
        return "null"
    if INT.match(s):
        return "int"
    if FLOAT.match(s):
        return "float"
    if THOUSANDS.match(s):
        return "number_with_thousands_sep"
    if DEC_COMMA.match(s):
        return "decimal_comma"
    for name, pat in DATE_PATTERNS.items():
        if pat.match(s):
            return f"date:{name}"
    if s.lower() in BOOL:
        return "bool"
    return "text"


def main() -> int:
    ap = argparse.ArgumentParser(description="Profile a CSV file (stdlib only).")
    ap.add_argument("path")
    ap.add_argument("--limit", type=int, default=0, help="max data rows to read (0 = all)")
    ap.add_argument("--delimiter")
    ap.add_argument("--encoding")
    a = ap.parse_args()

    try:
        enc = detect_encoding(a.path, a.encoding)
        with open(a.path, "rb") as fh:
            raw = fh.read(65536)
        crlf = raw.count(b"\r\n")
        lf = raw.count(b"\n") - crlf
        sample = raw.decode(enc.replace("-sig", ""), errors="replace").lstrip("")
        if a.delimiter:
            delim = a.delimiter.encode().decode("unicode_escape")
        else:
            try:
                delim = csv.Sniffer().sniff(sample, delimiters=",;\t|").delimiter
            except csv.Error:
                delim = ","
        fh = open(a.path, newline="", encoding=enc, errors="replace")
    except OSError as exc:
        print(f"ERROR: {exc}")
        return 2

    problems: list[str] = []
    notes: list[str] = []
    with fh:
        reader = csv.reader(fh, delimiter=delim)
        try:
            header = next(reader)
        except StopIteration:
            print("ERROR: file is empty")
            return 2
        width = len(header)
        dupes = [h for h, n in Counter(header).items() if n > 1]
        blanks = [i for i, h in enumerate(header) if not h.strip()]
        if dupes:
            problems.append(f"duplicate header names: {dupes}")
        if blanks:
            problems.append(f"blank header names at column index {blanks}")
        if any(h != h.strip() for h in header):
            notes.append("some header names have leading/trailing spaces")

        types = [Counter() for _ in header]
        distinct = [set() for _ in header]
        maxlen = [0] * width
        samples = [[] for _ in header]
        lead_zero = [0] * width
        long_digits = [0] * width
        inject = [0] * width
        ragged = Counter()
        ragged_first = []
        rows = 0
        for row in reader:
            rows += 1
            if len(row) != width:
                ragged[len(row)] += 1
                if len(ragged_first) < 5:
                    ragged_first.append(reader.line_num)
            for i, v in enumerate(row[:width]):
                t = classify(v)
                types[i][t] += 1
                if len(distinct[i]) < 10001:
                    distinct[i].add(v)
                maxlen[i] = max(maxlen[i], len(v))
                if t != "null" and len(samples[i]) < 3 and v not in samples[i]:
                    samples[i].append(v)
                s = v.strip()
                if t == "int" and len(s) > 1 and s[0] == "0":
                    lead_zero[i] += 1
                if t == "int" and len(s.lstrip("+-")) > 15:
                    long_digits[i] += 1
                if t == "text" and s[:1] in ("=", "+", "-", "@"):
                    inject[i] += 1
            if a.limit and rows >= a.limit:
                notes.append(f"stopped after --limit {a.limit} rows")
                break

    ending = "CRLF" if crlf > lf else "LF"
    print(f"File: {a.path}")
    print(f"Encoding: {enc}  Delimiter: {delim!r}  Line endings: {ending}  Columns: {width}  Data rows: {rows}")
    if ragged:
        problems.append(
            f"{sum(ragged.values())} rows with wrong field count {dict(ragged)} "
            f"(expected {width}); first at physical lines {ragged_first}"
        )
    print(f"\n{'column':28} {'type':26} {'nulls':>7} {'distinct':>9} {'maxlen':>7}  samples")
    for i, name in enumerate(header):
        c = types[i]
        nulls = c.pop("null", 0)
        if not c:
            kind = "empty"
        elif len(c) == 1:
            kind = next(iter(c))
        elif set(c) <= {"int", "float"}:
            kind = "float"
        else:
            top = c.most_common(2)
            kind = f"mixed({top[0][0]}>{top[1][0]})"
        d = len(distinct[i])
        dtxt = ">10000" if d > 10000 else str(d)
        print(f"{name[:28]:28} {kind[:26]:26} {nulls:>7} {dtxt:>9} {maxlen[i]:>7}  {samples[i]}")
        if lead_zero[i]:
            notes.append(f"'{name}': {lead_zero[i]} values with leading zeros - keep as text (IDs, ZIP codes)")
        if long_digits[i]:
            notes.append(f"'{name}': {long_digits[i]} integers longer than 15 digits - Excel will round them")
        if inject[i]:
            notes.append(f"'{name}': {inject[i]} text values start with = + - @ - escape before opening in a spreadsheet")
        if "decimal_comma" in c or "number_with_thousands_sep" in c:
            notes.append(f"'{name}': locale-formatted numbers - parse explicitly, do not cast directly")
        date_kinds = [k for k in c if k.startswith("date:")]
        if len(date_kinds) > 1 or (date_kinds and len(c) > 1):
            notes.append(f"'{name}': mixed date formats or dates mixed with other values {dict(c)}")
        if "date:slash" in c:
            notes.append(f"'{name}': slash dates (03/04/2025) are ambiguous: day-first or month-first - confirm with the source")

    print("\nProblems:" if problems else "\nProblems: none")
    for p in problems:
        print(f"  - {p}")
    print("Notes:" if notes else "Notes: none")
    for n in notes:
        print(f"  - {n}")
    return 1 if problems else 0


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