Data Profiling Tool (python, written by Codex)

Codex wrote this python project from the task below. It installed and ran on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: Data Profiling Tool

Write a program that profiles tabular datasets by analyzing column types, distributions, missing values, correlations, and data quality issues, producing a comprehensive data quality report.

FUNCTIONAL REQUIREMENTS:
- Accept a CSV or JSON data file path as a command-line argument
- Auto-detect column data types: numeric (integer vs float), string, boolean, date/timestamp, and categorical (low cardinality strings)
- For numeric columns: compute min, max, mean, median, standard deviation, skewness, and percentiles (25th, 50th, 75th, 95th, 99th)
- For string columns: compute min/max/average length, most common values (top 10), and unique count
- For all columns: count total values, missing/null values, missing percentage, and unique value count
- Detect potential data quality issues: columns that are entirely null, columns with a single unique value, numeric columns with extreme outliers (beyond 4 standard deviations), and string columns that look like misclassified numbers or dates
- Compute a pairwise correlation matrix for all numeric columns
- Print a formatted summary report to console showing key statistics per column
- Save the full profiling report as a JSON file with --output flag (default: data_profile.json)
- If no input file is given, generate a sample dataset with 1000 rows across at least 8 columns of mixed types including intentional quality issues, then profile it
- Handle files with inconsistent delimiters or encoding issues gracefully

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

# Data Profiling Tool (Python)

Profiles CSV/JSON datasets for column types, missingness, distributions, outliers, correlations, and data quality issues.

## Requirements

- Ubuntu 22.04
- Python 3.10+

## Dependencies

- Direct: none
- Transitive: none

Pinned in `requirements.txt` (standard library only).

## Run

With input:

```bash
python src/main.py /path/to/data.csv --output data_profile.json
```

JSON input:

```bash
python src/main.py /path/to/data.json --output data_profile.json
```

No input (generates 1000-row sample dataset):

```bash
python src/main.py
```

## Output

- Console summary table per column
- Full JSON report (`data_profile.json` by default)

## Robustness Notes

- CSV delimiter detection supports `,`, `;`, tab, and `|`
- Encoding fallback from UTF-8 to Latin-1 for decoding issues

# No external dependencies required.
# Direct dependencies: none
# Transitive dependencies: none

#!/usr/bin/env python3
import argparse
import csv
import json
import math
from collections import Counter
from datetime import datetime
from pathlib import Path
from typing import Any


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Data Profiling Tool")
    parser.add_argument("input_file", nargs="?", help="CSV or JSON input file")
    parser.add_argument("--output", default="data_profile.json", help="Output JSON report path")
    return parser.parse_args()


def read_text_with_fallback(path: Path) -> tuple[str, str]:
    raw = path.read_bytes()
    try:
        return raw.decode("utf-8"), "utf-8"
    except UnicodeDecodeError:
        return raw.decode("latin-1"), "latin-1"


def detect_delimiter(sample_lines: list[str]) -> str:
    candidates = [",", ";", "\t", "|"]
    best = ","
    best_score = -1
    for delim in candidates:
        score = sum(line.count(delim) for line in sample_lines)
        if score > best_score:
            best_score = score
            best = delim
    return best


def load_csv(path: Path) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    text, encoding = read_text_with_fallback(path)
    lines = [line for line in text.replace("\r\n", "\n").replace("\r", "\n").split("\n") if line.strip()]
    if not lines:
        return [], {"input_format": "csv", "encoding": encoding, "delimiter": ","}
    delimiter = detect_delimiter(lines[:5])
    reader = csv.DictReader(lines, delimiter=delimiter)
    rows = [dict(row) for row in reader]
    return rows, {"input_format": "csv", "encoding": encoding, "delimiter": delimiter}


def load_json(path: Path) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    text, encoding = read_text_with_fallback(path)
    rows: list[dict[str, Any]] = []
    try:
        parsed = json.loads(text)
        if isinstance(parsed, list):
            rows = [x for x in parsed if isinstance(x, dict)]
        elif isinstance(parsed, dict):
            rows = [parsed]
    except json.JSONDecodeError:
        rows = [json.loads(line) for line in text.splitlines() if line.strip()]
        rows = [x for x in rows if isinstance(x, dict)]
    return rows, {"input_format": "json", "encoding": encoding}


def is_missing(value: Any) -> bool:
    if value is None:
        return True
    s = str(value).strip().lower()
    return s in {"", "null", "na", "n/a", "none"}


def parse_number(value: Any) -> float | None:
    if value is None:
        return None
    s = str(value).strip()
    if not s:
        return None
    if not all(ch.isdigit() or ch in ".-+" for ch in s):
        return None
    try:
        return float(s)
    except ValueError:
        return None


def parse_bool(value: Any) -> bool | None:
    if value is None:
        return None
    s = str(value).strip().lower()
    if s in {"true", "1", "yes", "y"}:
        return True
    if s in {"false", "0", "no", "n"}:
        return False
    return None


def parse_date(value: Any) -> float | None:
    if value is None:
        return None
    s = str(value).strip()
    if not s:
        return None
    for fmt in (
        "%Y-%m-%d",
        "%Y-%m-%dT%H:%M:%S",
        "%Y-%m-%dT%H:%M:%SZ",
        "%Y-%m-%d %H:%M:%S",
    ):
        try:
            return datetime.strptime(s, fmt).timestamp()
        except ValueError:
            continue
    try:
        return datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp()
    except ValueError:
        return None


def mean(values: list[float]) -> float | None:
    return sum(values) / len(values) if values else None


def percentile(sorted_values: list[float], p: float) -> float | None:
    if not sorted_values:
        return None
    if len(sorted_values) == 1:
        return sorted_values[0]
    pos = (len(sorted_values) - 1) * p
    lo = int(math.floor(pos))
    hi = int(math.ceil(pos))
    if lo == hi:
        return sorted_values[lo]
    weight = pos - lo
    return sorted_values[lo] + (sorted_values[hi] - sorted_values[lo]) * weight


def stddev(values: list[float], m: float | None = None) -> float:
    if len(values) <= 1:
        return 0.0
    mu = mean(values) if m is None else m
    assert mu is not None
    var = sum((x - mu) ** 2 for x in values) / len(values)
    return math.sqrt(var)


def skewness(values: list[float], m: float | None = None, sd: float | None = None) -> float:
    if len(values) < 3:
        return 0.0
    mu = mean(values) if m is None else m
    sigma = stddev(values, mu) if sd is None else sd
    assert mu is not None
    if sigma == 0:
        return 0.0
    m3 = sum((x - mu) ** 3 for x in values) / len(values)
    return m3 / (sigma ** 3)


def detect_type(values: list[str]) -> str:
    if not values:
        return "string"
    bool_count = sum(parse_bool(v) is not None for v in values)
    number_count = sum(parse_number(v) is not None for v in values)
    date_count = sum(parse_date(v) is not None for v in values)
    if bool_count == len(values):
        return "boolean"
    if number_count == len(values):
        has_float = any("." in str(v) for v in values)
        return "float" if has_float else "integer"
    if date_count >= max(3, int(len(values) * 0.9)):
        return "date"
    unique_count = len(set(values))
    unique_ratio = unique_count / len(values)
    if unique_count <= 20 or unique_ratio <= 0.1:
        return "categorical"
    return "string"


def correlation(xs: list[float | None], ys: list[float | None]) -> float | None:
    paired = [(x, y) for x, y in zip(xs, ys) if x is not None and y is not None]
    if len(paired) < 2:
        return None
    xv = [x for x, _ in paired]
    yv = [y for _, y in paired]
    mx = mean(xv)
    my = mean(yv)
    assert mx is not None and my is not None
    sx = stddev(xv, mx)
    sy = stddev(yv, my)
    if sx == 0 or sy == 0:
        return 0.0
    cov = sum((x - mx) * (y - my) for x, y in paired) / len(paired)
    return cov / (sx * sy)


def profile_data(rows: list[dict[str, Any]]) -> dict[str, Any]:
    columns = sorted({k for row in rows for k in row.keys()})
    column_profiles: dict[str, Any] = {}
    issues: list[dict[str, Any]] = []
    numeric_columns: list[str] = []
    numeric_series: dict[str, list[float | None]] = {}

    for col in columns:
        raw = [row.get(col) for row in rows]
        missing_count = sum(is_missing(v) for v in raw)
        total = len(raw)
        missing_pct = (missing_count / total * 100) if total else 0.0
        non_missing = [v for v in raw if not is_missing(v)]
        non_missing_str = [str(v) for v in non_missing]
        unique_count = len(set(non_missing_str))
        dtype = detect_type(non_missing_str)

        profile: dict[str, Any] = {
            "type": dtype,
            "total_values": total,
            "missing_values": missing_count,
            "missing_percentage": round(missing_pct, 4),
            "unique_values": unique_count,
            "quality_issues": [],
        }

        if missing_count == total:
            profile["quality_issues"].append("entirely_null")
            issues.append({"column": col, "issue": "entirely_null"})
        if unique_count == 1 and non_missing_str:
            profile["quality_issues"].append("single_unique_value")
            issues.append({"column": col, "issue": "single_unique_value"})

        if dtype in {"integer", "float"}:
            nums = sorted(float(v) for v in non_missing_str if parse_number(v) is not None)
            mu = mean(nums) if nums else None
            sd = stddev(nums, mu) if nums else 0.0
            outliers = [v for v in nums if sd > 0 and mu is not None and abs(v - mu) > 4 * sd]
            if outliers:
                profile["quality_issues"].append("extreme_outliers")
                issues.append({"column": col, "issue": "extreme_outliers", "count": len(outliers)})
            profile["numeric_stats"] = {
                "min": nums[0] if nums else None,
                "max": nums[-1] if nums else None,
                "mean": mu,
                "median": percentile(nums, 0.5),
                "stddev": sd,
                "skewness": skewness(nums, mu, sd) if nums else 0.0,
                "percentiles": {
                    "p25": percentile(nums, 0.25),
                    "p50": percentile(nums, 0.5),
                    "p75": percentile(nums, 0.75),
                    "p95": percentile(nums, 0.95),
                    "p99": percentile(nums, 0.99),
                },
                "outliers_beyond_4std": outliers,
            }
            numeric_columns.append(col)
            numeric_series[col] = [parse_number(v) if not is_missing(v) else None for v in raw]
        elif dtype in {"string", "categorical"}:
            lengths = [len(v) for v in non_missing_str]
            numeric_like = sum(parse_number(v) is not None for v in non_missing_str)
            date_like = sum(parse_date(v) is not None for v in non_missing_str)
            if non_missing_str and numeric_like / len(non_missing_str) >= 0.8:
                profile["quality_issues"].append("string_looks_numeric")
                issues.append({"column": col, "issue": "string_looks_numeric"})
            if non_missing_str and date_like / len(non_missing_str) >= 0.8:
                profile["quality_issues"].append("string_looks_date")
                issues.append({"column": col, "issue": "string_looks_date"})
            profile["string_stats"] = {
                "min_length": min(lengths) if lengths else None,
                "max_length": max(lengths) if lengths else None,
                "avg_length": mean([float(x) for x in lengths]) if lengths else None,
                "unique_count": unique_count,
                "top_values": [{"value": v, "count": c} for v, c in Counter(non_missing_str).most_common(10)],
            }

        column_profiles[col] = profile

    correlations: dict[str, dict[str, float | None]] = {}
    for c1 in numeric_columns:
        correlations[c1] = {}
        for c2 in numeric_columns:
            correlations[c1][c2] = 1.0 if c1 == c2 else correlation(numeric_series[c1], numeric_series[c2])

    return {
        "row_count": len(rows),
        "column_count": len(columns),
        "columns": column_profiles,
        "correlations": correlations,
        "issues": issues,
    }


def print_summary(report: dict[str, Any]) -> None:
    header = ["Column", "Type", "Total", "Missing%", "Unique", "Notes"]
    rows = []
    for col, info in report["columns"].items():
        rows.append(
            [
                col,
                info["type"],
                str(info["total_values"]),
                f"{info['missing_percentage']:.2f}",
                str(info["unique_values"]),
                ",".join(info["quality_issues"]),
            ]
        )
    widths = [len(h) for h in header]
    for row in rows:
        for i, val in enumerate(row):
            widths[i] = max(widths[i], len(val))
    sep = "+-" + "-+-".join("-" * w for w in widths) + "-+"

    def fmt(row: list[str]) -> str:
        return "| " + " | ".join(row[i] + " " * (widths[i] - len(row[i])) for i in range(len(row))) + " |"

    print("Data Profiling Summary")
    print("======================")
    print(f"Rows: {report['row_count']}")
    print(f"Columns: {report['column_count']}")
    print(sep)
    print(fmt(header))
    print(sep)
    for row in rows:
        print(fmt(row))
    print(sep)


def generate_sample_rows() -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    countries = ["US", "CA", "GB", "DE", "IN"]
    for i in range(1000):
        age = 140 if i % 200 == 0 else 20 + (i % 45)
        income = 250000 if i % 150 == 0 else 30000 + (i % 120) * 800 + (i % 7) * 13.5
        rows.append(
            {
                "id": str(i + 1),
                "age": str(age),
                "income": f"{income:.2f}",
                "is_active": "true" if i % 2 == 0 else "false",
                "signup_date": f"2025-{(i % 12) + 1:02d}-{(i % 28) + 1:02d}T12:00:00Z",
                "country": countries[i % len(countries)],
                "status_code_str": str(1000 + (i % 4)),
                "comment": "" if i % 25 == 0 else f"note_{i % 17}",
                "constant_col": "CONST",
                "all_null_col": "",
            }
        )
    return rows


def write_sample_csv(rows: list[dict[str, Any]], path: Path) -> None:
    headers = list(rows[0].keys())
    with path.open("w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=headers)
        writer.writeheader()
        writer.writerows(rows)


def main() -> int:
    args = parse_args()
    output_path = Path(args.output).resolve()

    metadata: dict[str, Any] = {}
    if args.input_file:
        input_path = Path(args.input_file).resolve()
        if not input_path.exists():
            print(f"Input file not found: {input_path}")
            return 1
        ext = input_path.suffix.lower()
        if ext in {".json", ".jsonl", ".ndjson"}:
            rows, extra = load_json(input_path)
        else:
            rows, extra = load_csv(input_path)
        metadata.update({"input_file": str(input_path)})
        metadata.update(extra)
    else:
        rows = generate_sample_rows()
        sample_path = Path("sample_profile_data.csv").resolve()
        write_sample_csv(rows, sample_path)
        metadata["generated_sample"] = str(sample_path)
        print(f"No input file provided. Generated sample dataset: {sample_path}")

    report = {
        "generated_at": datetime.utcnow().isoformat() + "Z",
        **metadata,
        **profile_data(rows),
    }

    print_summary(report)
    output_path.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
    print(f"Saved JSON profile: {output_path}")
    return 0


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