convert_recipes_export.py

"""Convert Unsloth Data Recipes export to Alpaca fine-tune format.

Input:  build/train_en.raw.jsonl  (question / answer / evidence_quote fields)
Output: build/train_en.jsonl      (instruction / input / output / language fields)

Usage:
    python convert_recipes_export.py
    python convert_recipes_export.py --input build/train_en.raw.jsonl --output build/train_en.jsonl
"""
from __future__ import annotations

import argparse
import json
from pathlib import Path


def convert(src: Path, dst: Path) -> int:
    records = []
    skipped = 0

    with src.open(encoding="utf-8") as f:
        for raw in f:
            raw = raw.strip()
            if not raw:
                continue
            r = json.loads(raw)
            q = r.get("question", "").strip()
            a = r.get("answer", "").strip()
            ev = r.get("evidence_quote", "").strip()
            if not q or not a or not ev:
                skipped += 1
                continue
            records.append({
                "instruction": q,
                "input": ev,
                "output": a,
                "language": "en",
            })

    dst.parent.mkdir(parents=True, exist_ok=True)
    with dst.open("w", encoding="utf-8") as f:
        for rec in records:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")

    print(f"wrote {len(records)} records  (skipped {skipped} incomplete)")
    return len(records)


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--input", default="build/train_en.raw.jsonl", type=Path)
    parser.add_argument("--output", default="build/train_en.jsonl", type=Path)
    args = parser.parse_args()

    if not args.input.exists():
        raise SystemExit(f"Input not found: {args.input}")

    n = convert(args.input, args.output)
    if n < 100:
        raise SystemExit(f"Only {n} records — check the raw export for issues")

    # Quick schema sanity check
    with args.output.open(encoding="utf-8") as f:
        first = json.loads(f.readline())
    required = {"instruction", "input", "output", "language"}
    missing = required - set(first.keys())
    if missing:
        raise SystemExit(f"Schema error: missing fields {missing}")
    if first["language"] != "en":
        raise SystemExit(f"Schema error: expected language=en, got {first['language']}")
    print("schema OK")


if __name__ == "__main__":
    main()