scripts/run_quality_pipeline.py

#!/usr/bin/env python3
"""Run the full Mythos dataset quality pipeline."""

from __future__ import annotations

import json
import re
import subprocess
import sys
from collections import Counter
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
SCRIPTS = ROOT / "scripts"
AUDIT_DIR = ROOT / "data" / "audit"
REPORT_PATH = AUDIT_DIR / "full_quality_pipeline_report.md"


def run_script(name: str) -> tuple[int, str]:
    path = SCRIPTS / name
    result = subprocess.run(
        [sys.executable, str(path)],
        cwd=str(ROOT),
        capture_output=True,
        text=True,
    )
    output = (result.stdout or "") + (result.stderr or "")
    return result.returncode, output.strip()


def count_jsonl(path: Path) -> int:
    if not path.exists():
        return 0
    return sum(1 for line in path.open(encoding="utf-8") if line.strip())


def load_issue_counts(csv_path: Path) -> Counter:
    # read from weak rows instead
    c: Counter = Counter()
    weak = AUDIT_DIR / "weak_rows.jsonl"
    if weak.exists():
        for line in weak.open(encoding="utf-8"):
            if line.strip():
                entry = json.loads(line)
                for issue in entry.get("issues", []):
                    c[issue.split(":")[0]] += 1
    return c


def main() -> int:
    AUDIT_DIR.mkdir(parents=True, exist_ok=True)
    steps = [
        "audit_dataset_quality.py",
        "repair_weak_rows.py",
        "build_clean_training_set.py",
        "build_sft_messages.py",
        "audit_sft_messages.py",
        "preview_code_output_examples.py",
    ]
    logs: dict[str, str] = {}
    failed = False
    for step in steps:
        code, output = run_script(step)
        logs[step] = output
        print(f"\n=== {step} (exit {code}) ===\n{output[:2000]}")
        if code != 0:
            failed = True

    total = count_jsonl(AUDIT_DIR / "row_scores.csv")  # csv not jsonl
    # count from audit report data
    strong = count_jsonl(AUDIT_DIR / "strong_rows.jsonl")
    weak = count_jsonl(AUDIT_DIR / "weak_rows.jsonl")
    repaired = count_jsonl(ROOT / "data" / "repaired" / "repaired_rows.jsonl")
    rejected = count_jsonl(ROOT / "data" / "rejected" / "rejected_rows.jsonl")
    clean_canonical = count_jsonl(ROOT / "data" / "train" / "mythos_coder_clean_canonical.jsonl")
    clean_sft = count_jsonl(ROOT / "data" / "train" / "mythos_sft_messages_clean.jsonl")
    weak_sft = count_jsonl(AUDIT_DIR / "weak_sft_messages.jsonl")
    hard_fails = 0
    if (AUDIT_DIR / "sft_quality_report.md").exists():
        m = re.search(r"Hard fails: \*\*(\d+)\*\*", (AUDIT_DIR / "sft_quality_report.md").read_text(encoding="utf-8"))
        if m:
            hard_fails = int(m.group(1))

    issues = load_issue_counts(AUDIT_DIR / "row_scores.csv")

    # task type distribution from clean canonical
    task_before = Counter()
    task_after = Counter()
    for path in (ROOT / "data" / "converted").glob("*.jsonl"):
        for line in path.open(encoding="utf-8"):
            if line.strip():
                task_before[json.loads(line).get("task_type", "?")] += 1
    clean_path = ROOT / "data" / "train" / "mythos_coder_clean_canonical.jsonl"
    if clean_path.exists():
        for line in clean_path.open(encoding="utf-8"):
            if line.strip():
                task_after[json.loads(line).get("task_type", "?")] += 1

    sft_safe = "NOT SAFE"
    sft_report = AUDIT_DIR / "sft_quality_report.md"
    if sft_report.exists():
        text = sft_report.read_text(encoding="utf-8")
        if "**SAFE** to train" in text or "Hard fails: **0**" in text and "weak=0" not in logs.get("audit_sft_messages.py", ""):
            if "Hard fails: **0**" in text:
                sft_safe = "SAFE"

    lines = [
        "# Full Quality Pipeline Report",
        "",
        f"Pipeline {'FAILED' if failed else 'COMPLETED'}",
        "",
        "## Counts",
        "",
        f"- Converted rows scanned: **{strong + weak}**",
        f"- Strong rows: **{strong}**",
        f"- Weak rows: **{weak}**",
        f"- Repaired rows: **{repaired}**",
        f"- Rejected rows: **{rejected}**",
        f"- Clean canonical rows: **{clean_canonical}**",
        f"- Clean SFT rows: **{clean_sft}**",
        f"- Weak SFT rows: **{weak_sft}** (hard fails: {hard_fails})",
        "",
        "## Top repeated issues",
        "",
    ]
    for issue, count in issues.most_common(20):
        lines.append(f"- `{issue}`: {count}")

    lines.extend([
        "",
        "## Task type distribution",
        "",
        "### Before (all converted)",
        "",
    ])
    for t, c in task_before.most_common():
        lines.append(f"- `{t}`: {c}")
    lines.extend(["", "### After (clean canonical)", ""])
    for t, c in task_after.most_common():
        lines.append(f"- `{t}`: {c}")

    lines.extend([
        "",
        "## Training safety",
        "",
        f"Clean SFT status: **{sft_safe}**",
        "",
        "## Step logs",
        "",
    ])
    for step, output in logs.items():
        lines.append(f"### {step}")
        lines.append("```")
        lines.append(output[:1500] or "(no output)")
        lines.append("```")
        lines.append("")

    lines.extend([
        "## Next command to train",
        "",
        "Preview first:",
        "```bash",
        "python scripts/preview_code_output_examples.py",
        "```",
        "",
        "Then train (only if preview passes):",
        "```bash",
        "python scripts/train_lora_sft.py",
        "```",
        "",
        "ZeroGPU: copy `data/train/mythos_sft_messages_clean.jsonl` to Space before training.",
        "",
        "## Reports",
        "",
        "- `data/audit/dataset_quality_report.md`",
        "- `data/audit/repair_report.md`",
        "- `data/audit/sft_quality_report.md`",
        "- `data/audit/code_output_preview.md`",
    ])

    REPORT_PATH.write_text("\n".join(lines), encoding="utf-8")
    print(f"\nWrote {REPORT_PATH}")
    return 1 if failed else 0


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