Update app.py

app.py

@@ -1,183 +1,272 @@
-# agents.py
+# app.py
 import os
+import re
+import json
 import random
-from collections import Counter
-from typing import List, Dict, Any
-
-# Accept both env var names for backward compatibility
-GLM_API_KEY = os.getenv("ZHIPUAI_API_KEY") or os.getenv("zhipuai_api_key")
-
-
-class AnalyzerAgent:
+import subprocess
+import gradio as gr
+
+from agents import AnalyzerAgent, CoachAgent, PredictiveAgent
+from ocr_agent import OcrAgent
+
+# Constants
+DATA_DIR = "data"
+QUESTIONS_FILE = "questions.json"
+VALID_SUBJECTS = ["BM", "English", "Math", "History", "Science", "MoralStudies",
+                  "Accounting", "Economics", "Business"]
+
+os.makedirs(DATA_DIR, exist_ok=True)
+
+# Agents and OCR
+analyzer = AnalyzerAgent()
+coach_agent = CoachAgent()
+predictor = PredictiveAgent()
+ocr_agent = OcrAgent()
+
+# Load question bank safely
+def load_question_bank():
+    if not os.path.exists(QUESTIONS_FILE):
+        return []
+    try:
+        with open(QUESTIONS_FILE, "r", encoding="utf-8") as f:
+            content = f.read().strip()
+            return json.loads(content) if content else []
+    except Exception:
+        return []
+
+QUESTION_BANK = load_question_bank()
+
+
+# Merge runner
+def run_merge():
+    """Run merge_questions.py (rebuilds questions.json) and reload QUESTION_BANK."""
+    try:
+        subprocess.run(["python", "merge_questions.py"], check=True)
+        global QUESTION_BANK
+        QUESTION_BANK = load_question_bank()
+        return True, "Merge successful."
+    except subprocess.CalledProcessError as e:
+        return False, f"Merge failed: {e}"
+
+
+# Utility: normalize subject token and display
+def subject_token_from_display(display):
+    if not display:
+        return "bm"
+    return display.strip().lower()
+
+
+def normalize_display_subject(token):
+    if not token:
+        return "BM"
+    t = token.strip().lower()
+    mapping = {
+        "bm": "BM",
+        "bahasa": "BM",
+        "bahasamelayu": "BM",
+        "english": "English",
+        "math": "Math",
+        "mathematics": "Math",
+        "history": "History",
+        "sejarah": "History",
+        "science": "Science",
+        "physics": "Science",
+        "moral": "MoralStudies",
+        "moralstudies": "MoralStudies",
+    }
+    return mapping.get(t, token.capitalize())
+
+
+def autodetect_from_filename(path):
+    """Detect year and subject token from filename like spm_2018_bm.pdf"""
+    if not path:
+        return None, None
+    fname = os.path.basename(path)
+    m = re.search(r"spm[_\-]?(\d{4})[_\-]?([A-Za-z]+)", fname, re.IGNORECASE)
+    if not m:
+        return None, None
+    year = m.group(1)
+    subj = m.group(2).lower()
+    return year, subj
+
+
+# ===== OCR upload + auto-merge =====
+def process_pdf_and_merge(file_path, display_subject, year):
     """
-    Produces topic-level accuracy and weak-topic recommendations.
-    Input: per_question dict {qid: {"user":..., "correct":..., "topics":[...]}}
+    file_path: local filepath (gr.File type='filepath')
+    display_subject: e.g. "BM"
+    year: "2018"
     """
