scripts/audit_dataset_quality.py

#!/usr/bin/env python3
"""Audit canonical JSONL rows in data/converted/ for quality."""

from __future__ import annotations

import csv
import json
import sys
from collections import Counter
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from dataset_quality_utils import (
    SCORE_KEEP,
    SCORE_REPAIR,
    SCORE_WEAK,
    repair_recommendation,
    score_band,
    score_row,
)

ROOT = Path(__file__).resolve().parent.parent
CONVERTED_DIR = ROOT / "data" / "converted"
AUDIT_DIR = ROOT / "data" / "audit"
REPORT_PATH = AUDIT_DIR / "dataset_quality_report.md"
WEAK_PATH = AUDIT_DIR / "weak_rows.jsonl"
STRONG_PATH = AUDIT_DIR / "strong_rows.jsonl"
SCORES_CSV = AUDIT_DIR / "row_scores.csv"


def main() -> int:
    AUDIT_DIR.mkdir(parents=True, exist_ok=True)
    all_results: list[dict] = []
    issue_counter: Counter = Counter()

    for path in sorted(CONVERTED_DIR.glob("*.jsonl")):
        with path.open("r", encoding="utf-8") as f:
            for line_num, line in enumerate(f, 1):
                if not line.strip():
                    continue
                row = json.loads(line)
                score, issues = score_row(row)
                band = score_band(score)
                rec = repair_recommendation(score, issues)
                for issue in issues:
                    issue_counter[issue.split(":")[0]] += 1
                entry = {
                    "source_file": path.name,
                    "line_number": line_num,
                    "id": row.get("id", ""),
                    "task_type": row.get("task_type", ""),
                    "score": score,
                    "band": band,
                    "issues": issues,
                    "repair_recommendation": rec,
                    "row": row,
                }
                all_results.append(entry)

    weak = [e for e in all_results if e["score"] < SCORE_KEEP]
    strong = [e for e in all_results if e["score"] >= SCORE_KEEP]

    with WEAK_PATH.open("w", encoding="utf-8") as f:
        for e in weak:
            f.write(json.dumps(e, ensure_ascii=False) + "\n")

    with STRONG_PATH.open("w", encoding="utf-8") as f:
        for e in strong:
            f.write(json.dumps(e, ensure_ascii=False) + "\n")

    with SCORES_CSV.open("w", encoding="utf-8", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["source_file", "line_number", "id", "task_type", "score", "band", "issues"])
        for e in all_results:
            writer.writerow([
                e["source_file"], e["line_number"], e["id"], e["task_type"],
                e["score"], e["band"], ";".join(e["issues"]),
            ])

    bands = Counter(e["band"] for e in all_results)
    lines = [
        "# Dataset Quality Report",
        "",
        f"Scanned: `{CONVERTED_DIR}`",
        f"Total rows: **{len(all_results)}**",
        "",
        "## Score bands",
        "",
        f"- excellent (90-100): {bands.get('excellent', 0)}",
        f"- keep (80-89): {bands.get('keep', 0)}",
        f"- repair (65-79): {bands.get('repair', 0)}",
        f"- weak (40-64): {bands.get('weak', 0)}",
        f"- reject (0-39): {bands.get('reject', 0)}",
        "",
        f"- Strong rows written: `{STRONG_PATH}` ({len(strong)})",
        f"- Weak rows written: `{WEAK_PATH}` ({len(weak)})",
        "",
        "## Top issues",
        "",
    ]
    for issue, count in issue_counter.most_common(25):
        lines.append(f"- `{issue}`: {count}")

    lines.extend(["", "## Sample weak rows (first 10)", ""])
    for e in sorted(weak, key=lambda x: x["score"])[:10]:
        lines.append(f"### `{e['id']}` score={e['score']} ({e['source_file']}:{e['line_number']})")
        lines.append(f"- Issues: {', '.join(e['issues'][:8])}")
        lines.append(f"- Recommendation: {e['repair_recommendation']}")
        lines.append("")

    lines.extend(["", "## Sample strong rows (first 10)", ""])
    for e in sorted(strong, key=lambda x: -x["score"])[:10]:
        lines.append(f"### `{e['id']}` score={e['score']}")
        lines.append(f"- prompt: {e['row'].get('user_prompt', '')[:120]}...")
        lines.append("")

    REPORT_PATH.write_text("\n".join(lines), encoding="utf-8")
    print(f"Audited {len(all_results)} rows")
    print(f"Strong: {len(strong)}, Weak: {len(weak)}")
    print(f"Wrote {REPORT_PATH}")
    return 0


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