Create loader.py

loader.py

@@ -0,0 +1,163 @@
+# loader.py
+import os
+import json
+from typing import List, Dict, Any
+
+import qa_store
+
+CURRICULUM_PATH = "data/1_Year_U_1.jsonl"
+MANUAL_QA_PATH = "data/manual_qa.jsonl"
+
+
+def load_curriculum() -> None:
+    """
+    Load official textbook JSONL into qa_store.ENTRIES and AUTO_QA_KNOWLEDGE.
+    """
+    qa_store.ENTRIES.clear()
+    qa_store.AUTO_QA_KNOWLEDGE.clear()
+
+    if not os.path.exists(CURRICULUM_PATH):
+        qa_store.RAW_KNOWLEDGE = "ຍັງບໍ່ມີຂໍ້ມູນປະຫວັດສາດຖືກໂຫຼດ."
+        return
+
+    with open(CURRICULUM_PATH, "r", encoding="utf-8") as f:
+        for line in f:
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                obj: Dict[str, Any] = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+
+            if "text" not in obj:
+                continue
+
+            qa_store.ENTRIES.append(obj)
+
+            for pair in obj.get("qa", []):
+                q = (pair.get("q") or "").strip()
+                a = (pair.get("a") or "").strip()
+                if not q or not a:
+                    continue
+                norm_q = qa_store.normalize_question(q)
+                qa_store.AUTO_QA_KNOWLEDGE.append(
+                    {
+                        "norm_q": norm_q,
+                        "q": q,
+                        "a": a,
+                        "source": "auto",
+                        "id": obj.get("id", ""),
+                    }
+                )
+
+    if qa_store.ENTRIES:
+        qa_store.RAW_KNOWLEDGE = "\n\n".join(e["text"] for e in qa_store.ENTRIES)
+    else:
+        qa_store.RAW_KNOWLEDGE = "ຍັງບໍ່ມີຂໍ້ມູນປະຫວັດສາດທີ່ອ່ານໄດ້."
+
+
+def load_manual_qa() -> None:
+    """
+    Load manual_qa.jsonl into qa_store.MANUAL_QA_LIST and MANUAL_QA_INDEX.
+    """
+    qa_store.MANUAL_QA_LIST.clear()
+    qa_store.MANUAL_QA_INDEX.clear()
+    max_num = 0
+
+    if not os.path.exists(MANUAL_QA_PATH):
+        qa_store.NEXT_MANUAL_ID = 1
+        return
+
+    with open(MANUAL_QA_PATH, "r", encoding="utf-8") as f:
+        for line in f:
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                obj = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+
+            q = (obj.get("q") or "").strip()
+            a = (obj.get("a") or "").strip()
+            if not q or not a:
+                continue
+
+            entry_id = str(obj.get("id") or "")
+            if not entry_id:
+                max_num += 1
+                entry_id = f"manual_{max_num:04d}"
+
+            # track biggest number in id
+            import re as _re
+
+            m = _re.search(r"(\d+)$", entry_id)
+            if m:
+                max_num = max(max_num, int(m.group(1)))
+
+            norm_q = qa_store.normalize_question(q)
+            entry = {
+                "id": entry_id,
+                "q": q,
+                "a": a,
+                "norm_q": norm_q,
+            }
+            qa_store.MANUAL_QA_LIST.append(entry)
+            qa_store.MANUAL_QA_INDEX[norm_q] = entry
+
+    qa_store.NEXT_MANUAL_ID = max_num + 1 if max_num > 0 else 1
+
+
+def generate_new_manual_id() -> str:
+    """
+    Generate a new manual entry ID and increment the counter.
+    """
+    new_id = f"manual_{qa_store.NEXT_MANUAL_ID:04d}"
+    qa_store.NEXT_MANUAL_ID += 1
+    return new_id
+
+
+def save_manual_qa_file() -> None:
+    """
+    Persist MANUAL_QA_LIST to data/manual_qa.jsonl.
+    """
+    os.makedirs(os.path.dirname(MANUAL_QA_PATH), exist_ok=True)
+    with open(MANUAL_QA_PATH, "w", encoding="utf-8") as f:
+        for e in qa_store.MANUAL_QA_LIST:
+            obj = {"id": e["id"], "q": e["q"], "a": e["a"]}
+            f.write(json.dumps(obj, ensure_ascii=False) + "\n")
+
+
+def rebuild_combined_qa() -> None:
+    """
+    Combine auto and manual QA into QA_INDEX & ALL_QA_KNOWLEDGE.
+    Manual answers override auto ones if same normalized question.
+    """
+    qa_store.QA_INDEX.clear()
+    qa_store.ALL_QA_KNOWLEDGE.clear()
+
+    # auto first
+    for item in qa_store.AUTO_QA_KNOWLEDGE:
+        norm_q = item["norm_q"]
+        qa_store.QA_INDEX[norm_q] = item["a"]
+        qa_store.ALL_QA_KNOWLEDGE.append(item)
+
+    # manual overrides
+    for e in qa_store.MANUAL_QA_LIST:
+        item = {
+            "norm_q": e["norm_q"],
+            "q": e["q"],
+            "a": e["a"],
+            "source": "manual",
+            "id": e["id"],
+        }
+        qa_store.QA_INDEX[item["norm_q"]] = item["a"]
+        qa_store.ALL_QA_KNOWLEDGE.append(item)
+
+
+def manual_qa_table_data() -> List[List[str]]:
+    """
+    Table rows for Teacher Panel.
+    """
+    return [[e["id"], e["q"], e["a"]] for e in qa_store.MANUAL_QA_LIST]