scripts/run_validation.py

#!/usr/bin/env python3
"""Run validation and output results to a file."""
import json
from pathlib import Path

schema_path = Path(__file__).parent.parent / "schemas" / "training_example.schema.json"
train_path = Path(__file__).parent.parent / "datasets" / "mythos_coder_train.jsonl"
output_path = Path(__file__).parent.parent / "validation_results.txt"

with open(schema_path, "r", encoding="utf-8") as f:
    schema = json.load(f)

required = schema.get("required", [])
allowed_types = schema["properties"]["task_type"]["enum"]
allowed_difficulties = schema["properties"]["difficulty"]["enum"]

results = []
valid_count = 0
invalid_count = 0

with open(train_path, "r", encoding="utf-8") as f:
    for line_num, line in enumerate(f, 1):
        line = line.strip()
        if not line:
            continue

        try:
            example = json.loads(line)
            errors = []

            for field in required:
                if field not in example:
                    errors.append(f"Missing: {field}")

            if example.get("task_type") not in allowed_types:
                errors.append(f"Invalid task_type: {example.get('task_type')}")

            if example.get("difficulty") not in allowed_difficulties:
                errors.append(f"Invalid difficulty: {example.get('difficulty')}")

            score = example.get("quality_score")
            if not isinstance(score, int) or score < 1 or score > 10:
                errors.append(f"Invalid quality_score: {score}")

            if not isinstance(example.get("investigation_steps"), list):
                errors.append("investigation_steps must be array")

            if errors:
                invalid_count += 1
                results.append(f"Line {line_num}: {example.get('id', 'unknown')} - FAIL")
                for e in errors:
                    results.append(f"  - {e}")
            else:
                valid_count += 1
                results.append(f"Line {line_num}: {example.get('id', 'unknown')} - PASS")

        except json.JSONDecodeError as e:
            invalid_count += 1
            results.append(f"Line {line_num}: JSON ERROR - {e}")

results.insert(0, f"VALIDATION RESULTS for {train_path}")
results.insert(1, "=" * 50)
results.insert(2, f"Valid: {valid_count}, Invalid: {invalid_count}")
results.insert(3, "")

with open(output_path, "w", encoding="utf-8") as f:
    f.write("\n".join(results) + "\n")

print(f"Validation complete. Results written to {output_path}")