Data Analyst

#!/usr/bin/env python3
"""Profile CSV or JSON datasets: column statistics, null rates, cardinality, and outliers.

Reads a data file and computes per-column statistics including count, null
rate, unique values, min/max, mean, median, standard deviation, and
quartiles for numeric columns.  Flags potential quality issues.

Usage:
    python data_profiler.py --file data.csv
    python data_profiler.py --file data.json --json
    python data_profiler.py --file data.csv --top 5
"""

import argparse
import csv
import json
import math
import os
import sys
from collections import Counter


def _is_numeric(value: str) -> bool:
    try:
        float(value)
        return True
    except (ValueError, TypeError):
        return False


def _median(values: list) -> float:
    s = sorted(values)
    n = len(s)
    if n == 0:
        return 0.0
    mid = n // 2
    if n % 2 == 0:
        return (s[mid - 1] + s[mid]) / 2.0
    return s[mid]


def _percentile(values: list, p: float) -> float:
    s = sorted(values)
    n = len(s)
    if n == 0:
        return 0.0
    k = (n - 1) * p
    f = math.floor(k)
    c = math.ceil(k)
    if f == c:
        return s[int(k)]
    return s[f] * (c - k) + s[c] * (k - f)


def _std_dev(values: list, mean: float) -> float:
    if len(values) < 2:
        return 0.0
    variance = sum((v - mean) ** 2 for v in values) / (len(values) - 1)
    return math.sqrt(variance)


def load_data(file_path: str) -> list:
    ext = os.path.splitext(file_path)[1].lower()
    if ext == ".csv":
        with open(file_path, "r", newline="") as f:
            reader = csv.DictReader(f)
            return list(reader)
    elif ext == ".json":
        with open(file_path, "r") as f:
            data = json.load(f)
            if isinstance(data, list):
                return data
            raise ValueError("JSON file must contain an array of objects.")
    else:
        print(f"Error: Unsupported file type '{ext}'. Use .csv or .json.", file=sys.stderr)
        sys.exit(1)


def profile_column(name: str, values: list, top_n: int = 5) -> dict:
    total = len(values)
    nulls = sum(1 for v in values if v is None or str(v).strip() == "")
    non_null = [v for v in values if v is not None and str(v).strip() != ""]
    unique = len(set(str(v) for v in non_null))

    profile = {
        "column": name,
        "total_rows": total,
        "null_count": nulls,
        "null_pct": round(nulls / total * 100, 2) if total > 0 else 0.0,
        "unique_values": unique,
        "cardinality_pct": round(unique / total * 100, 2) if total > 0 else 0.0,
    }

    # Check if numeric
    numeric_vals = []
    for v in non_null:
        if _is_numeric(str(v)):
            numeric_vals.append(float(str(v)))

    if len(numeric_vals) > len(non_null) * 0.8 and numeric_vals:
        mean = sum(numeric_vals) / len(numeric_vals)
        std = _std_dev(numeric_vals, mean)
        profile["data_type"] = "numeric"
        profile["min"] = min(numeric_vals)
        profile["max"] = max(numeric_vals)
        profile["mean"] = round(mean, 4)
        profile["median"] = round(_median(numeric_vals), 4)
        profile["std_dev"] = round(std, 4)
        profile["q25"] = round(_percentile(numeric_vals, 0.25), 4)
        profile["q75"] = round(_percentile(numeric_vals, 0.75), 4)
        # Outlier detection (IQR method)
        iqr = profile["q75"] - profile["q25"]
        lower_bound = profile["q25"] - 1.5 * iqr
        upper_bound = profile["q75"] + 1.5 * iqr
        outliers = sum(1 for v in numeric_vals if v < lower_bound or v > upper_bound)
        profile["outlier_count"] = outliers
        profile["outlier_pct"] = round(outliers / len(numeric_vals) * 100, 2)
    else:
        profile["data_type"] = "text"
        str_vals = [str(v) for v in non_null]
        if str_vals:
            lengths = [len(s) for s in str_vals]
            profile["min_length"] = min(lengths)
            profile["max_length"] = max(lengths)
            profile["avg_length"] = round(sum(lengths) / len(lengths), 1)
        # Top values
        counter = Counter(str(v) for v in non_null)
        profile["top_values"] = [
            {"value": val, "count": cnt}
            for val, cnt in counter.most_common(top_n)
        ]

    # Quality flags
    flags = []
    if profile["null_pct"] > 50:
        flags.append("HIGH_NULL_RATE")
    if profile["null_pct"] > 0 and profile["null_pct"] <= 50:
        flags.append("HAS_NULLS")
    if unique == total and total > 1:
        flags.append("POTENTIALLY_UNIQUE_KEY")
    if unique == 1 and total > 1:
        flags.append("CONSTANT_VALUE")
    if profile["data_type"] == "numeric" and profile.get("outlier_pct", 0) > 5:
        flags.append("HIGH_OUTLIER_RATE")
    profile["quality_flags"] = flags

    return profile


