csv-profile.py

#!/usr/bin/env python3
"""Profile a CSV file using only the Python standard library."""

import argparse
import csv
import json
import math
from collections import Counter
from pathlib import Path


def parse_args():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("file", type=Path)
    parser.add_argument("--delimiter", default=",")
    parser.add_argument("--max-rows", type=int, default=100_000, help="Maximum rows to inspect (0 means all)")
    parser.add_argument("--top", type=int, default=5, help="Most common values per column")
    parser.add_argument("--json", action="store_true", dest="as_json")
    return parser.parse_args()


def main():
    args = parse_args()
    if args.max_rows < 0 or args.top < 1 or len(args.delimiter) != 1:
        raise SystemExit("--max-rows must be >= 0, --top >= 1, and delimiter one character")

    with args.file.open(newline="", encoding="utf-8-sig") as handle:
        reader = csv.DictReader(handle, delimiter=args.delimiter)
        if not reader.fieldnames:
            raise SystemExit("CSV header is missing")
        fields = reader.fieldnames
        if len(fields) != len(set(fields)):
            raise SystemExit("CSV header contains duplicate column names")
        stats = {name: {"nulls": 0, "numeric": 0, "min": None, "max": None, "values": Counter()} for name in fields}
        rows = 0
        for row in reader:
            if args.max_rows and rows >= args.max_rows:
                break
            rows += 1
            for name in fields:
                value = (row.get(name) or "").strip()
                column = stats[name]
                if not value:
                    column["nulls"] += 1
                    continue
                column["values"][value] += 1
                try:
                    number = float(value)
                    if math.isfinite(number):
                        column["numeric"] += 1
                        column["min"] = number if column["min"] is None else min(column["min"], number)
                        column["max"] = number if column["max"] is None else max(column["max"], number)
                except ValueError:
                    pass

    result = {"file": str(args.file), "rows_profiled": rows, "columns": {}}
    for name, column in stats.items():
        non_null = rows - column["nulls"]
        result["columns"][name] = {
            "nulls": column["nulls"],
            "null_percent": round(100 * column["nulls"] / rows, 2) if rows else 0,
            "distinct": len(column["values"]),
            "inferred_type": "numeric" if non_null and column["numeric"] == non_null else "text",
            "min": column["min"] if column["numeric"] == non_null and non_null else None,
            "max": column["max"] if column["numeric"] == non_null and non_null else None,
            "top_values": column["values"].most_common(args.top),
        }

    if args.as_json:
        print(json.dumps(result, indent=2, ensure_ascii=False))
    else:
        print(f"File: {args.file}\nRows profiled: {rows}\nColumns: {len(fields)}")
        for name, column in result["columns"].items():
            print(f"\n[{name}] type={column['inferred_type']} nulls={column['nulls']} ({column['null_percent']}%) distinct={column['distinct']}")
            if column["min"] is not None:
                print(f"  range: {column['min']} .. {column['max']}")
            print(f"  top: {column['top_values']}")


if __name__ == "__main__":
    main()