| """Read question files and normalise them to one internal format.
|
|
|
| The official dev/test format is released on Oct 5. Until then we accept the
|
| common field names below, so only this file needs adjusting if the real
|
| format differs.
|
|
|
| Internal record (a dict):
|
| id str
|
| level "easy" | "medium" | "hard" (lower-cased)
|
| question str
|
| options list[str]
|
| gold int | None 0-based index of the correct option (None for test)
|
| """
|
|
|
| from __future__ import annotations
|
|
|
| import json
|
| from pathlib import Path
|
|
|
| QUESTION_KEYS = ("question", "q", "prompt", "stem")
|
| OPTIONS_KEYS = ("options", "choices", "answers", "candidates")
|
| GOLD_KEYS = ("answer", "label", "gold", "correct", "answer_idx")
|
| LEVEL_KEYS = ("level", "difficulty")
|
| LETTERS = "ABCDEFGH"
|
|
|
|
|
| def read_jsonl(path: str | Path) -> list[dict]:
|
| """Read JSON Lines, or a single JSON array (the official dev file is an array).
|
|
|
| A folder is also accepted: the official TEST release is a folder tree
|
| (easy/ medium/ hard/, one JSON array per subject). All *.json / *.jsonl
|
| files under it are read in sorted path order and concatenated.
|
| """
|
| if Path(path).is_dir():
|
| files = sorted(
|
| f for f in Path(path).rglob("*")
|
| if f.suffix in (".json", ".jsonl") and not any(part.startswith(".") for part in f.parts)
|
| )
|
| if not files:
|
| raise ValueError(f"{path}: folder has no .json/.jsonl files")
|
| rows = []
|
| for f in files:
|
| rows.extend(read_jsonl(f))
|
| return rows
|
| text = Path(path).read_text(encoding="utf-8-sig")
|
| if text.lstrip().startswith("["):
|
| try:
|
| data = json.loads(text)
|
| except json.JSONDecodeError as e:
|
| raise ValueError(f"{path}: invalid JSON array ({e})") from e
|
| if not all(isinstance(r, dict) for r in data):
|
| raise ValueError(f"{path}: expected a JSON array of objects")
|
| return data
|
| rows = []
|
| with Path(path).open(encoding="utf-8-sig") as f:
|
| for line_no, line in enumerate(f, 1):
|
| line = line.strip()
|
| if not line:
|
| continue
|
| try:
|
| rows.append(json.loads(line))
|
| except json.JSONDecodeError as e:
|
| raise ValueError(f"{path}:{line_no}: invalid JSON ({e})") from e
|
| return rows
|
|
|
|
|
| def write_jsonl(rows: list[dict], path: str | Path) -> None:
|
| path = Path(path)
|
| path.parent.mkdir(parents=True, exist_ok=True)
|
| with path.open("w", encoding="utf-8", newline="\n") as f:
|
| for row in rows:
|
| f.write(json.dumps(row, ensure_ascii=False) + "\n")
|
|
|
|
|
| def level_of(row: dict):
|
| """Difficulty label: top-level level/difficulty, or metadata.difficulty (official format)."""
|
| value = _first(row, LEVEL_KEYS)
|
| if value is None and isinstance(row.get("metadata"), dict):
|
| value = _first(row["metadata"], LEVEL_KEYS)
|
| return value
|
|
|
|
|
| def row_ids(rows: list[dict]) -> list[str]:
|
| """Stable ids: 'id' if present, else 'q_no' when unique, else the row position."""
|
| if all("id" in r for r in rows):
|
| return [str(r["id"]) for r in rows]
|
| qn = [r.get("q_no") for r in rows]
|
| if all(q is not None for q in qn) and len(set(qn)) == len(qn):
|
| return [str(q) for q in qn]
|
| return [str(r["id"]) if "id" in r else f"row_{i:05d}" for i, r in enumerate(rows)]
|
|
|
|
|
| def _first(row: dict, keys: tuple[str, ...]):
|
| for key in keys:
|
| if key in row and row[key] is not None:
|
| return row[key]
|
| return None
|
|
|
|
|
| def _options(row: dict) -> list[str]:
|
| opts = _first(row, OPTIONS_KEYS)
|
| if isinstance(opts, dict):
|
| opts = [opts[k] for k in sorted(opts)]
|
| if opts is None:
|
| for pattern in ("option{}", "option_{}", "opt{}", "choice{}"):
|
| found = [row.get(pattern.format(i)) for i in range(1, 9)]
|
| found = [o for o in found if o is not None]
|
| if found:
|
| opts = found
|
| break
|
| if opts is None:
|
| found = [row.get(c) for c in LETTERS if row.get(c) is not None]
|
| opts = found or None
|
| if not opts:
|
| raise ValueError(f"no options found in record: {list(row)}")
|
| return [str(o).strip() for o in opts]
|
|
|
|
|
| def _raw_gold(value):
|
| """Return gold as an int (base not yet known) or a letter index marker."""
|
| if value is None:
|
| return None
|
| if isinstance(value, bool):
|
| raise ValueError(f"unexpected gold value {value!r}")
|
| if isinstance(value, int):
|
| return value
|
| text = str(value).strip()
|
| if text.isdigit():
|
| return int(text)
|
| if len(text) == 1 and text.upper() in LETTERS:
|
| return ("letter", LETTERS.index(text.upper()))
|
| raise ValueError(f"cannot parse gold answer {value!r}")
|
|
|
|
|
| def load_questions(path: str | Path, answer_base="auto") -> list[dict]:
|
| """Load a dev/test/train file into internal records.
|
|
|
| answer_base: "auto" guesses from the data (any 0 -> 0-based, otherwise
|
| 1-based). After Oct 5, confirm this once and set it explicitly in the
|
| config - a wrong base shifts every answer and silently ruins the score.
|
| """
|
| raw = read_jsonl(path)
|
| ids = row_ids(raw)
|
| records = []
|
| for i, row in enumerate(raw):
|
| rid = ids[i]
|
| question = _first(row, QUESTION_KEYS)
|
| if question is None:
|
| raise ValueError(f"{path}: record {rid} has no question field")
|
| level_raw = level_of(row)
|
| level = str(level_raw or "unknown").strip().lower()
|
| records.append(
|
| {
|
| "id": str(rid),
|
| "level": level,
|
| "level_raw": level_raw,
|
| "question": str(question).strip(),
|
| "options": _options(row),
|
| "subject": row.get("subject"),
|
| "category": row.get("category"),
|
| "q_no": row.get("q_no"),
|
| "_raw_gold": _raw_gold(_first(row, GOLD_KEYS)),
|
| }
|
| )
|
|
|
| numeric = [r["_raw_gold"] for r in records if isinstance(r["_raw_gold"], int)]
|
| if answer_base == "auto":
|
| base = 0 if (numeric and min(numeric) == 0) else 1
|
| else:
|
| base = int(answer_base)
|
|
|
| for r in records:
|
| g = r.pop("_raw_gold")
|
| if g is None:
|
| r["gold"] = None
|
| elif isinstance(g, tuple):
|
| r["gold"] = g[1]
|
| else:
|
| r["gold"] = g - base
|
| if r["gold"] is not None and not 0 <= r["gold"] < len(r["options"]):
|
|
|
| raw = str(r["gold"] + base)
|
| hits = [i for i, o in enumerate(r["options"]) if str(o).strip().lstrip("0") == raw.lstrip("0")]
|
| if len(hits) == 1:
|
| print(f"[data] {path}: record {r['id']} answer {raw} matched option text -> option {hits[0] + 1}")
|
| r["gold"] = hits[0]
|
| continue
|
| raise ValueError(
|
| f"{path}: record {r['id']} gold {r['gold'] + base} out of range "
|
| f"for {len(r['options'])} options (answer_base={base})"
|
| )
|
| return records
|
|
|
|
|
| def prediction_row(record: dict, answer: str) -> dict:
|
| """One line of predictions.jsonl in the official Codabench format:
|
| {"q_no":1,"subject":"Arts","category":"humanities","level":"easy","answer":"2"}
|
| Files without q_no (our own sample/train data) fall back to an "id" key."""
|
| level = record["level_raw"] if record.get("level_raw") is not None else record["level"]
|
|
|
|
|
| s = lambda v: "" if v is None else str(v)
|
| if record.get("q_no") is not None:
|
| q = record["q_no"]
|
| q = int(q) if str(q).strip().isdigit() else s(q)
|
| return {"q_no": q, "subject": s(record.get("subject")), "category": s(record.get("category")),
|
| "level": s(level), "answer": s(answer)}
|
| return {"id": s(record["id"]), "level": s(level), "answer": s(answer)}
|
|
|
|
|
| def match_key(d: dict):
|
| """Key that links a prediction line to its question: (subject, level, q_no) in
|
| the official format (q_no restarts per subject file, and the TEST set has the
|
| same subject in several levels), else the id."""
|
| if d.get("q_no") is not None:
|
| return ("q", str(d.get("subject")), str(d.get("level") or "").strip().lower(), str(d["q_no"]))
|
| return ("id", str(d["id"]))
|
|
|
|
|
| def has_gold(records: list[dict]) -> bool:
|
| return bool(records) and all(r["gold"] is not None for r in records)
|
|
|