def profile_dataset(data: list, top_n: int = 5) -> dict:
    if not data:
        return {"row_count": 0, "column_count": 0, "columns": []}

    columns = list(data[0].keys())
    col_profiles = []
    for col in columns:
        values = [row.get(col) for row in data]
        col_profiles.append(profile_column(col, values, top_n))

    quality_issues = sum(
        1 for p in col_profiles if p["quality_flags"]
    )

    return {
        "row_count": len(data),
        "column_count": len(columns),
        "columns_with_issues": quality_issues,
        "columns": col_profiles,
    }


def main():
    parser = argparse.ArgumentParser(
        description="Profile CSV or JSON datasets with column-level statistics."
    )
    parser.add_argument("--file", required=True, help="Path to CSV or JSON file")
    parser.add_argument("--top", type=int, default=5, help="Number of top values to show for text columns (default: 5)")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")
    args = parser.parse_args()

    if not os.path.exists(args.file):
        print(f"Error: File not found: {args.file}", file=sys.stderr)
        sys.exit(1)

    data = load_data(args.file)
    result = profile_dataset(data, args.top)

    if args.json:
        print(json.dumps(result, indent=2))
    else:
        print("Data Profile Report")
        print("=" * 60)
        print(f"Rows: {result['row_count']}  |  Columns: {result['column_count']}  |  Columns with issues: {result['columns_with_issues']}")
        print()
        for col in result["columns"]:
            print(f"--- {col['column']} ({col['data_type']}) ---")
            print(f"  Nulls: {col['null_count']} ({col['null_pct']}%)  |  Unique: {col['unique_values']} ({col['cardinality_pct']}%)")
            if col["data_type"] == "numeric":
                print(f"  Min: {col['min']}  Max: {col['max']}  Mean: {col['mean']}  Median: {col['median']}")
                print(f"  Std: {col['std_dev']}  Q25: {col['q25']}  Q75: {col['q75']}")
                if col.get("outlier_count", 0) > 0:
                    print(f"  Outliers: {col['outlier_count']} ({col['outlier_pct']}%)")
            else:
                if "min_length" in col:
                    print(f"  Length: min={col['min_length']}  max={col['max_length']}  avg={col['avg_length']}")
                if col.get("top_values"):
                    top_str = ", ".join(f"{t['value']} ({t['count']})" for t in col["top_values"][:3])
                    print(f"  Top: {top_str}")
            if col["quality_flags"]:
                print(f"  Flags: {', '.join(col['quality_flags'])}")
            print()

    sys.exit(0)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Analyze SQL queries for common performance issues and suggest optimizations.

Parses SQL text and checks for anti-patterns such as SELECT *, missing WHERE
clauses, implicit cartesian joins, unqualified column references in JOINs,
and excessive sub-query nesting.  Reports findings with severity and
actionable recommendations.

Usage:
    python query_optimizer.py --file query.sql
    python query_optimizer.py --sql "SELECT * FROM orders"
    python query_optimizer.py --file query.sql --json
