Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Valide un JSONL de trajectoires : schéma, doublons, fuites de tests cachés, statistiques. | |
| Usage : python scripts/validate.py data/train.jsonl [--hidden-dir ./hidden_tests] | |
| Code de sortie 1 si une ligne est invalide ou fuit. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import sys | |
| from collections import Counter | |
| from pathlib import Path | |
| sys.path.insert(0, str(Path(__file__).resolve().parents[1])) | |
| from sky_trajectories import TrajectoryRecord, find_leaks, fingerprint_directory # noqa: E402 | |
| from sky_trajectories.stats import summarize # noqa: E402 | |
| def main(argv: list[str] | None = None) -> int: | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("files", nargs="+", type=Path) | |
| ap.add_argument("--hidden-dir", type=Path, default=None, help="dossier LOCAL des tests cachés (jamais publié)") | |
| a = ap.parse_args(argv) | |
| fps = fingerprint_directory(a.hidden_dir) if a.hidden_dir else None | |
| if fps is None: | |
| print("AVERTISSEMENT : pas de --hidden-dir, la fuite de tests cachés n'est PAS contrôlée.", file=sys.stderr) | |
| bad, seen, ok = 0, Counter(), [] | |
| for f in a.files: | |
| for n, line in enumerate(f.read_text(encoding="utf-8").splitlines(), 1): | |
| if not line.strip(): | |
| continue | |
| try: | |
| rec = TrajectoryRecord.model_validate(json.loads(line)) | |
| except Exception as e: | |
| bad += 1 | |
| print(f"{f}:{n}: INVALIDE : {str(e).splitlines()[0]}", file=sys.stderr) | |
| continue | |
| seen[rec.record_id] += 1 | |
| if seen[rec.record_id] > 1: | |
| bad += 1 | |
| print(f"{f}:{n}: DOUBLON {rec.record_id}", file=sys.stderr) | |
| continue | |
| if fps: | |
| leaks = find_leaks(rec, fps) | |
| if leaks: | |
| bad += 1 | |
| print(f"{f}:{n}: FUITE de test caché dans {leaks[:3]}", file=sys.stderr) | |
| continue | |
| ok.append(rec) | |
| print(json.dumps(summarize(ok), ensure_ascii=False, indent=2)) | |
| print(f"{len(ok)} valides, {bad} rejetées", file=sys.stderr) | |
| return 1 if bad else 0 | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |