data-quality-check.py

#!/usr/bin/env python3
"""Validate common CSV quality rules and return a CI-friendly exit code."""

import argparse
import csv
from collections import Counter
from datetime import datetime
from pathlib import Path


def parse_range(value):
    try:
        column, minimum, maximum = value.split(":", 2)
        return column, float(minimum), float(maximum)
    except ValueError as error:
        raise argparse.ArgumentTypeError("range must be COLUMN:MIN:MAX") from error


def parse_args():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("file", type=Path)
    parser.add_argument("--required", action="append", default=[])
    parser.add_argument("--unique", action="append", default=[])
    parser.add_argument("--timestamp")
    parser.add_argument("--timestamp-format", default="%Y-%m-%dT%H:%M:%S")
    parser.add_argument("--range", action="append", default=[], type=parse_range, dest="ranges")
    parser.add_argument("--delimiter", default=",")
    return parser.parse_args()


def main():
    args = parse_args()
    if len(args.delimiter) != 1:
        raise SystemExit("--delimiter must be one character")
    errors = Counter()
    seen = {column: set() for column in args.unique}
    required_columns = set(args.required + args.unique + ([args.timestamp] if args.timestamp else []) + [item[0] for item in args.ranges])

    with args.file.open(newline="", encoding="utf-8-sig") as handle:
        reader = csv.DictReader(handle, delimiter=args.delimiter)
        if reader.fieldnames and len(reader.fieldnames) != len(set(reader.fieldnames)):
            print("ERROR duplicate column names in CSV header")
            return 2
        headers = set(reader.fieldnames or [])
        missing = required_columns - headers
        if missing:
            print("ERROR missing columns: " + ", ".join(sorted(missing)))
            return 2

        rows = 0
        for number, row in enumerate(reader, start=2):
            rows += 1
            for column in args.required:
                if not (row.get(column) or "").strip():
                    errors[f"required:{column}"] += 1
            for column in args.unique:
                value = (row.get(column) or "").strip()
                if not value:
                    errors[f"unique-empty:{column}"] += 1
                elif value in seen[column]:
                    errors[f"duplicate:{column}"] += 1
                else:
                    seen[column].add(value)
            if args.timestamp:
                try:
                    datetime.strptime(row[args.timestamp], args.timestamp_format)
                except (ValueError, TypeError):
                    errors[f"timestamp:{args.timestamp}"] += 1
            for column, minimum, maximum in args.ranges:
                try:
                    value = float(row[column])
                    if not minimum <= value <= maximum:
                        errors[f"range:{column}"] += 1
                except (ValueError, TypeError):
                    errors[f"numeric:{column}"] += 1

    print(f"Rows checked: {rows}")
    if errors:
        for rule, count in sorted(errors.items()):
            print(f"FAIL {rule}: {count}")
        return 1
    print("OK: all configured checks passed")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())