-    def analyze(self, per_question: Dict[str, Dict[str, Any]]) -> Dict[str, Any]:
-        topic_stats = {}
-        for qid, info in per_question.items():
-            topics = info.get("topics") or []
-            user = info.get("user")
-            correct = info.get("correct")
-            is_correct = (correct is not None and user is not None and str(user).strip() == str(correct).strip())
-            for t in topics:
-                if t not in topic_stats:
-                    topic_stats[t] = {"correct": 0, "total": 0}
-                topic_stats[t]["total"] += 1
-                if is_correct:
-                    topic_stats[t]["correct"] += 1
-
-        topic_accuracy = {}
-        weak_topics = []
-        for t, stats in topic_stats.items():
-            total = stats["total"]
-            correct = stats["correct"]
-            acc = round((correct / total) * 100, 2) if total > 0 else 0.0
-            topic_accuracy[t] = {"accuracy_percent": acc, "total": total}
-            if total >= 3 and acc < 65.0:
-                weak_topics.append(t)
-
-        recommendation = "Focus on: " + ", ".join(weak_topics) if weak_topics else "No major weak topics detected."
-
-        return {
-            "topic_accuracy": topic_accuracy,
-            "weak_topics": weak_topics,
-            "recommendation": recommendation
-        }
-
-
-class CoachAgent:
+    if not file_path:
+        return "No file uploaded."
+
+    subj_token = subject_token_from_display(display_subject)
+    try:
+        qfile, scheme_file = ocr_agent.extract_questions_to_files(
+            pdf_path=file_path, year=str(year), subject_token=subj_token, out_dir=DATA_DIR
+        )
+    except Exception as e:
+        return f"❌ OCR failed: {e}"
+
+    ok, msg = run_merge()
+    if ok:
+        return f"✅ OCR saved: {os.path.basename(qfile)} & {os.path.basename(scheme_file)}. Merge: {msg}"
+    else:
+        return f"⚠️ OCR saved: {os.path.basename(qfile)} & {os.path.basename(scheme_file)}. Merge: {msg}"
+
+
+# ===== Exam generation =====
+def generate_exam(display_subject, num_questions, include_predicted):
     """
-    Short actionable coaching guidance for Form5 SPM students.
+    display_subject: "BM" etc.
+    returns exam_data (list) and status text and exam_data (for state)
     """
