"""score / rescore jobs (M5). The model returns findings; marks are computed in code. score payload: {script_id, batch_id} rescore payload: {script_id, batch_id, unit_ids: [..] | null (= all), keep_overrides: true}""" from __future__ import annotations import json import pathlib import regex from .. import ai, config, jobs from ..db import Jsonb from ..ids import new_id from ..providers.base import TextPart from ..records import audit, raise_flag, record_usage from ..schemas.scoring import ScriptFeedback, ScriptFinding, UnitFinding, UnitsOnlyFinding from . import scoring_math as sm TEMPLATE = (pathlib.Path(__file__).parent.parent / "prompts" / "score_system.md").read_text(encoding="utf-8") PROMPT_VERSION = "score-v1" LANG_NAMES = {"bn": "Bengali (বাংলা)", "en": "English"} NOT_ATTEMPTED = {"bn": ("উত্তর দেওয়া হয়নি।", "এই অংশের কোনো উত্তর পাওয়া যায়নি, তাই নম্বর ০।"), "en": ("Not attempted.", "No answer was found for this part, so the mark is 0.")} NOT_COUNTED = {"bn": "পছন্দের নিয়মে গণ্য হয়নি", "en": "Not counted under the choice rule"} # ---------------------------------------------------------------- context loading def load_context(conn, script_id: str, tenant_id: str) -> dict: s = conn.execute("SELECT * FROM scripts WHERE id=%s AND tenant_id=%s FOR UPDATE", (script_id, tenant_id)).fetchone() if not s: raise jobs.FatalError("script not found") b = conn.execute("SELECT * FROM batches WHERE id=%s AND tenant_id=%s", (s["batch_id"], tenant_id)).fetchone() qs = conn.execute("SELECT * FROM question_sets WHERE id=%s", (b["question_set_id"],)).fetchone() meta = conn.execute( """SELECT es.name_en AS es, c.name_en AS cls, sub.name_en AS subject, cat.default_weights_json AS cat_w FROM subjects sub JOIN classes c ON c.id=sub.class_id JOIN education_systems es ON es.id=c.education_system_id JOIN subject_categories cat ON cat.id=sub.subject_category_id WHERE sub.id=%s""", (qs["subject_id"],)).fetchone() sections = {r["id"]: r for r in conn.execute("SELECT * FROM qs_sections WHERE question_set_id=%s ORDER BY sort", (qs["id"],)).fetchall()} questions = conn.execute("SELECT * FROM questions WHERE question_set_id=%s ORDER BY sort", (qs["id"],)).fetchall() if b["section_ids_json"]: questions = [q for q in questions if q["section_id"] in b["section_ids_json"]] units = conn.execute("""SELECT u.* FROM scoring_units u JOIN questions q ON q.id=u.question_id WHERE q.question_set_id=%s ORDER BY q.sort, u.sort""", (qs["id"],)).fetchall() qids = {q["id"] for q in questions} units = [u for u in units if u["question_id"] in qids] tenant = conn.execute("SELECT settings_json FROM tenants WHERE id=%s", (tenant_id,)).fetchone() lang = (tenant["settings_json"] or {}).get("feedback_lang", "bn") return {"script": s, "batch": b, "qs": qs, "meta": meta, "sections": sections, "questions": questions, "units": units, "lang": lang} def rubric_text(ctx: dict) -> str: """Stable, cache-friendly rendering of the question paper (identical for every script of a batch).""" qmap = {q["id"]: q for q in ctx["questions"]} out = [] for sec in ctx["sections"].values(): qs_in = [q for q in ctx["questions"] if q["section_id"] == sec["id"]] if not qs_in: continue rule = f" (answer any {sec['answer_count']} of {len(qs_in)})" if sec["answer_count"] else "" out.append(f"## Section {sec['label']}{rule}") for q in qs_in: out.append(f"### Question {q['label']} [type: {q['type']}, max {float(q['max_mark']):g}]\n{q['text']}") if q["stimulus_text"]: out.append(f"Stimulus: {q['stimulus_text']}") for u in [u for u in ctx["units"] if u["question_id"] == q["id"]]: head = f"#### Unit {u['id']} — {q['label']}{'(' + u['label'] + ')' if u['label'] else ''} [max {float(u['max_mark']):g}]" parts = [head, f"Model answer: {u['model_answer_text']}"] if u["model_answer_latex"]: parts.append(f"Model answer (LaTeX): {u['model_answer_latex']}") if u["key_points_json"]: kp = "; ".join(f"[{k['id']}{'*' if k.get('mandatory') else ''}] {k['text']} (weight {k.get('weight', 1)})" for k in u["key_points_json"]) parts.append(f"Key points (* = mandatory): {kp}") if u["instructions"]: parts.append(f"Marking instructions: {u['instructions']}") if u["alternatives_json"]: parts.append("Acceptable alternatives: " + " | ".join(map(str, u["alternatives_json"]))) if u["common_mistakes_json"]: parts.append("Common mistakes: " + " | ".join(map(str, u["common_mistakes_json"]))) parts.append(f"Expected answer language(s): {', '.join(q['expected_langs_json'])}") out.append("\n".join(parts)) del qmap return "\n\n".join(out) def system_prompt(ctx: dict, with_feedback: bool) -> str: syl = ctx["qs"]["syllabus_json"] or {} fb = ("After all units, also return strengths (3–5), weaknesses (3–5), suggestions (3–5, concrete and tied to " "syllabus chapters where relevant) and a summary of at most 300 characters for the whole script." if with_feedback else "Do not return script-level feedback.") return TEMPLATE.format(education_system=ctx["meta"]["es"], class_name=ctx["meta"]["cls"], subject=ctx["meta"]["subject"], syllabus=syl.get("label") or json.dumps(syl, ensure_ascii=False), feedback_lang_name=LANG_NAMES.get(ctx["lang"], ctx["lang"]), feedback_instruction=fb, rubric=rubric_text(ctx)) # ---------------------------------------------------------------- helpers def trim_graphemes(s: str, n: int) -> str: g = regex.findall(r"\X", s or "") return s if len(g) <= n else "".join(g[: n - 1]) + "…" def answers_text(segments: list[dict], units_by_id: dict, questions_by_id: dict) -> str: out = [] for seg in segments: u = units_by_id[seg["scoring_unit_id"]] q = questions_by_id[u["question_id"]] label = f"{q['label']}{'(' + u['label'] + ')' if u['label'] else ''}" att = f" — attempt {seg['attempt_no']}" if seg["attempt_no"] > 1 else "" out.append(f"=== Unit {u['id']} ({label}){att}\n{seg['content_text']}") return "\n\n".join(out) def estimate_tokens(*texts: str) -> int: return sum(len(t) for t in texts) // 3 # conservative for Bengali def ref_map(conn, segments: list[dict], tenant_id: str) -> dict[str, str]: """'p4-l19' -> element id, for every element in the given segments.""" ids = [e for s in segments for e in (s["element_ids_json"] or [])] if not ids: return {} rows = conn.execute("""SELECT e.id, e.sort, p.sort AS ps FROM elements e JOIN pages p ON p.id=e.page_id WHERE e.id = ANY(%s) AND e.tenant_id=%s""", (ids, tenant_id)).fetchall() return {f"p{r['ps']}-l{r['sort']}": r["id"] for r in rows} # ---------------------------------------------------------------- AI calls def call_model(conn, ctx, segments, units_by_id, questions_by_id, tenant_id, with_feedback: bool): """Returns (unit findings by segment key, feedback | None, [Result]).""" granularity = config.get(conn, "ai.score.granularity", tenant_id) limit = int(config.get(conn, "ai.score.max_input_tokens", tenant_id)) system = system_prompt(ctx, with_feedback=False) full_answers = answers_text(segments, units_by_id, questions_by_id) groups: list[list[dict]] if granularity == "question" or estimate_tokens(system, full_answers) > limit: by_q: dict[str, list[dict]] = {} for s in segments: by_q.setdefault(units_by_id[s["scoring_unit_id"]]["question_id"], []).append(s) groups = list(by_q.values()) else: groups = [segments] results, findings = [], {} feedback = None single_call = len(groups) == 1 and with_feedback for g in groups: schema = ScriptFinding if single_call else UnitsOnlyFinding sysp = system_prompt(ctx, with_feedback=single_call) user = TextPart("STUDENT ANSWERS\n" + answers_text(g, units_by_id, questions_by_id) + f"\n\nHandwriting legibility average (0-10): {ctx['script']['legibility_avg']}") res = ai.call(conn, stage="score", tenant_id=tenant_id, system=sysp, parts=[user], schema=schema, max_tokens=16000) results.append(res) expected = {s["scoring_unit_id"] for s in g} got = {f.unit_id: f for f in res.parsed.units} missing = expected - set(got) if missing: raise jobs.RetryableError(f"model omitted units: {sorted(missing)[:5]}") for s in g: findings[s["id"]] = got[s["scoring_unit_id"]] # duplicate attempts share one finding per unit if single_call: p = res.parsed feedback = ScriptFeedback(strengths=p.strengths, weaknesses=p.weaknesses, suggestions=p.suggestions, summary=p.summary) if with_feedback and feedback is None: briefs = "\n".join(f"- {units_by_id[s['scoring_unit_id']]['id']}: {findings[s['id']].brief}" for s in segments) res = ai.call(conn, stage="score", tenant_id=tenant_id, system=system_prompt(ctx, with_feedback=True), parts=[TextPart("Per-unit findings for this script:\n" + briefs + "\nReturn only the script-level feedback.")], schema=ScriptFeedback, max_tokens=4000) results.append(res) feedback = res.parsed return findings, feedback, results # ---------------------------------------------------------------- persistence def write_result(conn, *, tenant_id, segment_id, source, us: sm.UnitScore | None, final, raw, ratings, weights, counts, brief, detailed, errors, positives, key_points, flags, model_info, qs_version) -> None: v = conn.execute("SELECT coalesce(max(version),0)+1 AS v FROM score_results WHERE segment_id=%s", (segment_id,)).fetchone()["v"] conn.execute("UPDATE score_results SET is_current=false WHERE segment_id=%s AND is_current", (segment_id,)) conn.execute( """INSERT INTO score_results (id, tenant_id, segment_id, version, source, ratings_json, weights_json, error_counts_json, raw_score, final_score, brief, detailed, error_items_json, positive_items_json, key_points_json, flags_json, model_info_json, qs_version, is_current) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,true)""", (new_id(), tenant_id, segment_id, v, source, Jsonb(ratings), Jsonb(weights), Jsonb(counts), raw, final, brief, detailed, Jsonb(errors), Jsonb(positives), Jsonb(key_points), Jsonb(flags), Jsonb(model_info) if model_info else None, qs_version)) def finding_to_result(f: UnitFinding, refs: dict[str, str]) -> tuple[list, list, list]: def el(r): return refs.get(r) if r else None errors = [{**e.model_dump(), "remark": trim_graphemes(e.remark, 60), "element_id": el(e.line_ref)} for e in f.error_items] positives = [{**p.model_dump(), "element_id": el(p.line_ref)} for p in f.positive_items] kps = [{**k.model_dump(), "element_id": el(k.line_ref)} for k in f.key_points] return errors, positives, kps def recompute_totals(conn, ctx, tenant_id: str) -> None: """Duplicates, choice rules, question caps, script total (SG-04, SG-06, SC-12). Uses current results, so it is also correct after teacher overrides.""" s, qs = ctx["script"], ctx["qs"] rows = conn.execute( """SELECT sg.id, sg.scoring_unit_id, sg.attempt_no, sg.flags_json, sg.not_attempted, r.final_score, (SELECT min(p.sort * 100000 + e.sort) FROM elements e JOIN pages p ON p.id=e.page_id WHERE e.id IN (SELECT jsonb_array_elements_text(sg.element_ids_json))) AS first_pos FROM segments sg LEFT JOIN score_results r ON r.segment_id=sg.id AND r.is_current WHERE sg.script_id=%s AND sg.tenant_id=%s AND sg.scoring_unit_id IS NOT NULL""", (s["id"], tenant_id)).fetchall() # 1. duplicates: one counted attempt per unit by_unit: dict[str, list[dict]] = {} for r in rows: if "all_struck" not in (r["flags_json"] or []): by_unit.setdefault(r["scoring_unit_id"], []).append(r) keep: dict[str, dict] = {} for uid, atts in by_unit.items(): if qs["duplicate_mode"] == "first": keep[uid] = min(atts, key=lambda a: a["attempt_no"]) else: keep[uid] = max(atts, key=lambda a: (float(a["final_score"] or 0), -a["attempt_no"])) # 2. question marks with cap units = {u["id"]: u for u in ctx["units"]} qmark: dict[str, float] = {} qorder: dict[str, int] = {} attempted_q: set[str] = set() for uid, a in keep.items(): q_id = units[uid]["question_id"] qmark[q_id] = qmark.get(q_id, 0.0) + float(a["final_score"] or 0) if not a["not_attempted"]: attempted_q.add(q_id) pos = a["first_pos"] if a["first_pos"] is not None else 10**9 qorder[q_id] = min(qorder.get(q_id, 10**9), int(pos)) qmax = {q["id"]: float(q["max_mark"]) for q in ctx["questions"]} for q_id in qmark: qmark[q_id] = min(qmark[q_id], qmax[q_id]) # 3. choice rules per section counted_q: set[str] = set() for sec in ctx["sections"].values(): in_sec = [q["id"] for q in ctx["questions"] if q["section_id"] == sec["id"]] answered = [(q, qmark.get(q, 0.0), qorder.get(q, 10**9)) for q in in_sec if q in attempted_q] chosen = sm.apply_choice_rule(answered, sec["answer_count"], qs["choice_mode"]) counted_q |= chosen if sec["answer_count"] and len(answered) > sec["answer_count"]: raise_flag(conn, tenant_id=tenant_id, batch_id=s["batch_id"], type_="choice_rule_applied", script_id=s["id"], payload={"section": sec["label"], "answered": len(answered), "counted": sec["answer_count"]}) # 4. write counted flags and total for r in rows: uid = r["scoring_unit_id"] q_id = units[uid]["question_id"] counted = keep.get(uid) is not None and keep[uid]["id"] == r["id"] and \ (q_id in counted_q or r["not_attempted"]) conn.execute("UPDATE segments SET counted=%s WHERE id=%s AND tenant_id=%s", (counted, r["id"], tenant_id)) total = min(sum(qmark[q] for q in counted_q), float(qs["total_target_marks"])) conn.execute("UPDATE scripts SET total_score=%s, total_target=%s, updated_at=now() WHERE id=%s AND tenant_id=%s", (total, qs["total_target_marks"], s["id"], tenant_id)) # ---------------------------------------------------------------- job handlers def _score(conn, job, unit_ids: set[str] | None, keep_overrides: bool, with_feedback: bool) -> dict: p = job["payload_json"] tenant_id = job["tenant_id"] ctx = load_context(conn, p["script_id"], tenant_id) s, qs = ctx["script"], ctx["qs"] units_by_id = {u["id"]: u for u in ctx["units"]} questions_by_id = {q["id"]: q for q in ctx["questions"]} segs = conn.execute("""SELECT * FROM segments WHERE script_id=%s AND tenant_id=%s AND scoring_unit_id IS NOT NULL ORDER BY scoring_unit_id, attempt_no""", (s["id"], tenant_id)).fetchall() if unit_ids is not None: segs = [g for g in segs if g["scoring_unit_id"] in unit_ids] if keep_overrides: teacher = {r["segment_id"] for r in conn.execute( """SELECT segment_id FROM score_results r JOIN segments g ON g.id=r.segment_id WHERE g.script_id=%s AND r.is_current AND r.source='teacher'""", (s["id"],)).fetchall()} segs = [g for g in segs if g["id"] not in teacher] to_ai = [g for g in segs if not g["not_attempted"] and "all_struck" not in (g["flags_json"] or [])] lang = ctx["lang"] findings, feedback, results = ({}, None, []) if to_ai: findings, feedback, results = call_model(conn, ctx, to_ai, units_by_id, questions_by_id, tenant_id, with_feedback) refs = ref_map(conn, to_ai, tenant_id) cat_w = ctx["meta"]["cat_w"] penalties = {k: float(config.get(conn, f"scoring.{k}", tenant_id)) for k in ("spelling_penalty", "grammar_penalty", "max_spelling_penalty")} model_info = ({**results[0].model_info(), "prompt": PROMPT_VERSION} if results else None) minimal = 0 for g in segs: u = units_by_id[g["scoring_unit_id"]] max_mark, step = float(u["max_mark"]), float(qs["step"]) weights = sm.effective_weights(cat_w, qs["weights_json"], u["weights_json"]) if g["not_attempted"] or g["id"] not in findings: brief, detailed = NOT_ATTEMPTED.get(lang, NOT_ATTEMPTED["en"]) write_result(conn, tenant_id=tenant_id, segment_id=g["id"], source="system", us=None, final=0, raw=0, ratings={}, weights=weights, counts={}, brief=brief, detailed=detailed, errors=[], positives=[], key_points=[], flags=["not_attempted"], model_info=None, qs_version=qs["version"]) continue f = findings[g["id"]] errors, positives, kps = finding_to_result(f, refs) us = sm.compute_unit(max_mark=max_mark, weights=weights, correctness=f.correctness, completeness=f.completeness, error_items=errors, legibility_avg=float(s["legibility_avg"]) if s["legibility_avg"] is not None else None, step=step, rounding=qs["rounding"], content_gate=qs["content_gate"], **penalties) flags = [] if not u["key_points_json"] and not u["instructions"]: flags.append("rubric_minimal"); minimal += 1 if us.gated: flags.append("content_gate") if sm.uncertainty_matters(g["content_text"], max_mark, step): flags.append("uncertain_affects_score") raise_flag(conn, tenant_id=tenant_id, batch_id=s["batch_id"], type_="uncertain_affects_score", script_id=s["id"], segment_id=g["id"], payload={"unit_id": u["id"]}) write_result(conn, tenant_id=tenant_id, segment_id=g["id"], source="ai", us=us, final=us.final, raw=us.raw, ratings=us.ratings, weights=us.weights, counts=us.error_counts, brief=trim_graphemes(f.brief, 150), detailed=f.detailed, errors=errors, positives=positives, key_points=kps, flags=flags, model_info=model_info, qs_version=qs["version"]) recompute_totals(conn, ctx, tenant_id) if feedback: conn.execute("""UPDATE scripts SET strengths_json=%s, weaknesses_json=%s, suggestions_json=%s, summary=%s WHERE id=%s AND tenant_id=%s""", (Jsonb(feedback.strengths[:5]), Jsonb(feedback.weaknesses[:5]), Jsonb(feedback.suggestions[:5]), trim_graphemes(feedback.summary, 300), s["id"], tenant_id)) if minimal: raise_flag(conn, tenant_id=tenant_id, batch_id=s["batch_id"], type_="rubric_minimal", script_id=s["id"], payload={"units": minimal}) for res in results: record_usage(conn, tenant_id=tenant_id, op_type="score_script" if with_feedback else "rescore", credits_milli=int(config.get(conn, "credits.score_script" if with_feedback else "credits.rescore_script", tenant_id)) if res is results[0] else 0, result=res, batch_id=s["batch_id"], script_id=s["id"]) conn.execute("UPDATE scripts SET status='scored', updated_at=now() WHERE id=%s AND tenant_id=%s", (s["id"], tenant_id)) audit(conn, tenant_id=tenant_id, action="score" if with_feedback else "rescore", entity_type="script", entity_id=s["id"], after={"units": len(segs), "ai_units": len(to_ai)}, qs_version=qs["version"], model_info=model_info, tokens={"in": sum(r.tokens_in for r in results), "out": sum(r.tokens_out for r in results)}) jobs.enqueue(conn, "pdf", {"script_id": s["id"], "batch_id": s["batch_id"]}, tenant_id=tenant_id, priority=job["priority"], dedupe_key=f"pdf:{s['id']}", correlation_id=job["correlation_id"]) return {"units": len(segs), "ai_units": len(to_ai), "calls": len(results)} def handle(conn, job) -> dict: """Initial scoring of a whole script, with script-level feedback (SC-00).""" already = conn.execute("""SELECT count(*) AS n FROM score_results r JOIN segments g ON g.id=r.segment_id WHERE g.script_id=%s AND g.tenant_id=%s""", (job["payload_json"]["script_id"], job["tenant_id"])).fetchone()["n"] if already: return {"skipped": "already scored; use rescore"} return _score(conn, job, unit_ids=None, keep_overrides=True, with_feedback=True) def handle_rescore(conn, job) -> dict: p = job["payload_json"] unit_ids = set(p["unit_ids"]) if p.get("unit_ids") else None return _score(conn, job, unit_ids=unit_ids, keep_overrides=p.get("keep_overrides", True), with_feedback=unit_ids is None)