aops02/math-annotation-demo
0
1#!/usr/bin/env python32"""Prepare Potato annotation data from the math annotator Excel workbook.3 4This script intentionally uses only the Python standard library so it can run5in a fresh environment without openpyxl/pandas.6"""7 8from __future__ import annotations9 10import argparse11import csv12import html13import json14import re15import sys16import textwrap17import zipfile18from collections import defaultdict19from pathlib import Path20from xml.etree import ElementTree as ET21 22 23ROOT = Path(__file__).resolve().parents[2]24DEFAULT_INPUT = ROOT / "annotated_data" / "math_annotator_train&test_sets_simple_2q_train_2q_test.xlsx"25DEFAULT_OUTPUT_DIR = Path(__file__).resolve().parents[1] / "my-annotation-task" / "data"26 27SHEET_TO_SPLIT = {28 "math_annotator_training_set": "train",29 "math_annotator_testing_set": "test",30}31 32DISPLAY_COLUMNS = ["dialog_context", "correct_solution", "tutor_response"]33 34DIMENSIONS = [35 "Content Correctness",36 "Learner-State Assessment",37 "Issue Localization",38 "Disclosure Appropriateness",39 "Providing Guidance",40 "Coherence",41 "Actionability",42 "Clarity",43 "Conciseness",44 "Humanness",45]46 47VALID_LABELS = {"Yes", "To some extent", "No"}48 49NS = {50 "a": "http://schemas.openxmlformats.org/spreadsheetml/2006/main",51 "r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships",52}53 54 55def column_index(cell_ref: str) -> int:56 match = re.match(r"([A-Z]+)", cell_ref or "A")57 if not match:58 return 059 index = 060 for char in match.group(1):61 index = index * 26 + (ord(char) - ord("A") + 1)62 return index - 163 64 65def load_shared_strings(archive: zipfile.ZipFile) -> list[str]:66 if "xl/sharedStrings.xml" not in archive.namelist():67 return []68 root = ET.fromstring(archive.read("xl/sharedStrings.xml"))69 strings: list[str] = []70 for string_item in root.findall("a:si", NS):71 strings.append("".join(node.text or "" for node in string_item.findall(".//a:t", NS)))72 return strings73 74 75def get_cell_text(cell: ET.Element, shared_strings: list[str]) -> str:76 cell_type = cell.attrib.get("t")77 value_node = cell.find("a:v", NS)78 if cell_type == "s" and value_node is not None and value_node.text:79 return shared_strings[int(value_node.text)]80 if cell_type == "inlineStr":81 return "".join(node.text or "" for node in cell.findall(".//a:t", NS))82 if value_node is not None:83 return value_node.text or ""84 return ""85 86 87def resolve_sheet_path(target: str) -> str:88 target = target.lstrip("/")89 if target.startswith("xl/"):90 return target91 return f"xl/{target}"92 93 94def read_workbook(path: Path) -> dict[str, list[list[str]]]:95 with zipfile.ZipFile(path) as archive:96 shared_strings = load_shared_strings(archive)97 workbook_root = ET.fromstring(archive.read("xl/workbook.xml"))98 rels_root = ET.fromstring(archive.read("xl/_rels/workbook.xml.rels"))99 relationship_targets = {100 rel.attrib["Id"]: rel.attrib["Target"]101 for rel in rels_root102 }103 104 sheets: dict[str, list[list[str]]] = {}105 for sheet in workbook_root.find("a:sheets", NS):106 sheet_name = sheet.attrib["name"]107 rel_id = sheet.attrib[f"{{{NS['r']}}}id"]108 sheet_path = resolve_sheet_path(relationship_targets[rel_id])109 sheet_root = ET.fromstring(archive.read(sheet_path))110 rows: list[list[str]] = []111 112 for row in sheet_root.findall(".//a:sheetData/a:row", NS):113 values_by_col = {114 column_index(cell.attrib.get("r", "A")): get_cell_text(cell, shared_strings)115 for cell in row.findall("a:c", NS)116 }117 if values_by_col:118 width = max(values_by_col) + 1119 rows.append([values_by_col.get(col, "") for col in range(width)])120 else:121 rows.append([])122 sheets[sheet_name] = rows123 return sheets124 125 126def normalize_rows(raw_rows: list[list[str]]) -> list[dict[str, str]]:127 nonempty_rows = [row for row in raw_rows if any(str(value).strip() for value in row)]128 if not nonempty_rows:129 return []130 131 header = [str(value).strip() for value in nonempty_rows[0]]132 rows: list[dict[str, str]] = []133 for raw_row in nonempty_rows[1:]:134 padded = raw_row + [""] * (len(header) - len(raw_row))135 rows.append({header[index]: str(padded[index]).strip() for index in range(len(header))})136 return rows137 138 139def slugify(value: str) -> str:140 value = value.lower().strip()141 value = re.sub(r"[^a-z0-9]+", "_", value)142 return value.strip("_") or "item"143 144 145def escape_text_node(text: str) -> str:146 """Escape HTML text content without turning quotes into visible entities."""147 return html.escape(html.unescape(text), quote=False)148 149 150def inline_format(text: str) -> str:151 escaped = escape_text_node(text)152 escaped = re.sub(r"`([^`]+)`", r"<code>\1</code>", escaped)153 escaped = re.sub(r"\*\*([^*]+)\*\*", r"<strong>\1</strong>", escaped)154 escaped = re.sub(155 r"(?m)^(Tutor|Student):",156 lambda match: f"<strong>{match.group(1)}:</strong>",157 escaped,158 )159 return escaped160 161 162def rich_text(text: str) -> str:163 text = (text or "").strip()164 if not text:165 return "<p><em>No content provided.</em></p>"166 167 parts = re.split(r"(```(?:[a-zA-Z0-9_+-]+)?\n.*?\n```)", text, flags=re.DOTALL)168 rendered: list[str] = []169 for part in parts:170 if not part:171 continue172 fence_match = re.match(r"```(?:[a-zA-Z0-9_+-]+)?\n(.*?)\n```", part, flags=re.DOTALL)173 if fence_match:174 code = escape_text_node(fence_match.group(1).strip("\n"))175 rendered.append(f'<pre class="potato-code-block"><code>{code}</code></pre>')176 continue177 178 paragraphs = [paragraph.strip() for paragraph in re.split(r"\n\s*\n", part) if paragraph.strip()]179 for paragraph in paragraphs:180 paragraph_html = inline_format(paragraph).replace("\n", "<br>")181 rendered.append(f"<p>{paragraph_html}</p>")182 return "\n".join(rendered)183 184 185def section_html(title: str, body: str, *, extra_class: str = "") -> str:186 class_name = "potato-text-section"187 if extra_class:188 class_name = f"{class_name} {extra_class}"189 return (190 f'<div class="{class_name}">'191 f"<h3>{html.escape(title)}</h3>"192 f"{rich_text(body)}"193 "</div>"194 )195 196 197def make_text2show(row: dict[str, str]) -> str:198 return textwrap.dedent(199 f"""\200 Dialog context:201 {row["dialog_context"]}202 203 Reference (correct) solution:204 {row["correct_solution"]}205 206 Next tutor response:207 {row["tutor_response"]}208 """209 ).strip()210 211 212def make_text2show_html(row: dict[str, str], split: str, item_id: str) -> str:213 split_label = split.upper()214 return "\n".join(215 [216 f'<div class="potato-instance-meta"><span>{split_label}</span><span>{html.escape(item_id)}</span></div>',217 section_html("Dialog context", row["dialog_context"]),218 section_html("Reference (correct) solution", row["correct_solution"], extra_class="reference-solution"),219 section_html("Next tutor response", row["tutor_response"], extra_class="tutor-response"),220 ]221 )222 223 224def validate_labels(row: dict[str, str], row_id: str) -> None:225 for dimension in DIMENSIONS:226 label = row.get(dimension, "").strip()227 if label not in VALID_LABELS:228 raise ValueError(f"{row_id}: invalid label for {dimension!r}: {label!r}")229 230 231def prepare_records(workbook_path: Path) -> dict[str, list[dict[str, str]]]:232 sheets = read_workbook(workbook_path)233 missing_sheets = [sheet for sheet in SHEET_TO_SPLIT if sheet not in sheets]234 if missing_sheets:235 raise ValueError(f"Missing expected sheets: {missing_sheets}")236 237 records_by_split: dict[str, list[dict[str, str]]] = {}238 for sheet_name, split in SHEET_TO_SPLIT.items():239 rows = normalize_rows(sheets[sheet_name])240 if not rows:241 raise ValueError(f"Sheet {sheet_name!r} has no data rows")242 243 required = DISPLAY_COLUMNS + DIMENSIONS244 missing_columns = sorted({column for column in required if column not in rows[0]})245 if missing_columns:246 raise ValueError(f"Sheet {sheet_name!r} missing columns: {missing_columns}")247 248 dialog_numbers: dict[str, int] = {}249 response_counts: defaultdict[int, int] = defaultdict(int)250 split_records: list[dict[str, str]] = []251 252 for row in rows:253 dialog_context = row["dialog_context"]254 if dialog_context not in dialog_numbers:255 dialog_numbers[dialog_context] = len(dialog_numbers) + 1256 dialog_id = dialog_numbers[dialog_context]257 response_counts[dialog_id] += 1258 response_id = response_counts[dialog_id]259 item_id = f"math_{split}_q{dialog_id:02d}_r{response_id:02d}"260 261 validate_labels(row, item_id)262 263 record = {264 "id": item_id,265 "split": split,266 "domain": "math",267 "dialog_id": f"q{dialog_id:02d}",268 "response_id": f"r{response_id:02d}",269 "text2show": make_text2show(row),270 "text2show_html": make_text2show_html(row, split, item_id),271 }272 for column in DISPLAY_COLUMNS + DIMENSIONS:273 record[column] = row[column]274 split_records.append(record)275 records_by_split[split] = split_records276 return records_by_split277 278 279def write_csv(path: Path, rows: list[dict[str, str]]) -> None:280 fieldnames = [281 "id",282 "split",283 "domain",284 "dialog_id",285 "response_id",286 "text2show",287 "text2show_html",288 *DISPLAY_COLUMNS,289 *DIMENSIONS,290 ]291 with path.open("w", newline="", encoding="utf-8") as handle:292 writer = csv.DictWriter(handle, fieldnames=fieldnames)293 writer.writeheader()294 writer.writerows(rows)295 296 297def to_gold_item(record: dict[str, str], *, key_name: str) -> dict[str, object]:298 labels = {dimension: record[dimension] for dimension in DIMENSIONS}299 return {300 "id": record["id"],301 "text": record["text2show"],302 "text2show": record["text2show"],303 "text2show_html": record["text2show_html"],304 key_name: labels,305 "explanation": "Gold labels are provided by the curated math annotator training/test workbook.",306 "metadata": {307 "split": record["split"],308 "domain": record["domain"],309 "dialog_id": record["dialog_id"],310 "response_id": record["response_id"],311 },312 }313 314 315def write_json(path: Path, payload: object) -> None:316 path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")317 318 319def main() -> int:320 parser = argparse.ArgumentParser(description=__doc__)321 parser.add_argument("--input", type=Path, default=DEFAULT_INPUT, help="Source .xlsx workbook")322 parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR, help="Output data directory")323 args = parser.parse_args()324 325 if not args.input.exists():326 print(f"Input workbook not found: {args.input}", file=sys.stderr)327 return 1328 329 records_by_split = prepare_records(args.input)330 output_dir = args.output_dir331 output_dir.mkdir(parents=True, exist_ok=True)332 333 all_records = records_by_split["train"] + records_by_split["test"]334 write_csv(output_dir / "math_annotator_training_set_with_id_text2show.csv", records_by_split["train"])335 write_csv(output_dir / "math_annotator_testing_set_with_id_text2show.csv", records_by_split["test"])336 write_csv(output_dir / "math_annotator_demo_all_with_id_text2show.csv", all_records)337 338 write_json(339 output_dir / "training_questions.json",340 [to_gold_item(record, key_name="correct_answers") for record in records_by_split["train"]],341 )342 write_json(343 output_dir / "gold_standards.json",344 [to_gold_item(record, key_name="gold_label") for record in records_by_split["test"]],345 )346 347 summary = {348 "input": str(args.input),349 "outputs": {350 "train_csv": "math_annotator_training_set_with_id_text2show.csv",351 "test_csv": "math_annotator_testing_set_with_id_text2show.csv",352 "combined_csv": "math_annotator_demo_all_with_id_text2show.csv",353 "training_questions": "training_questions.json",354 "gold_standards": "gold_standards.json",355 },356 "counts": {split: len(records) for split, records in records_by_split.items()},357 "dimensions": DIMENSIONS,358 "labels": sorted(VALID_LABELS),359 }360 write_json(output_dir / "data_summary.json", summary)361 362 print(json.dumps(summary, ensure_ascii=False, indent=2))363 return 0364 365 366if __name__ == "__main__":367 raise SystemExit(main())368