-    def coach(self, analysis: Dict[str, Any], level: str, subject: str) -> Dict[str, Any]:
-        weak = analysis.get("weak_topics", [])
-        if not weak:
-            tips = [
-                "Keep revising key topics and time yourself on mock papers.",
-                "Review incorrect solutions and understand each step.",
-                "Do a mixed-topic mock weekly to build stamina."
-            ]
-        else:
-            tips = [
-                f"Spend 20–30 minutes daily on {weak[0]} (split into focused tasks).",
-                "Solve short targeted questions and check worked solutions.",
-                "Teach a concept to someone else — it stabilizes understanding."
-            ]
-
-        practice = []
-        for i, t in enumerate(weak[:3], start=1):
-            practice.append({
-                "text": f"Short practice prompt on {t}: (write/solve one short item)",
-                "topic": t
-            })
-
-        return {"tips": tips, "study_plan": "20 min/day for weak topics + weekly mock", "practice": practice}
-
-
-class PredictiveAgent:
+    subj_key = f"Form5_{display_subject}"
+    pool = [q for q in QUESTION_BANK if q.get("subject") == subj_key]
+
+    predicted_questions = []
+    if include_predicted:
+        predicted_questions = predictor.generate_predictions(level="Form5",
+                                                            subject=display_subject,
+                                                            n=min(10, max(1, num_questions // 2)),
+                                                            question_bank=QUESTION_BANK)
+
+    combined = pool + predicted_questions
+    if not combined:
+        return [], f"No questions found for {display_subject}. Upload papers (2018–2024).", []
+
+    random.shuffle(combined)
+    selected = combined[:min(num_questions, len(combined))]
+
+    # For safety, return minimal exam objects
+    exam_data = []
+    for q in selected:
+        # if predicted questions include correct_answer, it can be included (but they are in-memory)
+        exam_data.append({
+            "id": q.get("id"),
+            "text": q.get("text"),
+            "choices": q.get("choices", []),
+            "topics": q.get("topics", []),
+            "source": q.get("source", "pastpaper")
+        })
+    return exam_data, f"Prepared {len(exam_data)} questions ({len(predicted_questions)} predicted)", exam_data
+
+
+# ===== Submit & grade =====
+def submit_exam(answers_json, exam_state, display_subject):
     """
-    Generates heuristic or LLM-based predicted Form5 questions (in-memory only).
-    Public methods:
-      - predict(subject, level, count) -> list of question dicts
-      - summary(level, subject) -> dict
+    answers_json: dict where keys are stringified ids -> answer text (or choice text)
+    exam_state: the exam_data (list) saved in gr.State
     """
-
-    def __init__(self):
-        self.api_key = GLM_API_KEY
-
-    def _top_topics_from_bank(self, question_bank: List[Dict], subject_display: str, top_k=6):
-        subj_key = f"Form5_{subject_display}"
-        counter = Counter()
-        total = 0
-        for q in question_bank:
-            if q.get("subject") != subj_key:
-                continue
-            total += 1
-            for t in q.get("topics", []):
-                counter[t] += 1
-        if total == 0:
-            return []
-        return [t for t, _ in counter.most_common(top_k)]
-
-    def predict(self, subject: str, level: str = "Form5", count: int = 5) -> List[Dict]:
-        """
-        Return `count` predicted MCQs. If no GLM key present, produce conservative heuristic items.
-        Predictions have id >= 900000, source='predicted', and may include 'confidence'.
-        """
-        preds = []
-        base = 900000
-        # fallback topics per subject
-        fallback_topics = {
-            "BM": ["perbendaharaan_kata", "tatabahasa"],
-            "English": ["vocabulary", "grammar"],
-            "Math": ["algebra", "geometry"],
-            "History": ["events", "dates"],
-            "Science": ["physics", "chemistry"],
-            "MoralStudies": ["ethics", "values"]
-        }
-        topics = fallback_topics.get(subject, ["general"])
-
-        # Try to use a simple LLM call if API key present (non-blocking, conservative)
-        # NOTE: We keep the interface simple: if GLM unavailable or fails, fall back to heuristics.
-        if self.api_key:
-            try:
-                # Placeholder: implement GLM call here if you provide endpoint details.
-                # For now, fall back to heuristics to avoid runtime dependency.
-                raise RuntimeError("GLM call not implemented in this environment")
-            except Exception:
-                pass
-
-        # Heuristic generation
-        for i in range(count):
-            t = topics[i % len(topics)]
-            q = self._heuristic_question(subject, t, idx=i + 1)
-            q["id"] = base + i
-            q["source"] = "predicted"
-            q["confidence"] = round(random.uniform(0.35, 0.75), 2)
-            preds.append(q)
-        return preds
-
-    def _heuristic_question(self, subject: str, topic: str, idx: int) -> Dict:
-        # provide realistic-looking stems & 4 choices tailored by subject
-        if subject == "BM":
-            stem = f"Pilih sinonim bagi perkataan 'gembira'."
-            choices = ["Sedih", "Gembira", "Marah", "Letih"]
-            correct = "Gembira"
-        elif subject == "English":
-            stem = "Choose the correct synonym for 'happy'."
-            choices = ["Sad", "Joyful", "Angry", "Tired"]
-            correct = "Joyful"
-        elif subject == "Math":
-            stem = "If 2x + 3 = 11, what is x?"
-            choices = ["2", "3", "4", "5"]
-            correct = "4"
-        elif subject == "Science":
-            stem = "What is the SI unit of force?"
-            choices = ["Joule", "Newton", "Pascal", "Watt"]
-            correct = "Newton"
-        elif subject == "History":
-            stem = "Which year is associated with Malayan independence?"
-            choices = ["1945", "1957", "1963", "1975"]
-            correct = "1957"
-        elif subject == "MoralStudies":
-            stem = "Which value best represents mutual respect?"
-            choices = ["Greed", "Respect", "Laziness", "Selfishness"]
-            correct = "Respect"
+    exam_data = exam_state or []
+    if not exam_data:
+        return "No exam data found.", {}, {}, {}, gr.update(visible=False), gr.update(visible=True)
+
+    correct = 0
+    graded = 0
+    per_question = {}
+
+    for q in exam_data:
+        qid = q.get("id")
+        key = str(qid)
+        user_ans = answers_json.get(key) if isinstance(answers_json, dict) else None
+
+        # determine correct answer
+        correct_ans = None
+        if q.get("source") == "predicted":
+            # predicted question may have correct_answer inside QUESTION_BANK? predictor sets it when generating.
+            # We didn't include correct_answer in exam state by default; attempt to find inside QUESTION_BANK (unlikely)
+            correct_ans = q.get("correct_answer")
         else:
-            stem = f"Practice predicted question on {topic}."
-            choices = ["A", "B", "C", "D"]
-            correct = "A"
-
-        return {"text": stem, "choices": choices, "correct_answer": correct, "topics": [topic], "difficulty": 3}
-
-    def summary(self, level: str, subject: str, question_bank: List[Dict] = None) -> Dict:
-        # Provide simple summary: top topics from bank if available
-        topics = self._top_topics_from_bank(question_bank or [], subject) if question_bank else []
-        return {"level": level, "subject": subject, "top_topics": topics, "note": "Predictions are practice-oriented heuristics."}
+            orig = next((item for item in QUESTION_BANK if item.get("id") == qid), None)
+            if orig:
+                correct_ans = orig.get("correct_answer")
+
+        per_question[str(qid)] = {"user": user_ans, "correct": correct_ans, "topics": q.get("topics", [])}
+
+        # grade only when correct_answer is not None
+        if correct_ans is not None:
+            graded += 1
+            if user_ans is not None and str(user_ans).strip() == str(correct_ans).strip():
+                correct += 1
+
+    score = round(100 * correct / graded, 2) if graded > 0 else "N/A (no answer keys)"
+
+    analysis = analyzer.analyze(per_question)
+    coach = coach_agent.coach(analysis, "Form5", display_subject)
+    pred_summary = predictor.summary(level="Form5", subject=display_subject, question_bank=QUESTION_BANK)
+
+    return (
+        f"Your Score: {score}%",
+        analysis,
+        coach,
+        pred_summary,
+        gr.update(visible=True),
+        gr.update(visible=True)
+    )
+
+
+# ===== Prefill handler for upload UI =====
+def prefill_subject_year_from_file(file_path):
+    if not file_path:
+        return "BM", "2018"
+    year, subj_token = autodetect_from_filename(file_path)
+    subj_display = normalize_display_subject(subj_token) if subj_token else "BM"
+    return subj_display, year if year else "2018"
+
+
+# ===== Gradio UI =====
+with gr.Blocks() as demo:
+    gr.Markdown("## SPM Exam Simulator — Form 5 (Past papers 2018–2024)")
+
+    with gr.Tab("Upload (OCR → JSON → Merge)"):
+        pdf_file = gr.File(label="Upload SPM PDF (e.g., spm_2018_bm.pdf)", type="filepath")
+        subject_dropdown = gr.Dropdown(choices=VALID_SUBJECTS, value="BM", label="Subject (override)")
+        year_dropdown = gr.Dropdown(choices=[str(y) for y in range(2018, 2025)], value="2018", label="Year")
+        process_btn = gr.Button("Process PDF → JSON + Merge")
+        ocr_status = gr.Textbox(label="Status", interactive=False)
+
+        pdf_file.change(prefill_subject_year_from_file, inputs=[pdf_file], outputs=[subject_dropdown, year_dropdown])
+        process_btn.click(process_pdf_and_merge, inputs=[pdf_file, subject_dropdown, year_dropdown], outputs=[ocr_status])
+
+    with gr.Tab("Exam Simulator"):
+        subject_sel = gr.Dropdown(choices=["BM", "English", "Math", "History", "Science", "MoralStudies"],
+                                  value="Math", label="Subject")
+        num_q = gr.Slider(minimum=5, maximum=50, step=5, value=10, label="Number of Questions")
+        include_pred = gr.Checkbox(value=True, label="Include AI-predicted questions (in-memory only)")
+        start_btn = gr.Button("Start Exam")
+        exam_state = gr.State()
+
+        exam_display = gr.JSON(label="Exam Questions")
+        status_display = gr.Textbox(label="Status", interactive=False)
+
+        start_btn.click(generate_exam,
+                        inputs=[subject_sel, num_q, include_pred],
+                        outputs=[exam_display, status_display, exam_state])
+
+    with gr.Tab("Submit & Results"):
+        answers_input = gr.JSON(label='Submit Your Answers as JSON (e.g. {"1001":"Seronok", "900000":"4"})')
+        submit_btn = gr.Button("Submit Answers")
+
+        score_out = gr.Textbox(label="Score")
+        analysis_out = gr.JSON(label="Weakness Analysis")
+        coach_out = gr.JSON(label="Personalized Coaching")
+        pred_out = gr.JSON(label="Prediction Summary")
+
+        back_btn = gr.Button("← Back to Exam", visible=False)
+        retry_btn = gr.Button("Retry", visible=False)
+
+        # submit takes (answers_input, exam_state, subject_sel)
+        submit_btn.click(
+            submit_exam,
+            inputs=[answers_input, exam_state, subject_sel],
+            outputs=[score_out, analysis_out, coach_out, pred_out, back_btn, retry_btn]
+        )
+
+# Launch
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860)
 
-    if __name__ == "__main__":
-        demo.launch(server_name="0.0.0.0", server_port=7860)