import re from aix import jobs from aix.main import _handlers, run_one from .conftest import add_script from .helpers import make_pdf EXTRACT = { 1: {"kind": "cover", "legibility": 8, "cover": {"roll": "২৪"}, "elements": [{"type": "line", "text": "রোল: ২৪", "confidence": 0.97, "lang": ["bn"]}]}, 2: {"kind": "answer", "legibility": 7, "elements": [ {"type": "heading", "text": "১", "confidence": 0.99, "bbox": [0.05, 0.05, 0.1, 0.08]}, {"type": "line", "text": "বাংলাদেশের রাজধানী ঢাকা।", "confidence": 0.95, "lang": ["bn"], "bbox": [0.1, 0.05, 0.9, 0.08]}, {"type": "heading", "text": "২", "confidence": 0.98}, {"type": "heading", "text": "ক)", "confidence": 0.98}, {"type": "line", "text": "কবির নাম জসিমউদ্দিন", "confidence": 0.9, "lang": ["bn"]}, {"type": "heading", "text": "খ)", "confidence": 0.98}, {"type": "line", "text": "উদ্দীপকে গ্রামের ছবি আছে", "confidence": 0.6, "lang": ["bn"], "flags": ["uncertain"], "candidates": ["গ্রামের", "গ্রামীণ"]}]}, 3: {"kind": "answer", "legibility": 6, "elements": [ {"type": "heading", "text": "৩", "confidence": 0.98}, {"type": "heading", "text": "ক)", "confidence": 0.98}, {"type": "line", "text": "নদীর নাম পদ্মা", "confidence": 0.93, "lang": ["bn"]}, {"type": "heading", "text": "খ)", "confidence": 0.98}, {"type": "equation", "text": "x = 2 + 3", "latex": "x = 2 + 3", "confidence": 0.9}, {"type": "line", "text": "অতএব উত্তর ৫", "confidence": 0.94, "lang": ["bn"]}]}, } # Section B (any 1 of 2): Q2 should score lower than Q3, so Q3 counts under 'best'. QUALITY = {"u1": (1.0, 1.0), "u2a": (0.5, 0.5), "u2b": (0.4, 0.4), "u3a": (1.0, 1.0), "u3b": (0.9, 0.8)} def score_response(world): rev = {v: k for k, v in world.items()} def build(call): text = call["parts"][0].text unit_ids = re.findall(r"=== Unit (\S+)", text) units = [] for uid in dict.fromkeys(unit_ids): c, comp = QUALITY[rev[uid]] line = re.search(rf"=== Unit {uid}.*?\n\[(p\d+-l\d+)\]", text, re.S) units.append({"unit_id": uid, "correctness": c, "completeness": comp, "key_points": [{"key_point_id": "KP1", "status": "present" if c > 0.5 else "partial", "line_ref": line.group(1) if line else None}], "error_items": ([{"type": "spelling", "line_ref": line.group(1), "as_written": "জসিমউদ্দিন", "correction": "জসীমউদ্‌দীন", "remark": "বানান ভুল"}] if rev[uid] == "u2a" else []), "positive_items": [], "brief": "সংক্ষিপ্ত কারণ " * 20, "detailed": "বিস্তারিত"}) return {"units": units, "strengths": ["s1", "s2", "s3"], "weaknesses": ["w1", "w2", "w3"], "suggestions": ["g1", "g2", "g3"], "summary": "সারাংশ"} return build def drain(conn, handlers, limit=50): n = 0 while n < limit and run_one(conn, handlers): n += 1 return n def test_full_pipeline(conn, storage, world, fake): sid = add_script(conn, storage, world, make_pdf(["text", "text", "text", "blank"]), roll="23") handlers = {k: v for k, v in _handlers().items() if k != "noop"} jobs.enqueue(conn, "rasterize", {"script_id": sid, "batch_id": world["batch"]}, tenant_id=world["tenant"]) assert run_one(conn, handlers) # Make page order deterministic for the fake provider: earlier pages get higher priority. conn.execute("""UPDATE jobs j SET priority = 1000 - p.sort FROM pages p WHERE j.type='extract_page' AND p.id = j.payload_json->>'page_id'""") for page in (1, 2, 3): fake.push("ExtractionPage", EXTRACT[page]) fake.push("ScriptFinding", score_response(world)) drain(conn, handlers) failed = conn.execute("SELECT type, last_error FROM jobs WHERE status='failed'").fetchall() assert not failed, failed s = conn.execute("SELECT * FROM scripts WHERE id=%s", (sid,)).fetchone() assert s["status"] == "pdf_generated" assert s["roll_from_cover"] == "24" assert float(s["legibility_avg"]) == 6.5 pages = conn.execute("SELECT sort, kind FROM pages WHERE script_id=%s ORDER BY sort", (sid,)).fetchall() assert [p["kind"] for p in pages] == ["cover", "answer", "answer", "blank"] rows = conn.execute( """SELECT u.id AS unit, g.counted, g.not_attempted, r.final_score, r.brief, r.error_items_json, r.ratings_json FROM segments g JOIN score_results r ON r.segment_id=g.id AND r.is_current JOIN scoring_units u ON u.id=g.scoring_unit_id WHERE g.script_id=%s""", (sid,)).fetchall() rev = {v: k for k, v in world.items()} res = {rev[r["unit"]]: r for r in rows} assert set(res) == {"u1", "u2a", "u2b", "u3a", "u3b"} # u1: perfect content, legibility 6.5 -> handwriting 0.65 * 5% -> raw 1.965 -> 2.0 assert float(res["u1"]["final_score"]) == 2.0 # Choice rule: Q3 (better) counts, Q2 does not. assert res["u3a"]["counted"] and res["u3b"]["counted"] assert not res["u2a"]["counted"] and not res["u2b"]["counted"] # error item mapped back to an element id; brief trimmed to 150 graphemes assert res["u2a"]["error_items_json"][0]["element_id"] import regex assert len(regex.findall(r"\X", res["u1"]["brief"])) == 150 # graphemes, not code points (SC-14) total = float(res["u1"]["final_score"]) + float(res["u3a"]["final_score"]) + float(res["u3b"]["final_score"]) assert float(s["total_score"]) == total and float(s["total_target"]) == 5 assert s["strengths_json"] == ["s1", "s2", "s3"] flags = {f["type"] for f in conn.execute("SELECT type FROM flags WHERE script_id=%s", (sid,)).fetchall()} assert {"cover_roll_mismatch", "low_confidence_extraction", "uncertain_affects_score", "choice_rule_applied"} <= flags usage = conn.execute("SELECT op_type, count(*) AS n, sum(credits_milli) AS c FROM usage_records GROUP BY op_type").fetchall() u = {r["op_type"]: r for r in usage} assert u["extract_page"]["n"] == 3 and u["extract_page"]["c"] == 3000 # blank page not charged assert u["score_script"]["n"] == 1 # One scoring call for the whole script (SC-00); PDF built even though most elements have no bbox. assert sum(1 for c in fake.calls if c["schema"] == "ScriptFinding") == 1 assert conn.execute("SELECT status FROM jobs WHERE type='pdf'").fetchone()["status"] == "done" assert conn.execute("SELECT count(*) AS n FROM script_pdfs WHERE script_id=%s", (sid,)).fetchone()["n"] == 1 def test_rescore_keeps_teacher_override(conn, storage, world, fake): test_full_pipeline(conn, storage, world, fake) seg = conn.execute("""SELECT g.id, r.version FROM segments g JOIN score_results r ON r.segment_id=g.id AND r.is_current WHERE g.scoring_unit_id=%s""", (world["u1"],)).fetchone() # Teacher override (what Laravel's OverrideService writes) from aix.pipeline.score import write_result write_result(conn, tenant_id=world["tenant"], segment_id=seg["id"], source="teacher", us=None, final=1.5, raw=1.5, ratings={}, weights={}, counts={}, brief="teacher", detailed="teacher", errors=[], positives=[], key_points=[], flags=[], model_info=None, qs_version=1) sid = conn.execute("SELECT id FROM scripts").fetchone()["id"] fake.push("UnitsOnlyFinding", lambda call: score_response(world)(call)) jobs.enqueue(conn, "rescore", {"script_id": sid, "batch_id": world["batch"], "unit_ids": [world["u1"], world["u3b"]]}, tenant_id=world["tenant"]) drain(conn, {k: v for k, v in _handlers().items() if k == "rescore"}) r = conn.execute("""SELECT source, final_score FROM score_results r JOIN segments g ON g.id=r.segment_id WHERE g.scoring_unit_id=%s AND r.is_current""", (world["u1"],)).fetchone() assert r["source"] == "teacher" and float(r["final_score"]) == 1.5 # only u3b went to the model last = fake.calls[-1]["parts"][0].text assert world["u3b"] in last and world["u1"] not in last