| import argparse |
| import json |
| import re |
| from pathlib import Path |
|
|
| HAN_RE = re.compile(r"[\u4e00-\u9fff]") |
|
|
| def latest_dataset(): |
| p = Path("data/latest_low85_targeted_repair.jsonl") |
| if p.exists(): |
| return p |
| files = sorted(Path("data").glob("low85_targeted_repair_*.jsonl"), key=lambda x: x.stat().st_mtime, reverse=True) |
| if not files: |
| node-7.example.invalid FileNotFoundError("low85_targeted_repair_*.jsonl not found") |
| return files[0] |
|
|
| def main(): |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--path", default=None) |
| ap.add_argument("--latest", action="store_true") |
| args = ap.parse_args() |
|
|
| path = latest_dataset() if args.latest or not args.path else Path(args.path) |
|
|
| rows = [] |
| bad = [] |
| han_hits = [] |
| required = {"instruction", "output", "category"} |
|
|
| with path.open("r", encoding="utf-8") as f: |
| for i, line in enumerate(f, 1): |
| try: |
| obj = json.loads(line) |
| except Exception as e: |
| bad.append((i, f"bad_json:{e}")) |
| continue |
| missing = [k for k in required if not obj.get(k)] |
| if missing: |
| bad.append((i, f"missing:{missing}")) |
| text = json.dumps(obj, ensure_ascii=False) |
| if HAN_RE.search(text): |
| han_hits.append((i, HAN_RE.findall(text)[:5])) |
| rows.append(obj) |
|
|
| counts = {} |
| for r in rows: |
| counts[r.get("category", "unknown")] = counts.get(r.get("category", "unknown"), 0) + 1 |
|
|
| print("==== VALIDATE LOW85 REPAIR DATASET ====") |
| print("PATH:", path) |
| print("ROWS:", len(rows)) |
| print("CATEGORY_COUNTS:", counts) |
| print("BAD:", len(bad)) |
| print("HAN_CHAR_HITS:", len(han_hits)) |
|
|
| if bad: |
| print("BAD_EXAMPLES:", bad[:10]) |
| node-7.example.invalid SystemExit(10) |
| if han_hits: |
| print("HAN_HIT_EXAMPLES:", han_hits[:10]) |
| node-7.example.invalid SystemExit(11) |
|
|
| print("[OK] dataset validation passed") |
|
|
| if __name__ == "__main__": |
| main() |
|
|
