Update app.py

app.py

@@ -1,167 +1,291 @@
+# app.py
 import os
 import json
 import random
 import sqlite3
+import math
 import gradio as gr
 
 from agents import AnalyzerAgent, CoachAgent, PredictiveAgent
+# no upload in app; ocr_agent used offline
+# from ocr_agent import OcrAgent
 
-# Paths
-QUESTIONS_FILE = "questions.json"
+# Paths & constants
+QUESTIONS_JSON = "questions.json"
 DB_FILE = "exam.db"
+MEDIA_DIR = "media"
+MAX_SLOTS = 30   # number of preallocated question slots shown in UI (adjustable)
 
-# Initialize agents
+os.makedirs(MEDIA_DIR, exist_ok=True)
+
+# Agents
 analyzer = AnalyzerAgent()
 coach = CoachAgent()
 predictor = PredictiveAgent()
 
-
-# ---------- Helpers ----------
-def load_question_bank():
-    """Load questions from JSON file."""
-    if not os.path.exists(QUESTIONS_FILE) or os.path.getsize(QUESTIONS_FILE) == 0:
+# ---------- DB & JSON helpers ----------
+def load_questions_json():
+    """Load questions.json if present, else return empty list."""
+    if not os.path.exists(QUESTIONS_JSON) or os.path.getsize(QUESTIONS_JSON) == 0:
         return []
     try:
-        with open(QUESTIONS_FILE, "r", encoding="utf-8") as f:
+        with open(QUESTIONS_JSON, "r", encoding="utf-8") as f:
             return json.load(f)
     except Exception:
         return []
 
-
 def init_db():
