Data Quality Checks skill for Codex

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

interface:
  display_name: "Data Quality Checks"
  short_description: "Stop bad data before the dashboard sees it"
  default_prompt: "$data-quality-checks Add data quality checks so a broken load never reaches the finance dashboard"

#!/usr/bin/env python3
"""Profile a CSV file and optionally enforce basic data quality rules.

Standard library only. Streams the file, so it works on files larger than memory
(distinct counts are capped to keep memory bounded).

Usage:
  python3 profile_csv.py data.csv
  python3 profile_csv.py data.csv --key order_id --not-null order_id,amount \
      --max-null-rate email=0.05 --min-rows 1000 --json

Exit codes: 0 all rules pass (or no rules given), 1 a rule failed, 2 usage or read error.
"""
from __future__ import annotations

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

DISTINCT_CAP = 100_000
NULL_TOKENS = {"", "null", "none", "nan", "n/a", "na"}
PATTERNS = [
    ("integer", re.compile(r"^[+-]?\d+$")),
    ("decimal", re.compile(r"^[+-]?(\d+\.\d*|\.\d+|\d+)([eE][+-]?\d+)?$")),
    ("date", re.compile(r"^\d{4}-\d{2}-\d{2}$")),
    ("timestamp", re.compile(r"^\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}(:\d{2}(\.\d+)?)?(Z|[+-]\d{2}:?\d{2})?$")),
    ("boolean", re.compile(r"^(true|false|t|f|yes|no|0|1)$", re.I)),
]


def classify(value: str) -> str:
    for name, pattern in PATTERNS:
        if pattern.match(value):
            return name
    return "text"


def parse_kv(items: list[str]) -> dict[str, float]:
    out = {}
    for item in items:
        col, _, rate = item.partition("=")
        if not col or not rate:
            raise SystemExit(f"bad --max-null-rate value {item!r}; expected column=0.05")
        out[col] = float(rate)
    return out


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("path")
    ap.add_argument("--delimiter", default=",")
    ap.add_argument("--encoding", default="utf-8-sig")
    ap.add_argument("--key", help="comma-separated columns that must be unique together")
    ap.add_argument("--not-null", default="", help="comma-separated columns that must have no nulls")
    ap.add_argument("--max-null-rate", action="append", default=[], help="column=rate, repeatable")
    ap.add_argument("--min-rows", type=int, default=0)
    ap.add_argument("--json", action="store_true", help="print the profile as JSON")
    args = ap.parse_args()

    try:
        fh = open(args.path, newline="", encoding=args.encoding)
    except OSError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2

    with fh:
        reader = csv.DictReader(fh, delimiter=args.delimiter)
        columns = reader.fieldnames or []
        if not columns:
            print("error: no header row found", file=sys.stderr)
            return 2
        key_cols = [c.strip() for c in args.key.split(",")] if args.key else []
        not_null = [c.strip() for c in args.not_null.split(",") if c.strip()]
        max_null = parse_kv(args.max_null_rate)
        missing = [c for c in key_cols + not_null + list(max_null) if c not in columns]
        if missing:
            print(f"error: unknown columns {missing}; header is {columns}", file=sys.stderr)
            return 2

        nulls = Counter()
        kinds = {c: Counter() for c in columns}
        distinct = {c: set() for c in columns}
        capped = set()
        seen_keys: set[tuple] = set()
        dup_keys = 0
        dup_examples: list[tuple] = []
        ragged = 0
        rows = 0

        try:
            for row in reader:
                rows += 1
                if None in row or any(v is None for v in row.values()):
                    ragged += 1
                for c in columns:
                    v = (row.get(c) or "").strip()
                    if v.lower() in NULL_TOKENS:
                        nulls[c] += 1
                        continue
                    kinds[c][classify(v)] += 1
                    if c not in capped:
                        distinct[c].add(v)
                        if len(distinct[c]) > DISTINCT_CAP:
                            capped.add(c)
                            distinct[c].clear()
                if key_cols:
                    k = tuple((row.get(c) or "").strip() for c in key_cols)
                    if k in seen_keys:
                        dup_keys += 1
                        if len(dup_examples) < 5:
                            dup_examples.append(k)
                    else:
                        seen_keys.add(k)
        except (csv.Error, UnicodeDecodeError) as exc:
            print(f"error at data row {rows + 1}: {exc}", file=sys.stderr)
            return 2

    profile = {"file": args.path, "rows": rows, "ragged_rows": ragged, "columns": {}}
    for c in columns:
        rate = nulls[c] / rows if rows else 0.0
        top_kind = kinds[c].most_common(1)[0][0] if kinds[c] else "empty"
        profile["columns"][c] = {
            "null_rate": round(rate, 4),
            "distinct": f">{DISTINCT_CAP}" if c in capped else len(distinct[c]),
            "inferred": top_kind,
            "mixed_types": len(kinds[c]) > 1 and not set(kinds[c]) <= {"integer", "decimal"},
        }

    failures = []
    if rows < args.min_rows:
        failures.append(f"row count {rows} < min {args.min_rows}")
    for c in not_null:
        if nulls[c]:
            failures.append(f"{c}: {nulls[c]} nulls (expected none)")
    for c, limit in max_null.items():
        rate = profile["columns"][c]["null_rate"]
        if rate > limit:
            failures.append(f"{c}: null rate {rate:.2%} > {limit:.2%}")
    if dup_keys:
        failures.append(f"key ({', '.join(key_cols)}): {dup_keys} duplicate rows, e.g. {dup_examples}")
    if ragged:
        failures.append(f"{ragged} rows have a different field count than the header")
    profile["failures"] = failures

    if args.json:
        print(json.dumps(profile, indent=2))
    else:
        print(f"{args.path}: {rows} rows, {len(columns)} columns")
        width = max(len(c) for c in columns)
        print(f"{'column'.ljust(width)}  null_rate  distinct  inferred   mixed")
        for c, p in profile["columns"].items():
            print(f"{c.ljust(width)}  {p['null_rate']:>9.2%}  {str(p['distinct']):>8}  {p['inferred']:<9}  {'yes' if p['mixed_types'] else ''}")
        print("RESULT: " + ("PASS" if not failures else "FAIL"))
        for f in failures:
            print(f"  - {f}")
    return 1 if failures else 0


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