scripts/append_example.py

#!/usr/bin/env python3
"""
append_example.py

Appends a new training example to datasets/mythos_coder_train.jsonl.

Usage:
    python append_example.py --file example.json
    python append_example.py --stdin < example.json
    python append_example.py --inline '{"id":"...",...}'

The example will be validated against the schema before appending.
"""

import argparse
import json
import sys
from pathlib import Path


def load_schema():
    """Load the JSON schema for validation."""
    schema_path = Path(__file__).parent.parent / "schemas" / "training_example.schema.json"
    with open(schema_path, "r", encoding="utf-8") as f:
        return json.load(f)


def validate_example(example, schema):
    """Validate a single example against the schema."""
    required = schema.get("required", [])
    errors = []

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

    if "task_type" in example:
        allowed_types = schema["properties"]["task_type"]["enum"]
        if example["task_type"] not in allowed_types:
            errors.append(f"Invalid task_type: {example['task_type']}. Must be one of: {allowed_types}")

    if "difficulty" in example:
        allowed_difficulties = schema["properties"]["difficulty"]["enum"]
        if example["difficulty"] not in allowed_difficulties:
            errors.append(f"Invalid difficulty: {example['difficulty']}. Must be one of: {allowed_difficulties}")

    if "quality_score" in example:
        score = example["quality_score"]
        if not isinstance(score, int) or score < 1 or score > 10:
            errors.append(f"Invalid quality_score: {score}. Must be integer 1-10.")

    if "investigation_steps" in example:
        if not isinstance(example["investigation_steps"], list):
            errors.append("investigation_steps must be an array")
        elif not all(isinstance(step, str) for step in example["investigation_steps"]):
            errors.append("All investigation_steps must be strings")

    return errors


def append_example(example, train_path):
    """Append a single example to the training file."""
    with open(train_path, "a", encoding="utf-8") as f:
        f.write(json.dumps(example, ensure_ascii=False) + "\n")


def main():
    parser = argparse.ArgumentParser(description="Append a training example to the dataset")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--file", "-f", help="Path to JSON file containing the example")
    group.add_argument("--stdin", "-s", action="store_true", help="Read example from stdin")
    group.add_argument("--inline", "-i", help="Inline JSON string")

    args = parser.parse_args()

    # Load the example
    try:
        if args.file:
            with open(args.file, "r", encoding="utf-8") as f:
                example = json.load(f)
        elif args.stdin:
            example = json.load(sys.stdin)
        else:
            example = json.loads(args.inline)
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON - {e}", file=sys.stderr)
        sys.exit(1)

    # Load schema and validate
    try:
        schema = load_schema()
    except FileNotFoundError:
        print("Error: Schema file not found. Run from project root.", file=sys.stderr)
        sys.exit(1)

    errors = validate_example(example, schema)
    if errors:
        print("Validation failed:", file=sys.stderr)
        for error in errors:
            print(f"  - {error}", file=sys.stderr)
        sys.exit(1)

    # Append to training file
    project_root = Path(__file__).parent.parent
    train_path = project_root / "datasets" / "mythos_coder_train.jsonl"
    train_path.parent.mkdir(parents=True, exist_ok=True)

    append_example(example, train_path)
    print(f"Successfully appended example '{example.get('id', 'unknown')}' to {train_path}")


if __name__ == "__main__":
    main()