-    """Setup SQLite if not exists and sync with JSON."""
+    """Create SQLite DB and ensure questions table exists. Sync JSON -> DB on startup."""
     conn = sqlite3.connect(DB_FILE)
     cur = conn.cursor()
     cur.execute("""
     CREATE TABLE IF NOT EXISTS questions (
         id INTEGER PRIMARY KEY,
         text TEXT,
-        choices TEXT,
+        choices TEXT,    -- JSON-encoded list of strings
         answer TEXT,
         subject TEXT,
         paper INTEGER,
         year INTEGER,
-        image TEXT
-    )""")
+        image TEXT,
+        source TEXT
+    )
+    """)
     conn.commit()
-
-    # Sync JSON → DB
-    questions = load_question_bank()
+    conn.close()
+    # sync JSON -> DB (safe)
+    sync_json_to_db()
+
+def sync_json_to_db():
+    """Insert or ignore entries from questions.json into DB."""
+    questions = load_questions_json()
+    if not questions:
+        return
+    conn = sqlite3.connect(DB_FILE)
+    cur = conn.cursor()
     for q in questions:
-        cur.execute("INSERT OR IGNORE INTO questions VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
-                    (q["id"], q["text"], json.dumps(q["choices"]),
-                     q["answer"], q["subject"], q["paper"], q["year"], q.get("image")))
+        qid = int(q.get("id", 0) or 0)
+        # ensure proper types and defaults
+        text = q.get("text", "")
+        choices = json.dumps(q.get("choices", []), ensure_ascii=False)
+        answer = q.get("answer", "")
+        subject = q.get("subject", "")
+        paper = int(q.get("paper", 2) or 2)
+        year = int(q.get("year", 0) or 0)
+        image = q.get("image")
+        source = q.get("source", "")
+        # insert if id not present
+        if qid:
+            cur.execute("INSERT OR IGNORE INTO questions (id,text,choices,answer,subject,paper,year,image,source) VALUES (?,?,?,?,?,?,?,?,?)",
+                        (qid, text, choices, answer, subject, paper, year, image, source))
+        else:
+            # generate a new id by letting sqlite autoincrement
+            cur.execute("INSERT INTO questions (text,choices,answer,subject,paper,year,image,source) VALUES (?,?,?,?,?,?,?,?)",
+                        (text, choices, answer, subject, paper, year, image, source))
     conn.commit()
     conn.close()
 
-
-def get_questions(subject, paper=2, num_questions=10):
-    """Fetch random questions from DB."""
+def get_questions_from_db(subject, paper=2):
+    """Return all questions for subject & paper as list of dicts."""
     conn = sqlite3.connect(DB_FILE)
     cur = conn.cursor()
-    cur.execute("SELECT * FROM questions WHERE subject=? AND paper=?", (subject, paper))
+    cur.execute("SELECT id, text, choices, answer, subject, paper, year, image, source FROM questions WHERE subject=? AND paper=?", (subject, paper))
     rows = cur.fetchall()
     conn.close()
-
-    if not rows:
-        return []
-
-    selected = random.sample(rows, min(num_questions, len(rows)))
     questions = []
-    for row in selected:
+    for r in rows:
+        choices = []
+        try:
+            choices = json.loads(r[2]) if r[2] else []
+        except Exception:
+            choices = []
         questions.append({
-            "id": row[0],
-            "text": row[1],
-            "choices": json.loads(row[2]),
-            "answer": row[3],
-            "subject": row[4],
-            "paper": row[5],
-            "year": row[6],
-            "image": row[7]
+            "id": r[0],
+            "text": r[1],
+            "choices": choices,
+            "answer": r[3],
+            "subject": r[4],
+            "paper": r[5],
+            "year": r[6],
+            "image": r[7],
+            "source": r[8]
         })
     return questions
 
+# Initialize DB & sync JSON
+init_db()
 
-# ---------- Exam Logic ----------
+# ---------- Exam logic ----------
 def start_exam(subject, num_questions):
-    """Prepare questions for UI rendering."""
-    questions = get_questions(subject, paper=2, num_questions=num_questions)
-    return questions
-
-
-def submit_exam(answers, questions):
-    """Evaluate submitted answers."""
-    correct = 0
+    """
+    Returns:
+      - markdown_texts: list of length MAX_SLOTS (strings)
+      - image_paths: list of length MAX_SLOTS (filepaths or empty string)
+      - radio_defaults: list of length MAX_SLOTS (None)
+      - state: list of selected question dicts (len=num_questions)
+    """
+    # ensure latest JSON -> DB
+    sync_json_to_db()
+
+    pool = get_questions_from_db(subject, paper=2)
+    if not pool:
+        # if empty, return placeholders
+        texts = [f"Q{i+1}: (no question)" for i in range(MAX_SLOTS)]
+        imgs = ["" for _ in range(MAX_SLOTS)]
+        radios = [None for _ in range(MAX_SLOTS)]
+        return texts + imgs + radios + [[]]
+
+    selected = random.sample(pool, min(num_questions, len(pool)))
+    # build display text with lettered choices
+    markdowns = []
+    images = []
+    radios = []
+    for i in range(MAX_SLOTS):
+        if i < len(selected):
+            q = selected[i]
+            # create question block: text + options lines A/B/C/D
+            text_block = q["text"].strip()
+            if q.get("choices"):
+                # ensure we have up to 4 choices
+                opts = q["choices"]
+                # normalize: if choices already prefixed with 'A.' etc, keep them
+                lines = []
+                letters = ["A", "B", "C", "D"]
+                for j, opt in enumerate(opts):
+                    if j < 4:
+                        # if opt includes 'A.' style, keep raw; else prefix
+                        if isinstance(opt, str) and opt.strip().upper().startswith(tuple([l + "." for l in letters])):
+                            lines.append(opt.strip())
+                        else:
+                            lines.append(f"{letters[j]}. {opt}")
+                text_block += "\n\n" + "\n".join(lines)
+            else:
+                # no choices present (rare for Paper2) — leave as-is
+                text_block += "\n\n(A/B/C/D not available)"
+            markdowns.append(text_block)
+            images.append(q.get("image") or "")
+            radios.append(None)
+        else:
+            markdowns.append("")
+            images.append("")
+            radios.append(None)
+
+    # outputs = markdowns (MAX) + images (MAX) + radios (MAX) + [selected-state]
+    return markdowns + images + radios + [selected]
+
+def submit_exam(*args):
+    """
+    args layout: first MAX_SLOTS answers (each None or "A"/"B"/...), last arg is state (selected questions)
+    returns: score_text, wrong_details_text, ai_advice_text
+    """
+    if len(args) < MAX_SLOTS + 1:
+        return "⚠️ Submission failed (bad format).", "", ""
+
+    answers = list(args[:MAX_SLOTS])
+    selected = args[MAX_SLOTS] or []
+
+    total = len(selected)
+    correct_cnt = 0
     wrong_details = []
-
-    for q, ans in zip(questions, answers):
-        if ans == q["answer"]:
-            correct += 1
+    wrong_qids = []
+
+    for i, q in enumerate(selected):
+        ans = (answers[i] or "").strip()
+        expected = str(q.get("answer") or "").strip()
+        # normalization: expected may be 'A' or full text. Accept either letter match or text match.
+        is_correct = False
+        if expected == "":
+            # no answer key: cannot auto-grade; mark as ungraded (count as wrong for now)
+            is_correct = False
         else:
-            wrong_details.append(f"Q{q['id']}: Correct answer was {q['answer']}")
-
-    score_text = f"Score: {correct}/{len(questions)}\nCorrect: {correct}, Wrong: {len(questions) - correct}"
-    analysis = coach.coach("Student got result: " + score_text)
-    return score_text, "\n".join(wrong_details), analysis
-
+            # letter vs text compare
+            if len(expected) == 1 and expected.upper() in ["A", "B", "C", "D"]:
+                if ans.upper() == expected.upper():
+                    is_correct = True
+            else:
+                # expected is likely full text. compare lower-case normalized text
+                if ans.lower() == expected.lower() or ans.lower() in q.get("choices", []):
+                    is_correct = True
+
+        if is_correct:
+            correct_cnt += 1
+        else:
+            correct_label = expected
+            # if expected is a letter, append the text of that choice if available
+            if expected and len(expected) == 1 and q.get("choices"):
+                idx = ord(expected.upper()) - ord("A")
+                if 0 <= idx < len(q["choices"]):
+                    correct_label = f"{expected.upper()}. {q['choices'][idx]}"
+            wrong_details.append(f"Q{i+1} (id:{q['id']}): Your answer: {ans or '(no answer)'} | Correct: {correct_label}")
+            wrong_qids.append(q['id'])
+
+    score_text = f"Score: {correct_cnt} / {total}   (Correct {correct_cnt}, Wrong {total - correct_cnt})"
+
+    # Use ZhipuAI coach to analyze the exam result and give advice
+    try:
+        # build a short context for coach
+        summary = {
+            "total": total,
+            "correct": correct_cnt,
+            "wrong": total - correct_cnt,
+            "wrong_qids": wrong_qids,
+            "subject": selected[0]["subject"] if total>0 else ""
+        }
+        advice = coach.coach(json.dumps(summary, ensure_ascii=False))
+    except Exception as e:
+        advice = f"(AI coach not available) {str(e)}"
+
+    return score_text, "\n\n".join(wrong_details) or "All correct!", advice
 
 # ---------- Gradio UI ----------
-with gr.Blocks() as demo:
-    gr.Markdown("## 📝 SPM Exam Simulator (Form 5)")
-
-    with gr.Tab("Paper 2 Simulator"):
-        subject_sim = gr.Dropdown(["BM", "English", "Math", "History", "Science", "MoralStudies"], label="Subject")
-        num_q = gr.Slider(1, 50, step=1, value=10, label="Number of Questions")
-        run_btn = gr.Button("Start Exam")
-
-        question_state = gr.State()
-        answer_inputs = []
-
-        exam_area = gr.Column(visible=False)
-        results_area = gr.Column(visible=False)
-
-        with exam_area:
-            question_boxes = []
-            for i in range(50):  # max 50
-                q_text = gr.Markdown(visible=False)
-                q_img = gr.Image(type="filepath", visible=False)
-                q_choice = gr.Radio(choices=[], label=f"Answer Q{i+1}", visible=False)
-                question_boxes.append((q_text, q_img, q_choice))
-                answer_inputs.append(q_choice)
-            submit_btn = gr.Button("Submit")
-
-        with results_area:
-            score_out = gr.Textbox(label="Score")
-            wrong_out = gr.Textbox(label="Wrong Answers")
-            advice_out = gr.Textbox(label="AI Advice")
-
-        # Start exam
-        def render_exam(subject, num):
-            qs = start_exam(subject, num)
-            updates = []
-            for i, q in enumerate(qs):
-                updates.append(gr.update(value=f"**Q{i+1}. {q['text']}**", visible=True))  # text
-                updates.append(gr.update(value=q["image"], visible=bool(q["image"])))      # image
-                updates.append(gr.update(choices=q["choices"], visible=True))              # choices
-            # Hide unused slots
-            for j in range(len(qs), 50):
-                updates.extend([gr.update(visible=False), gr.update(visible=False), gr.update(visible=False)])
-            return [qs, gr.update(visible=True), *updates]
-
-        run_btn.click(render_exam, [subject_sim, num_q], [question_state, exam_area] + sum(question_boxes, ()))
-
-        # Submit exam
-        submit_btn.click(
-            fn=submit_exam,
-            inputs=[answer_inputs, question_state],
-            outputs=[score_out, wrong_out, advice_out]
-        ).then(lambda: gr.update(visible=True), None, results_area)
-
-
-# Init DB before launch
-init_db()
+subjects_list = []
+# default subjects pulled from DB
+conn = sqlite3.connect(DB_FILE)
+cur = conn.cursor()
+cur.execute("SELECT DISTINCT subject FROM questions")
+rows = cur.fetchall()
+conn.close()
+subjects_list = [r[0] for r in rows] or ["BM", "English", "Math", "History", "Science", "MoralStudies"]
+
+with gr.Blocks(title="SPM Paper 2 Simulator") as demo:
+    gr.Markdown("# 🧾 SPM Paper 2 — Simulator (Form 5)")
+    gr.Markdown("This simulator renders Paper 2 style MCQ (A/B/C/D). Select subject and how many questions you want. Questions may include diagrams which will be shown under the question.")
+
+    with gr.Row():
+        subject_dd = gr.Dropdown(subjects_list, label="Subject", value=subjects_list[0])
+        num_q = gr.Slider(1, MAX_SLOTS, value=10, step=1, label="Number of Questions")
+
+    start_btn = gr.Button("Start Simulation")
+
+    # Pre-allocate slots for Markdown, Image, Radio — MAX_SLOTS each
+    q_texts = [gr.Markdown(f"Q{i+1}: (not loaded)") for i in range(MAX_SLOTS)]
+    q_images = [gr.Image(type="filepath", visible=False) for _ in range(MAX_SLOTS)]
+    q_radios = [gr.Radio(choices=["A", "B", "C", "D"], label=f"Answer Q{i+1}") for i in range(MAX_SLOTS)]
+
+    # state to hold selected question dicts
+    selected_state = gr.State([])
+
+    start_btn.click(
+        fn=start_exam,
+        inputs=[subject_dd, num_q],
+        outputs=[*q_texts, *q_images, *q_radios, selected_state]
+    )
+
+    submit_btn = gr.Button("Submit Answers")
+    score_out = gr.Textbox(label="Score Summary")
+    wrong_out = gr.Textbox(label="Wrong Answers", lines=8)
+    advice_out = gr.Textbox(label="AI Advice", lines=6)
+
+    submit_btn.click(
+        fn=submit_exam,
+        inputs=[*q_radios, selected_state],
+        outputs=[score_out, wrong_out, advice_out]
+    )
+
+# Launch
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860)
 
-demo.launch(server_name="0.0.0.0", server_port=7860)