"""

import argparse
import json
import re
import sys


# ---------------------------------------------------------------------------
# Rule definitions
# ---------------------------------------------------------------------------

def _check_select_star(sql: str) -> list:
    """Flag SELECT * usage."""
    issues = []
    for i, line in enumerate(sql.splitlines(), 1):
        if re.search(r"\bSELECT\s+\*", line, re.IGNORECASE):
            issues.append({
                "rule": "SELECT_STAR",
                "severity": "warning",
                "line": i,
                "message": "SELECT * fetches all columns; specify only needed columns to reduce I/O.",
                "suggestion": "Replace SELECT * with an explicit column list.",
            })
    return issues


def _check_missing_where(sql: str) -> list:
    """Flag queries that read large tables without a WHERE clause."""
    issues = []
    # Simplified: look for FROM without a subsequent WHERE at the statement level
    statements = re.split(r";", sql)
    for stmt in statements:
        stmt_stripped = stmt.strip()
        if not stmt_stripped:
            continue
        has_from = re.search(r"\bFROM\b", stmt_stripped, re.IGNORECASE)
        has_where = re.search(r"\bWHERE\b", stmt_stripped, re.IGNORECASE)
        has_limit = re.search(r"\bLIMIT\b", stmt_stripped, re.IGNORECASE)
        if has_from and not has_where and not has_limit:
            issues.append({
                "rule": "MISSING_WHERE",
                "severity": "warning",
                "line": None,
                "message": "Query reads from a table without a WHERE or LIMIT clause.",
                "suggestion": "Add filtering predicates or a LIMIT to avoid full table scans.",
            })
    return issues


def _check_cartesian_join(sql: str) -> list:
    """Flag comma-separated FROM (implicit cross join)."""
    issues = []
    # Match FROM a, b pattern (no JOIN keyword)
    pattern = re.compile(
        r"\bFROM\s+\w+\s*,\s*\w+", re.IGNORECASE
    )
    for i, line in enumerate(sql.splitlines(), 1):
        if pattern.search(line):
            issues.append({
                "rule": "IMPLICIT_CROSS_JOIN",
                "severity": "critical",
                "line": i,
                "message": "Comma-separated FROM clause may produce a cartesian product.",
                "suggestion": "Use explicit JOIN ... ON syntax instead of comma-separated tables.",
            })
    return issues


def _check_subquery_nesting(sql: str) -> list:
    """Flag deeply nested sub-queries (>3 levels)."""
    issues = []
    depth = 0
    max_depth = 0
    for ch in sql:
        if ch == "(":
            depth += 1
            if depth > max_depth:
                max_depth = depth
        elif ch == ")":
            depth = max(0, depth - 1)
    if max_depth > 3:
        issues.append({
            "rule": "DEEP_NESTING",
            "severity": "warning",
            "line": None,
            "message": f"Query has {max_depth} levels of nesting; readability and performance degrade beyond 3.",
            "suggestion": "Refactor deep sub-queries into CTEs (WITH clauses) for clarity and potential optimization.",
        })
    return issues


def _check_or_in_where(sql: str) -> list:
    """Flag excessive OR chains that may prevent index usage."""
    issues = []
    or_count = len(re.findall(r"\bOR\b", sql, re.IGNORECASE))
    if or_count >= 5:
        issues.append({
            "rule": "EXCESSIVE_OR",
            "severity": "info",
            "line": None,
            "message": f"Found {or_count} OR conditions; large OR chains can prevent index usage.",
            "suggestion": "Consider replacing OR chains with IN (...) or UNION ALL for better index utilization.",
        })
    return issues


def _check_functions_on_indexed_columns(sql: str) -> list:
    """Flag common anti-pattern of wrapping indexed columns in functions."""
    issues = []
    patterns = [
        (r"\bWHERE\b.*\b(?:UPPER|LOWER|TRIM|CAST|DATE)\s*\(", "Function on column in WHERE clause may prevent index usage."),
    ]
    for pat, msg in patterns:
        if re.search(pat, sql, re.IGNORECASE | re.DOTALL):
            issues.append({
                "rule": "FUNCTION_ON_COLUMN",
                "severity": "warning",
                "line": None,
                "message": msg,
                "suggestion": "Apply the function to the comparison value instead, or use a computed/expression index.",
            })
    return issues


def _check_order_without_limit(sql: str) -> list:
    """Flag ORDER BY without LIMIT."""
    issues = []
    has_order = re.search(r"\bORDER\s+BY\b", sql, re.IGNORECASE)
    has_limit = re.search(r"\bLIMIT\b", sql, re.IGNORECASE)
    if has_order and not has_limit:
        issues.append({
            "rule": "ORDER_WITHOUT_LIMIT",
            "severity": "info",
            "line": None,
            "message": "ORDER BY without LIMIT sorts the entire result set, which can be expensive.",
            "suggestion": "Add a LIMIT clause if you only need the top/bottom N rows.",
        })
    return issues


ALL_CHECKS = [
    _check_select_star,
    _check_missing_where,
    _check_cartesian_join,
    _check_subquery_nesting,
    _check_or_in_where,
    _check_functions_on_indexed_columns,
    _check_order_without_limit,
]


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def analyze(sql: str) -> dict:
    issues = []
    for check_fn in ALL_CHECKS:
        issues.extend(check_fn(sql))

    severity_order = {"critical": 0, "warning": 1, "info": 2}
    issues.sort(key=lambda x: severity_order.get(x["severity"], 99))

    summary = {
        "total_issues": len(issues),
        "critical": sum(1 for i in issues if i["severity"] == "critical"),
        "warnings": sum(1 for i in issues if i["severity"] == "warning"),
        "info": sum(1 for i in issues if i["severity"] == "info"),
    }
    return {"summary": summary, "issues": issues}


def main():
    parser = argparse.ArgumentParser(
        description="Analyze SQL queries for performance issues and anti-patterns."
    )
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--file", help="Path to a .sql file to analyze")
    group.add_argument("--sql", help="Inline SQL string to analyze")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")
    args = parser.parse_args()

    if args.file:
        try:
            with open(args.file, "r") as f:
                sql = f.read()
        except FileNotFoundError:
            print(f"Error: File not found: {args.file}", file=sys.stderr)
            sys.exit(1)
    else:
        sql = args.sql

    result = analyze(sql)

    if args.json:
        print(json.dumps(result, indent=2))
    else:
        s = result["summary"]
        print("SQL Query Optimization Report")
        print("=" * 50)
        print(f"Issues found: {s['total_issues']}  (critical: {s['critical']}, warnings: {s['warnings']}, info: {s['info']})")
        print()
        if not result["issues"]:
            print("No issues detected. Query looks well-structured.")
        for issue in result["issues"]:
            sev = issue["severity"].upper()
            line_str = f" (line {issue['line']})" if issue["line"] else ""
            print(f"[{sev}] {issue['rule']}{line_str}")
            print(f"  {issue['message']}")
            print(f"  -> {issue['suggestion']}")
            print()

    sys.exit(1 if result["summary"]["critical"] > 0 else 0)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Generate summary reports from CSV or JSON data files.

Produces a structured summary report with key statistics, trends, and
highlights.  Supports markdown and plain-text output.

Usage:
    python report_generator.py --file sales.csv --title "Monthly Sales Report"
    python report_generator.py --file data.json --format markdown --json
    python report_generator.py --file data.csv --group-by region
"""

import argparse
import csv
import json
import math
import os
import sys
from collections import defaultdict
from datetime import datetime


def _is_numeric(value: str) -> bool:
    try:
        float(value)
        return True
    except (ValueError, TypeError):
        return False


def _median(values: list) -> float:
    s = sorted(values)
    n = len(s)
    if n == 0:
        return 0.0
    mid = n // 2
    return (s[mid - 1] + s[mid]) / 2.0 if n % 2 == 0 else s[mid]


def load_data(file_path: str) -> list:
    ext = os.path.splitext(file_path)[1].lower()
    if ext == ".csv":
        with open(file_path, "r", newline="") as f:
            return list(csv.DictReader(f))
    elif ext == ".json":
        with open(file_path, "r") as f:
            data = json.load(f)
            if isinstance(data, list):
                return data
            raise ValueError("JSON must contain an array of objects.")
    else:
        print(f"Error: Unsupported file type '{ext}'.", file=sys.stderr)
        sys.exit(1)


def _detect_numeric_columns(data: list) -> list:
    if not data:
        return []
    cols = []
    for col in data[0].keys():
        sample = [row.get(col) for row in data[:100] if row.get(col) is not None and str(row.get(col)).strip()]
        if sample and sum(1 for v in sample if _is_numeric(str(v))) > len(sample) * 0.8:
            cols.append(col)
    return cols


def _compute_numeric_summary(data: list, col: str) -> dict:
    values = []
    for row in data:
        v = row.get(col)
        if v is not None and _is_numeric(str(v)):
            values.append(float(str(v)))
    if not values:
        return {}
    total = sum(values)
    mean = total / len(values)
    return {
        "column": col,
        "count": len(values),
        "sum": round(total, 2),
        "mean": round(mean, 2),
        "median": round(_median(values), 2),
        "min": round(min(values), 2),
        "max": round(max(values), 2),
    }


def _compute_group_summary(data: list, group_col: str, numeric_cols: list) -> list:
    groups = defaultdict(list)
    for row in data:
        key = str(row.get(group_col, "Unknown"))
        groups[key].append(row)

    summaries = []
    for group_name, rows in sorted(groups.items()):
        entry = {"group": group_name, "count": len(rows)}
        for col in numeric_cols:
            values = []
            for row in rows:
                v = row.get(col)
                if v is not None and _is_numeric(str(v)):
                    values.append(float(str(v)))
            if values:
                entry[f"{col}_sum"] = round(sum(values), 2)
                entry[f"{col}_mean"] = round(sum(values) / len(values), 2)
        summaries.append(entry)
    return summaries


