sahil-12kumar/IL_CMS_Tools
1
1"""2solution_assist.py — bulk solution authoring + answer-key marking (paste-into-any-AI).3 4The SME pastes CMS question IDs. The app fetches each question (stem + options),5builds ONE copy-paste prompt that asks an AI to SOLVE each question and write a6step-by-step solution, and — after the SME pastes the JSON reply back — writes the7solution and/or the correct answer key back to the CMS.8 9The stored answer is deliberately withheld from the prompt (same solve-blind rule10as verify_assist): the AI must reach its own answer, so a disagreement with the11stored key is real signal and not anchoring. Each row therefore carries a verdict:12 13 NO KEY — nothing stored yet; the AI's answer is the key to set14 MATCH — the AI agrees with the stored key15 MISMATCH — the AI disagrees; the key is NOT overwritten unless the SME ticks it16 UNSURE — not comparable (AI could not solve, or the stored key is unreadable)17 18Writing the key is TYPE-AWARE: the CMS stores `correct_answer` as a list of191-based option-index strings for MCQs but as the literal value for numeric types,20so the answer is re-encoded per question type from the LIVE record at push time21(never from the possibly-stale fetch used to build the prompt).22 23Reuses existing primitives only:24 * ai_assist.fetch_contents — the CMS fetch (stem + options, math preserved)25 * verify_assist — the answer canonicalisation / comparison helpers26 * ai_tagger/push_tags.py — qb_headers_from_state(), fetch(), QB base URL27"""28from __future__ import annotations29 30import json31import os32import re33import sys34import time35from concurrent.futures import ThreadPoolExecutor, as_completed36 37import requests38 39# push_tags/cms_text live in ai_tagger/ — put it on the path here rather than40# relying on whichever sibling module happened to be imported first.41_AI = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ai_tagger")42if _AI not in sys.path:43 sys.path.insert(0, _AI)44 45# Shared answer canonicalisation. These live in verify_assist because Verify46# defined them first; both tools must compare answers the SAME way or a question47# could read as MATCH in one tool and MISMATCH in the other.48from verify_assist import _canon, _compare, _norm, answer_display, clean_answer49 50# Results columns (Excel + UI table).51OUT_COLS = ["question_id", "question_type", "verdict", "stored_answer", "ai_answer",52 "key_action", "solution_action", "message"]53 54# ── Question-type kinds ───────────────────────────────────────────────────────55# The CMS `question_type` string is already canonical on a live record ("single56# correct mcq", "numerical value question", …), but stay tolerant of aliases.57K_SINGLE, K_MULTI, K_NUMERIC, K_INTEGER, K_TF = "single", "multiple", "numeric", "integer", "tf"58 59 60def type_kind(qtype) -> str:61 """Coarse kind of a CMS question_type string — drives how the key is encoded."""62 s = _norm(qtype)63 if "multiple" in s or s in ("msq", "multi"):64 return K_MULTI65 if "integer" in s:66 return K_INTEGER67 if "numer" in s or s == "nat":68 return K_NUMERIC69 if "true" in s or "false" in s:70 return K_TF71 return K_SINGLE72 73 74_MATRIX = "matrix-match question"75_PASSAGE = "passage question (sub-questions carry their own keys)"76 77 78def unsupported_reason(q):79 """Why this LIVE question's key cannot be written ("" when it can be).80 81 Matrix-match and passage questions are refused rather than guessed at: their82 keys are nested / per-child structures that no create path in this repo83 produces, so a wrong write would silently corrupt a live question."""84 q = q or {}85 opts = q.get("answer_options") or []86 if q.get("child_questions"):87 return _PASSAGE88 if len(opts) >= 2 and all(isinstance(o, (list, tuple)) for o in opts):89 return _MATRIX90 return ""91 92 93def record_block_reason(rec):94 """Same check against a fetched ai_assist record (which flattens matrix columns95 into the stem, so the raw shape is gone) — used at review time to grey out the96 key checkbox before the SME ever tries to push."""97 rec = rec or {}98 if rec.get("is_passage"):99 return _PASSAGE100 if rec.get("is_matrix"):101 return _MATRIX102 return ""103 104 105# ── 1. Build the solve + write-solution prompt ────────────────────────────────106def build_prompt(contents, style="steps") -> str:107 """`contents` = ai_assist.fetch_contents() records. The stored answer is NOT108 included — the AI solves blind so its answer is independent evidence."""109 q_blocks = []110 for c in contents:111 if c.get("error"):112 continue113 opts = c.get("options") or []114 opts_str = "\n".join(f" ({chr(65 + i)}) {o}" for i, o in enumerate(opts))115 block = [f"QID: {c['question_id']}"]116 if c.get("question_type"):117 block.append(f"Type: {c['question_type']}")118 block.append(f"Question: {c.get('question_text', '')}")119 if opts_str:120 block.append("Options:\n" + opts_str)121 q_blocks.append("\n".join(block))122 123 # "2-4 steps, one step per line" produced 3-4 long prose sentences with the124 # maths inline — technically compliant, but it read as a running paragraph125 # rather than a worked solution. The shape is spelled out explicitly below.126 if style == "steps":127 shape = """a NUMBERED, step-by-step derivation. Structure it exactly like this:128 129Step 1: <what you are doing, one short sentence>130<the formula or working for that step, on its own line>131Step 2: <next thing you are doing>132<its working, on its own line>133...134Answer: <your final answer>135 136- Use 3-6 steps. Each step is ONE operation — state a principle, substitute \137values, or simplify. Never chain two operations into one step.138- Keep the prose part of a step under about 20 words. Push the algebra onto its \139own line underneath it.140- When you are eliminating options, put each option on its own line under a \141single step, e.g. "KCl: $i \\approx 2 \\Rightarrow 1$ M".142- The last line MUST start with "Answer:" and state the result."""143 else:144 shape = ("a single short paragraph that ENDS at your stated answer")145 146 return f"""You are an expert exam subject-matter expert writing model solutions. \147Solve each question below and write the solution a student will read.148 149RULES150- Work each question out yourself and state the answer you arrive at in "answer". \151For a multiple-choice question give the option LETTER(s) — "B", or "A, C" when more than \152one is correct. For a numeric/integer question give the number only (no units, no words).153- "solution" is {shape}154- Put every formula and symbol in LaTeX between single dollar signs, e.g. $v = u + at$, \155$\\theta = 30^\\circ$. Use $...$ — NOT \\( \\) and NOT \\[ \\].156- Separate lines with a real newline character inside the JSON string (\\n). Do NOT use \157markdown headings, bullets, bold, or tables — "Step 1:" is plain text, not "**Step 1:**".158- Do NOT restate the full question, and do NOT write "Option (B)" as the whole solution \159— show the working.160- A "[figure]" marker means the question has an image you CANNOT see. If — and only if — \161the question cannot be solved without that image, set "answer" and "solution" to "" and \162"solvable" to false. If the text alone is enough, solve it normally.163- Never guess. If you are unsure, say so with "confidence": "low".164 165QUESTIONS:166{chr(10).join(chr(10).join(['', b]) for b in q_blocks)}167 168OUTPUT169Return ONLY a JSON array — one object per QID, no markdown fences, no commentary. Keys:170question_id, answer, solution, solvable (true/false), confidence ("high"/"medium"/"low").171"""172 173 174# ── 2. Parse the AI reply into reviewable rows ────────────────────────────────175def parse_reply(json_text, records):176 """Return (rows, warnings). `records` = {qid: fetched-content-dict}.177 178 Each row carries the proposed solution + answer, the stored answer, a verdict,179 and the DEFAULT push intent: a solution is pushed whenever the AI wrote one,180 but the key is pre-ticked only when there is nothing to lose (NO KEY) or the181 AI agrees (MATCH). A MISMATCH needs a deliberate tick from the SME."""182 text = (json_text or "").strip()183 m = re.search(r"\[.*\]", text, re.S) # tolerate ```json fences / prose184 if m:185 text = m.group(0)186 try:187 arr = json.loads(text)188 except Exception as e: # noqa: BLE001189 raise ValueError(f"Could not parse the AI's JSON: {e}")190 if isinstance(arr, dict):191 arr = [arr]192 if not isinstance(arr, list) or not arr:193 raise ValueError("Expected a non-empty JSON array of objects.")194 195 rows, warnings, seen = [], [], set()196 for i, obj in enumerate(arr):197 if not isinstance(obj, dict):198 warnings.append(f"item {i + 1}: not a JSON object — skipped")199 continue200 o = {_norm(k).replace(" ", "_"): v for k, v in obj.items()}201 qid = str(o.get("question_id") or o.get("qid") or o.get("id") or "").strip()202 rec = records.get(qid) or {}203 if not rec:204 warnings.append(f"{qid or f'item {i + 1}'}: not one of the fetched QIDs — skipped")205 continue206 seen.add(qid)207 208 ai_answer = str(o.get("answer") or "").strip()209 solution = str(o.get("solution") or "").strip()210 solvable = o.get("solvable")211 solvable = True if solvable is None else bool(solvable)212 stored = str(rec.get("correct") or "").strip()213 214 if not solvable or not ai_answer:215 verdict = "UNSURE"216 elif not stored:217 verdict = "NO KEY"218 else:219 verdict = _compare(stored, ai_answer, rec.get("options"))220 221 blocked = record_block_reason(rec) # matrix / passage — key cannot be written222 can_key = bool(ai_answer) and verdict in ("NO KEY", "MATCH", "MISMATCH") and not blocked223 rows.append({224 "question_id": qid,225 "question_type": rec.get("question_type") or "",226 "verdict": verdict,227 "stored_answer": answer_display(stored, rec.get("options")) or "(none)",228 "ai_answer": ai_answer or "(none)",229 "solution": solution,230 "confidence": str(o.get("confidence") or "").strip().lower(),231 "blocked": blocked or "",232 # default intents the UI renders as checkboxes233 "push_solution": bool(solution),234 "push_key": can_key and verdict in ("NO KEY", "MATCH"),235 "can_key": can_key,236 "key_action": "", "solution_action": "", "message": "",237 })238 239 for qid in records:240 if qid not in seen:241 warnings.append(f"{qid}: no entry in the AI reply — not solved")242 if not rows:243 raise ValueError("No usable rows in the pasted JSON.")244 245 order = {"NO KEY": 0, "MISMATCH": 1, "UNSURE": 2, "MATCH": 3}246 rows.sort(key=lambda r: order.get(r["verdict"], 4))247 return rows, warnings248 249 250# ── 3. Encode an answer for the CMS, per question type ────────────────────────251_NUM_TOKEN_RE = re.compile(r"-?\d[\d,]*\.?\d*")252 253 254def _clean_number(ans):255 """The answer's numeric token as a clean string ('19.60 m' -> '19.60'), or None.256 The digits the AI wrote are kept verbatim — going via float would turn 0.3 into257 0.30000000000000004 and change what the CMS stores."""258 m = _NUM_TOKEN_RE.search(str(ans or "").replace(",", ""))259 return m.group(0) if m else None260 261 262def encode_answer(kind, ans, options):263 """CMS `correct_answer` list for this question type. Raises ValueError with an264 SME-readable reason when the answer cannot be encoded safely.265 266 MCQ keys are 1-based option INDEX strings (["2"]); numeric/integer keys are the267 literal value (["19.6"]); true/false uses the option index when the question268 carries True/False options and the plain word otherwise."""269 ans = str(ans or "").strip()270 if not ans:271 raise ValueError("no answer to write")272 n = len(options or [])273 274 if kind in (K_NUMERIC, K_INTEGER):275 num = _clean_number(ans)276 if num is None:277 raise ValueError(f"'{ans}' is not a number")278 if kind == K_INTEGER and not re.fullmatch(r"-?\d+", num):279 raise ValueError(f"integer-type question but the answer '{num}' is not an integer")280 return [num]281 282 if kind == K_TF and n < 2:283 t = _norm(ans)284 if t.startswith("t") or t == "1":285 return ["true"]286 if t.startswith("f") or t == "0":287 return ["false"]288 raise ValueError(f"'{ans}' is not true/false")289 290 # Choice question: resolve the answer to option letter(s), then to 1-based indexes.291 clean = clean_answer(ans)292 toks = [t for t in re.split(r"[,\s/;]+", clean) if t.strip()]293 if toks and all(re.fullmatch(r"[1-9]", t) for t in toks):294 # Bare digits mean the option NUMBER (the CMS's own 1-based convention,295 # cf. app.ANSWER_MAP) — checked BEFORE option-text matching, which would296 # otherwise match "2" to whichever option merely contains a 2.297 val = {chr(64 + int(t)) for t in toks}298 else:299 kindv, val = _canon(clean, options)300 if kindv != "letters":301 raise ValueError(f"'{ans}' does not name one of the options")302 letters = sorted(val)303 if kind == K_SINGLE and len(letters) > 1:304 raise ValueError(f"single-correct question but the AI gave {len(letters)} answers")305 if kind == K_MULTI and len(letters) < 1:306 raise ValueError("multiple-correct question with no answer")307 out = []308 for L in letters:309 i = ord(L) - 64 # A -> 1310 if n and not (1 <= i <= n):311 raise ValueError(f"option ({L}) does not exist — the question has {n} option(s)")312 out.append(str(i))313 return out314 315 316def solution_ranges_for(ans, existing=None):317 """CMS `solutionRanges` for a numeric answer: an exact-match range whose start318 and end are both the answer. Shape confirmed in ai_tagger/TAGGING_WORKFLOW.md:319 [{"name": <precision enum>, "start": ans, "end": ans}].320 321 The authoritative helper (il_cms.solution_ranges_for, in the external322 IL_GPT_Backend) is used when that backend is importable; otherwise we rebuild323 the entry here, KEEPING the precision `name` the question already carries so a324 decimal question doesn't get silently re-bucketed."""325 try:326 from il_cms import solution_ranges_for as _authoritative327 return _authoritative(ans)328 except Exception: # noqa: BLE001 — backend not installed on this host329 pass330 name = "Default Range"331 if existing and isinstance(existing, list) and isinstance(existing[0], dict):332 name = existing[0].get("name") or name333 elif "." in str(ans):334 # No prior range to copy the precision enum from: keep the default bucket335 # rather than inventing an enum value the CMS may not know.336 name = "Default Range"337 return [{"name": name, "start": str(ans), "end": str(ans)}]338 339 340# ── 4. Push solution + key back to the CMS ────────────────────────────────────341def _readable_options(q):342 """Live `answer_options` as plain text, so an answer given as option TEXT343 ("2 m/s") can still be matched to its letter — matching against the raw CMS344 HTML/MathML would never hit."""345 import cms_text346 return [cms_text.html_to_readable_inline(o.get("data", o) if isinstance(o, dict) else o)347 for o in (q.get("answer_options") or [])]348 349 350 351def push_one(h, qid, solution_html=None, answer=None, allow_approved=False):352 """Write a solution and/or a correct answer onto ONE live question.353 354 Returns (ok, message, detail) where detail = {key_action, solution_action}.355 The record is fetched fresh and PUT back whole (the CMS has no field-level356 PATCH), then re-fetched to confirm the write actually landed — the same357 fetch → mutate → PUT → verify shape as ai_tagger/push_solutions.py."""358 import push_tags as pt359 360 detail = {"key_action": "", "solution_action": ""}361 data = pt.fetch(h, qid)362 q = data.get("question") or {}363 status = _norm(data.get("status"))364 if status == "approved" and not allow_approved:365 return False, "question is approved — not modified", detail366 367 qtype = data.get("question_type") or ""368 blocked = unsupported_reason(q)369 370 now = int(time.time() * 1000)371 if solution_html:372 sols = q.get("solution") or []373 if sols and isinstance(sols[0], dict):374 sols[0]["data"] = solution_html375 sols[0]["updated_at"] = now376 sols[0]["is_visible"] = True377 else:378 sols = [{"data": solution_html, "is_visible": True,379 "created_at": now, "updated_at": now}]380 q["solution"] = sols381 detail["solution_action"] = "written"382 383 enc = None384 if answer:385 if blocked:386 detail["key_action"] = f"skipped ({blocked})"387 else:388 try:389 enc = encode_answer(type_kind(qtype), answer, _readable_options(q))390 except ValueError as e:391 detail["key_action"] = f"skipped ({e})"392 else:393 q["correct_answer"] = enc394 detail["key_action"] = "set to " + ", ".join(enc)395 if type_kind(qtype) in (K_NUMERIC, K_INTEGER):396 q["solutionRanges"] = solution_ranges_for(397 enc[0], q.get("solutionRanges"))398 399 if not detail["solution_action"] and not enc:400 return False, detail["key_action"] or "nothing to write", detail401 402 data["question"] = q403 pt.ensure_question_pattern(data) # keep approvable; see push_tags for why404 r = requests.put(f"{pt.QB}/questions/{qid}", json=data, headers=h, timeout=120)405 if not r.ok:406 return False, f"PUT [{r.status_code}]: {r.text[:180]}", detail407 408 # Verify: a 200 is not proof the values stored — re-read and compare.409 time.sleep(0.3)410 try:411 check = (pt.fetch(h, qid) or {}).get("question", {}) or {}412 except Exception as e: # noqa: BLE001 — the write may still have landed413 return True, f"written (could not verify: {e})", detail414 if solution_html:415 got = ((check.get("solution") or [{}])[0] or {}).get("data", "") or ""416 if got.strip() != solution_html.strip():417 return False, f"PUT ok but the solution did not store ({len(got)} chars back)", detail418 if enc:419 got = [str(x) for x in (check.get("correct_answer") or [])]420 if got != enc:421 return False, f"PUT ok but the key did not store (got {got or 'nothing'})", detail422 return True, "verified", detail423 424 425def push_rows(state_file, rows, allow_approved=False, progress=None, workers=5):426 """Push every row that has an intent ticked. Mutates each row's key_action /427 solution_action / message in place and returns (ok_count, fail_count).428 progress(done, total) is called after each question.429 430 Rows are pushed in PARALLEL (5 workers): push_one is network-bound (fetch →431 PUT → settle sleep → verify), so serializing the batch made the sleep and the432 round-trips add up one row at a time. Each worker pushes its OWN row and433 returns (index, good, msg, detail); the collector writes results back into434 that row's dict and drives progress — no shared mutation between threads."""435 import push_tags as pt436 437 h = pt.qb_headers_from_state(state_file)438 total = len(rows)439 440 def _do(index):441 r = rows[index]442 sol = r.get("solution_html") if r.get("push_solution") else None443 ans = r.get("ai_answer") if r.get("push_key") else None444 try:445 good, msg, detail = push_one(h, r["question_id"], solution_html=sol,446 answer=ans, allow_approved=allow_approved)447 except Exception as e: # noqa: BLE001 — one bad QID must not stop the batch448 return index, False, str(e)[:180], {}449 return index, good, msg, detail450 451 ok = fail = done = 0452 with ThreadPoolExecutor(max_workers=max(1, workers)) as ex:453 futs = [ex.submit(_do, i) for i in range(total)]454 for fut in as_completed(futs):455 index, good, msg, detail = fut.result()456 r = rows[index]457 r["message"] = msg458 r["key_action"] = detail.get("key_action", "")459 r["solution_action"] = detail.get("solution_action", "")460 r["pushed"] = good461 ok += good462 fail += not good463 done += 1464 if progress:465 try:466 progress(done, total)467 except Exception:468 pass469 return ok, fail470 471 472# ── 5. Results Excel ──────────────────────────────────────────────────────────473def write_xlsx(rows, out_path):474 from openpyxl import Workbook475 wb = Workbook()476 ws = wb.active477 ws.title = "Solutions"478 ws.append(OUT_COLS)479 for r in rows:480 ws.append([r.get(c, "") for c in OUT_COLS])481 wb.save(out_path)482