"""assemble job (M4): ordered elements -> answer segments per scoring unit. Deterministic, no AI. Payload: {script_id, batch_id}. Design §8.1, spec SG-01…SG-08.""" from __future__ import annotations from dataclasses import dataclass, field from .. import jobs from ..db import Jsonb from ..ids import new_id from ..records import audit, raise_flag from .labels import label_prefix, looks_like_label, normalize, parse CONTINUATION_WORDS = ("continue", "contd", "cont.", "চলমান", "বাকি", "অবশিষ্ট", "পরের অংশ", "পূর্বের পর") CONF_HEADING = 0.95 # explicit heading element with full label CONF_SUB_UNDER_OPEN = 0.92 # "ক)" resolved under the open question CONF_LINE_LABEL = 0.85 # label found at the start of an ordinary line CONF_PREAMBLE = 0.70 # text under a question heading with no part label FLAG_BELOW = 0.90 @dataclass class Attempt: unit_id: str | None # None = unmapped confidence: float element_ids: list[str] = field(default_factory=list) lines: list[str] = field(default_factory=list) struck: list[bool] = field(default_factory=list) flags: list[str] = field(default_factory=list) raw_label: str | None = None def load_structure(conn, batch_id: str, tenant_id: str): batch = conn.execute("SELECT * FROM batches WHERE id=%s AND tenant_id=%s", (batch_id, tenant_id)).fetchone() section_ids = batch["section_ids_json"] or [] qrows = conn.execute( """SELECT q.*, s.answer_count FROM questions q JOIN qs_sections s ON s.id=q.section_id WHERE q.question_set_id=%s ORDER BY s.sort, q.sort""", (batch["question_set_id"],)).fetchall() if section_ids: qrows = [q for q in qrows if q["section_id"] in section_ids] urows = 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 u.sort""", (batch["question_set_id"],)).fetchall() units_by_q: dict[str, list[dict]] = {} for u in urows: units_by_q.setdefault(u["question_id"], []).append(u) return batch, [q for q in qrows], units_by_q def build_index(questions, units_by_q): """normalized question number -> question; (question_id, sub) -> unit.""" q_index: dict[str, dict] = {} for q in questions: for lab in [q["label"], *(q["alt_labels_json"] or [])]: qn, _ = parse(lab) if qn: q_index.setdefault(qn, q) u_index: dict[tuple[str, str | None], dict] = {} for q in questions: for u in units_by_q.get(q["id"], []): sub = None if u["label"]: _, sub = parse(u["label"]) sub = sub or normalize(u["label"]) u_index[(q["id"], sub)] = u return q_index, u_index def load_elements(conn, script_id: str, tenant_id: str): return conn.execute( """SELECT e.*, p.sort AS page_sort, p.kind AS page_kind FROM elements e JOIN pages p ON p.id=e.page_id WHERE p.script_id=%s AND e.tenant_id=%s AND p.kind IN ('answer','supplementary','rough') ORDER BY p.sort, e.sort""", (script_id, tenant_id)).fetchall() def ref(el) -> str: return f"p{el['page_sort']}-l{el['sort']}" def render(el) -> str: """One element as text for the scoring prompt (design §M4 step 7).""" flags = el["flags_json"] or [] if "illegible" in flags: body = "⟦illegible⟧" elif el["type"] == "equation" or el["type"] == "chem": body = f"${el['latex']}$" if el["latex"] else (el["text"] or "") elif el["type"] == "figure": body = f"[FIGURE: {el['description']}]" elif el["type"] == "table" and el["table_json"]: rows = el["table_json"] body = "\n".join("| " + " | ".join(r) + " |" for r in rows) else: body = el["text"] or "" if "uncertain" in flags and el["candidates_json"]: body = f"⟦uncertain: {'|'.join(el['candidates_json'])}⟧" if "struck" in flags: body = f"~~{body}~~" return f"[{ref(el)}] {body}" def segment(elements, questions, units_by_q, q_index, u_index) -> list[Attempt]: attempts: list[Attempt] = [] by_unit: dict[str, list[Attempt]] = {} current: Attempt | None = None open_q: dict | None = None preamble: Attempt | None = None # lines under a multi-part question heading before any part label def start(unit_id, conf, el, raw) -> Attempt: nonlocal current, preamble a = Attempt(unit_id=unit_id, confidence=conf, raw_label=raw) q_units = [u["id"] for u in units_by_q.get(open_q["id"], [])] if open_q else [] if preamble and preamble.lines and unit_id in q_units: if unit_id == q_units[0]: # Text under the question heading, then the first part's label: one answer. a.element_ids += preamble.element_ids; a.lines += preamble.lines; a.struck += preamble.struck preamble.lines.clear(); preamble.element_ids.clear(); preamble.struck.clear() else: # The student skipped the first part's label: the preamble is the first part's answer. flush_preamble() a.element_ids.append(el["id"]); a.lines.append(render(el)); a.struck.append(False) attempts.append(a) if unit_id: by_unit.setdefault(unit_id, []).append(a) current = a return a def flush_preamble(): nonlocal preamble if preamble and preamble.lines and open_q: units = units_by_q.get(open_q["id"], []) if units: first = units[0]["id"] tgt = by_unit.get(first) if tgt: t = tgt[-1] t.element_ids += preamble.element_ids; t.lines += preamble.lines; t.struck += preamble.struck else: preamble.unit_id, preamble.confidence = first, CONF_PREAMBLE attempts.append(preamble); by_unit.setdefault(first, []).append(preamble) preamble = None for el in elements: flags = el["flags_json"] or [] if "rough" in flags or el["page_kind"] == "rough": continue text = el["text"] or "" is_heading = el["type"] == "heading" if is_heading or (el["type"] == "line" and looks_like_label(text)): raw = el["q_label"] or text label_part = raw qn, sub = parse(label_part if is_heading else label_prefix(text)) conf = CONF_HEADING if is_heading else CONF_LINE_LABEL q = q_index.get(qn) if qn else None if qn and not q: flush_preamble(); open_q = None start(None, conf, el, raw) # label for a question not in this paper continue if q: if q is not open_q: flush_preamble() open_q = q elif sub and open_q: conf = min(conf, CONF_SUB_UNDER_OPEN) if not open_q: start(None, conf, el, raw) continue units = units_by_q.get(open_q["id"], []) unit = u_index.get((open_q["id"], sub)) if sub else (units[0] if len(units) == 1 else None) if unit is None and sub is None: # Multi-part question heading without a part: collect a preamble. preamble = Attempt(unit_id=None, confidence=CONF_PREAMBLE) current = preamble if not is_heading: # the line also carries answer text preamble.element_ids.append(el["id"]); preamble.lines.append(render(el)); preamble.struck.append(False) continue if unit is None: start(None, conf, el, raw) # part letter not in this question continue prior = by_unit.get(unit["id"]) if prior and any(w in text.lower() for w in CONTINUATION_WORDS): current = prior[-1] current.element_ids.append(el["id"]); current.lines.append(render(el)); current.struck.append(False) continue a = start(unit["id"], conf, el, raw) if prior: for x in prior + [a]: if "duplicate" not in x.flags: x.flags.append("duplicate") continue # ordinary content line if current is None: current = Attempt(unit_id=None, confidence=1.0) attempts.append(current) current.element_ids.append(el["id"]); current.lines.append(render(el)) current.struck.append("struck" in flags) flush_preamble() return [a for a in attempts if a.lines] def handle(conn, job) -> dict: p = job["payload_json"] tenant_id, script_id = job["tenant_id"], p["script_id"] script = conn.execute("SELECT * FROM scripts WHERE id=%s AND tenant_id=%s FOR UPDATE", (script_id, tenant_id)).fetchone() if not script: raise jobs.FatalError("script not found") batch, questions, units_by_q = load_structure(conn, script["batch_id"], tenant_id) q_index, u_index = build_index(questions, units_by_q) elements = load_elements(conn, script_id, tenant_id) # Idempotent: rebuild segments only if none are scored yet. scored = conn.execute("""SELECT count(*) AS n FROM score_results r JOIN segments s ON s.id=r.segment_id WHERE s.script_id=%s AND s.tenant_id=%s""", (script_id, tenant_id)).fetchone()["n"] if scored: return {"skipped": "already scored; use rescore"} conn.execute("DELETE FROM segments WHERE script_id=%s AND tenant_id=%s", (script_id, tenant_id)) conn.execute("""DELETE FROM flags WHERE script_id=%s AND tenant_id=%s AND status='open' AND type IN ('unmapped_content','low_mapping_confidence','duplicate_answer')""", (script_id, tenant_id)) attempts = segment(elements, questions, units_by_q, q_index, u_index) attempt_no: dict[str, int] = {} seen_units: set[str] = set() unmapped = 0 for a in attempts: all_struck = bool(a.struck) and all(a.struck) n = attempt_no[a.unit_id] = attempt_no.get(a.unit_id, 0) + 1 if a.unit_id else 1 seg_id = new_id() flags = list(a.flags) + (["all_struck"] if all_struck else []) conn.execute( """INSERT INTO segments (id, tenant_id, script_id, scoring_unit_id, element_ids_json, content_text, confidence, flags_json, counted, attempt_no) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""", (seg_id, tenant_id, script_id, a.unit_id, Jsonb(a.element_ids), "\n".join(a.lines), round(a.confidence, 3), Jsonb(flags), a.unit_id is not None and not all_struck, n)) if a.unit_id is None: unmapped += 1 raise_flag(conn, tenant_id=tenant_id, batch_id=batch["id"], type_="unmapped_content", script_id=script_id, segment_id=seg_id, payload={"label": a.raw_label, "first_line": a.lines[0][:120]}) continue seen_units.add(a.unit_id) if a.confidence < FLAG_BELOW: raise_flag(conn, tenant_id=tenant_id, batch_id=batch["id"], type_="low_mapping_confidence", script_id=script_id, segment_id=seg_id, payload={"confidence": a.confidence, "label": a.raw_label}) if "duplicate" in a.flags and n > 1: raise_flag(conn, tenant_id=tenant_id, batch_id=batch["id"], type_="duplicate_answer", script_id=script_id, segment_id=seg_id, payload={"attempt_no": n}) # Not attempted (SG-07): one empty segment per unit with no content. for q in questions: for u in units_by_q.get(q["id"], []): if u["id"] not in seen_units: conn.execute( """INSERT INTO segments (id, tenant_id, script_id, scoring_unit_id, element_ids_json, content_text, confidence, counted, not_attempted) VALUES (%s,%s,%s,%s,'[]'::jsonb,'',1.0,true,true)""", (new_id(), tenant_id, script_id, u["id"])) leg = conn.execute("""SELECT avg(legibility) AS a FROM pages WHERE script_id=%s AND tenant_id=%s AND kind IN ('answer','supplementary') AND legibility IS NOT NULL""", (script_id, tenant_id)).fetchone()["a"] conn.execute("UPDATE scripts SET status='mapped', legibility_avg=%s, updated_at=now() WHERE id=%s AND tenant_id=%s", (round(float(leg), 2) if leg is not None else None, script_id, tenant_id)) audit(conn, tenant_id=tenant_id, action="assemble", entity_type="script", entity_id=script_id, after={"segments": len(attempts), "unmapped": unmapped}) review_required = (conn.execute("SELECT settings_json->>'review_mode' AS m FROM tenants WHERE id=%s", (tenant_id,)).fetchone()["m"] == "review_required") open_flags = conn.execute("""SELECT count(*) AS n FROM flags WHERE script_id=%s AND tenant_id=%s AND status='open' AND severity IN ('block','warn')""", (script_id, tenant_id)).fetchone()["n"] if review_required and open_flags: conn.execute("UPDATE scripts SET status='awaiting_review' WHERE id=%s AND tenant_id=%s", (script_id, tenant_id)) else: jobs.enqueue(conn, "score", {"script_id": script_id, "batch_id": batch["id"]}, tenant_id=tenant_id, priority=job["priority"], dedupe_key=f"score:{script_id}", correlation_id=job["correlation_id"]) return {"segments": len(attempts), "unmapped": unmapped, "not_attempted": sum( 1 for q in questions for u in units_by_q.get(q["id"], []) if u["id"] not in seen_units)}