def generate_report(data: list, title: str, group_by: str = None) -> dict:
    numeric_cols = _detect_numeric_columns(data)

    report = {
        "title": title,
        "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "total_rows": len(data),
        "total_columns": len(data[0].keys()) if data else 0,
        "numeric_columns": len(numeric_cols),
    }

    # Numeric summaries
    summaries = []
    for col in numeric_cols:
        s = _compute_numeric_summary(data, col)
        if s:
            summaries.append(s)
    report["numeric_summaries"] = summaries

    # Group-by breakdown
    if group_by and data:
        if group_by not in data[0]:
            print(f"Warning: Column '{group_by}' not found. Skipping group-by.", file=sys.stderr)
        else:
            report["group_by"] = group_by
            report["group_summaries"] = _compute_group_summary(data, group_by, numeric_cols)

    # Highlights
    highlights = []
    for s in summaries:
        if s["max"] > s["mean"] * 3 and s["count"] > 10:
            highlights.append(f"{s['column']}: max ({s['max']}) is >3x the mean ({s['mean']}), indicating outliers.")
    if len(data) > 0 and numeric_cols:
        highlights.append(f"Dataset contains {len(data)} rows across {len(data[0].keys())} columns with {len(numeric_cols)} numeric fields.")
    report["highlights"] = highlights

    return report


def format_markdown(report: dict) -> str:
    lines = [f"# {report['title']}", ""]
    lines.append(f"*Generated: {report['generated_at']}*\n")
    lines.append(f"**Rows:** {report['total_rows']} | **Columns:** {report['total_columns']} | **Numeric:** {report['numeric_columns']}\n")

    if report.get("numeric_summaries"):
        lines.append("## Numeric Summaries\n")
        lines.append("| Column | Count | Sum | Mean | Median | Min | Max |")
        lines.append("|--------|-------|-----|------|--------|-----|-----|")
        for s in report["numeric_summaries"]:
            lines.append(f"| {s['column']} | {s['count']} | {s['sum']} | {s['mean']} | {s['median']} | {s['min']} | {s['max']} |")
        lines.append("")

    if report.get("group_summaries"):
        lines.append(f"## Breakdown by {report['group_by']}\n")
        if report["group_summaries"]:
            headers = list(report["group_summaries"][0].keys())
            lines.append("| " + " | ".join(headers) + " |")
            lines.append("|" + "|".join(["---"] * len(headers)) + "|")
            for row in report["group_summaries"]:
                lines.append("| " + " | ".join(str(row.get(h, "")) for h in headers) + " |")
        lines.append("")

    if report.get("highlights"):
        lines.append("## Highlights\n")
        for h in report["highlights"]:
            lines.append(f"- {h}")

    return "\n".join(lines)


def format_text(report: dict) -> str:
    lines = [report["title"], "=" * len(report["title"])]
    lines.append(f"Generated: {report['generated_at']}")
    lines.append(f"Rows: {report['total_rows']}  Columns: {report['total_columns']}  Numeric: {report['numeric_columns']}")
    lines.append("")

    if report.get("numeric_summaries"):
        lines.append("Numeric Summaries:")
        lines.append("-" * 40)
        for s in report["numeric_summaries"]:
            lines.append(f"  {s['column']}: sum={s['sum']}, mean={s['mean']}, median={s['median']}, range=[{s['min']}, {s['max']}]")
        lines.append("")

    if report.get("group_summaries"):
        lines.append(f"Breakdown by {report['group_by']}:")
        lines.append("-" * 40)
        for row in report["group_summaries"]:
            parts = [f"{k}={v}" for k, v in row.items()]
            lines.append(f"  {', '.join(parts)}")
        lines.append("")

    if report.get("highlights"):
        lines.append("Highlights:")
        for h in report["highlights"]:
            lines.append(f"  * {h}")

    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(description="Generate summary reports from CSV or JSON data.")
    parser.add_argument("--file", required=True, help="Path to CSV or JSON data file")
    parser.add_argument("--title", default="Data Summary Report", help="Report title")
    parser.add_argument("--group-by", help="Column name to group results by")
    parser.add_argument("--format", choices=["text", "markdown"], default="text", help="Output format (default: text)")
    parser.add_argument("--json", action="store_true", help="Output raw report data as JSON")
    args = parser.parse_args()

    if not os.path.exists(args.file):
        print(f"Error: File not found: {args.file}", file=sys.stderr)
        sys.exit(1)

    data = load_data(args.file)
    if not data:
        print("Error: No data rows found in file.", file=sys.stderr)
        sys.exit(1)

    report = generate_report(data, args.title, args.group_by)

    if args.json:
        print(json.dumps(report, indent=2))
    elif args.format == "markdown":
        print(format_markdown(report))
    else:
        print(format_text(report))

    sys.exit(0)


if __name__ == "__main__":
    main()