Add automated Moodle question import script

moodlle upload script/moodle_config.md

@@ -0,0 +1,135 @@
+#  How to Configure MOODLE_CMID and MOODLE_PARENT_CATEGORY
+
+This script requires two Moodle-specific values:
+
+* `MOODLE_CMID`
+* `MOODLE_PARENT_CATEGORY`
+
+These values are tied to a specific course and its question bank.
+
+
+##  Example .env structure
+
+```
+BASE_URL=https://ilw.onlinetest.spoken-tutorial.org
+MOODLE_WS_TOKEN=
+MOODLE_CMID=192
+MOODLE_PARENT_CATEGORY=1359,89025
+QUESTIONBANK_DIR=ST-linux-questionbank
+MOODLE_USERNAME=
+MOODLE_PASSWORD=
+```
+
+
+
+---
+
+##  1. Getting `MOODLE_CMID`
+
+### What it is:
+
+The **Course Module ID (cmid)** for the question bank.
+
+### How to find it:
+
+1. Open your Moodle course
+
+2. Go to:
+
+   ```
+   Question bank → Questions
+   ```
+
+3. Look at the URL in your browser:
+
+   Example:
+
+   ```
+   https://ilw.onlinetest.spoken-tutorial.org/question/edit.php?cmid=192
+   ```
+
+4. Extract:
+
+   ```
+   cmid=192
+   ```
+
+### Set in `.env`:
+
+```
+MOODLE_CMID=192
+```
+
+---
+
+##  2. Getting `MOODLE_PARENT_CATEGORY`
+
+### What it is:
+
+The **target question category** where questions will be imported.
+
+It has TWO parts:
+
+```
+category_id,context_id
+```
+
+Example:
+
+```
+1359,89025
+```
+
+---
+
+### How to find it:
+
+1. Go to:
+
+   ```
+   Question bank → Questions
+   ```
+
+2. In the category dropdown, select the desired category
+   (e.g., "Default for DC101")
+
+3. Look at the URL:
+
+   Example:
+
+   ```
+   https://ilw.onlinetest.spoken-tutorial.org/question/edit.php?cmid=192&cat=1359%2C89025&qpage=0&category=1359%2C89025
+   ```
+
+4. Extract:
+
+   ```
+   cat=1359%2C89025
+   ```
+
+5. Decode `%2C` → `,`
+
+   Final value:
+
+   ```
+   1359,89025
+   ```
+
+---
+
+### Set in `.env`:
+
+```
+MOODLE_PARENT_CATEGORY=1359,89025
+```
+
+---
+
+##  Important Notes
+
+* The `context_id` (second number) is critical
+  -> If incorrect, categories may not appear or imports may go to the wrong place.
+
+* If using a different course, these values will change.
+
+---

moodlle upload script/upload_questions.py

@@ -0,0 +1,234 @@
+import os, json, logging, requests, re, argparse, glob, time
+from xml.sax.saxutils import escape
+from dotenv import load_dotenv
+
+load_dotenv()
+BASE_URL = os.getenv("BASE_URL")
+MOODLE_USERNAME = os.getenv("MOODLE_USERNAME")
+MOODLE_PASSWORD = os.getenv("MOODLE_PASSWORD")
+MOODLE_PARENT_CATEGORY = os.getenv("MOODLE_PARENT_CATEGORY")
+MOODLE_CMID = os.getenv("MOODLE_CMID")
+QUESTIONBANK_DIR = os.getenv("QUESTIONBANK_DIR")
+
+# Regex Patterns 
+RE_SESSKEY = r'(?:name=["\']sesskey["\'][^>]*value=["\']|"sesskey"\s*:\s*")([^"\']+)'
+RE_LOGIN_TOKEN = r'name=["\']logintoken["\'][^>]*value=["\']([^"\']*)'
+RE_CAT_OPTION = r'<option\s+value="(\d+,\d+)"[^>]*>\s*([^<]+)'
+RE_CAT_CLEAN = r'(&nbsp;|\s*\(\d+\)\s*$)'
+RE_DRAFT_ITEMID = r'name=["\']newfile["\'][^>]*value=["\'](\d+)|value=["\'](\d+)["\'][^>]*name=["\']newfile["\']'
+
+logging.basicConfig(level=logging.INFO, format="%(levelname)s | %(message)s")
+session = requests.Session()
+session.headers.update({"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"})
+sesskey = None
+
+def ext(html, regex, default=None):
+    m = re.search(regex, html)
+    return next((g for g in m.groups() if g), default) if m else default
+
+
+def log(lvl, event, **f):
+    getattr(logging, lvl)(json.dumps({"event": event, **f}))
+
+
+def moodle_login():
+    # Login flow for Moodle UI automation
+    global sesskey
+    if not (MOODLE_USERNAME and MOODLE_PASSWORD): return log("error", "login_failed", reason="Missing credentials") or False
+    try:
+        url = f"{BASE_URL}/login/index.php"
+        r = session.get(url, timeout=30)
+
+        # Extract Moodle login token from HTML
+        token = ext(r.text, RE_LOGIN_TOKEN)
+        if not token: return log("error", "login_failed", reason="No logintoken found") or False
+        session.headers["Referer"] = url
+
+        # Submit login credentials
+        r = session.post(url, data={"username": MOODLE_USERNAME, "password": MOODLE_PASSWORD, "logintoken": token}, timeout=30)
+
+        # Extract Moodle sesskey from HTML/JSON login response
+        sesskey = ext(r.text, RE_SESSKEY)
+        if sesskey: return log("info", "login_success") or True
+        return log("error", "login_failed", reason="No sesskey after login") or False
+    except Exception as e: return log("error", "login_failed", reason=str(e)) or False
+
+
+def find_or_create_category(name):
+    global sesskey
+    url = f"{BASE_URL}/question/category.php?cmid={MOODLE_CMID}"
+    try:
+        r = session.get(url, timeout=30)
+        sesskey = ext(r.text, RE_SESSKEY, sesskey)
+        cats = {re.sub(RE_CAT_CLEAN, '', m.group(2)).strip().lower(): m.group(1) 
+                for m in re.finditer(RE_CAT_OPTION, r.text)}
+
+        # Try exact match 
+        if name.lower() in cats: return log("info", "category_found", name=name, category_id=cats[name.lower()]) or cats[name.lower()]
+
+        # Try partial match (if already exists but named slightly differently)
+        for k in cats:
+            if name.lower() in k: return log("info", "category_found_partial", name=name, matched=k, category_id=cats[k]) or cats[k]
+
+        # Create category
+        data = {
+            "cmid": MOODLE_CMID,
+            "sesskey": sesskey,
+            "_qf__question_category_edit_form": "1",
+            "mform_isexpanded_id_categoryheader": "1",
+            "name": name,
+            "parent": MOODLE_PARENT_CATEGORY,
+            "info[text]": "",
+            "info[format]": "1",
+            "id": "0",
+            "submitbutton": "Add category"
+        }
+        r = session.post(url, data=data, timeout=30)
+
+        # Re-fetch page to see new category
+        r = session.get(url, timeout=30)
+        sesskey = ext(r.text, RE_SESSKEY, sesskey)
+        cats = {re.sub(RE_CAT_CLEAN, '', m.group(2)).strip().lower(): m.group(1) 
+                for m in re.finditer(RE_CAT_OPTION, r.text)}
+
+        # Try matching
+        if name.lower() in cats: return log("info", "category_created", name=name, category_id=cats[name.lower()]) or cats[name.lower()]
+
+        for k in cats:
+            if name.lower() in k: return log("info", "category_created_partial", name=name, matched=k, category_id=cats[k]) or cats[k]
+
+        # Log failure 
+        available = list(cats.keys())[:5]
+        return log("error", "category_create_failed", name=name, reason="Not found after create", available_examples=available) or None
+    except Exception as e: return log("error", "category_create_failed", name=name, reason=str(e)) or None
+
+
+def parse_filename(filepath):
+    m = re.match(r'^(\d+)[-_](.+?)[-_](easy|medium)(?:\d+)?$', os.path.splitext(os.path.basename(filepath))[0], re.IGNORECASE)
+    if not m: return None, None
+    raw = re.sub(r'([a-z])([A-Z])', r'\1 \2', m.group(2).replace('-', ' ').replace('_', ' '))
+    return ' '.join(w.capitalize() for w in raw.split()).strip(), m.group(3).lower()
+
+
+def convert_aiken_to_xml(text, mark, filename):
+    # AIKEN to Moodle XML conversion 
+    questions, lines = [], [l.strip() for l in text.splitlines() if l.strip()]
+    cur_q, choices = None, {}
+    for l in lines:
+        am = re.match(r'^ANSWER:\s*([A-Z])$', l, re.I)
+        if am:
+            if cur_q and choices: questions.append({"text": cur_q, "choices": choices, "correct": am.group(1).upper()})
+            cur_q, choices = None, {}
+            continue
+        cm = re.match(r'^([A-Z])[.\)]\s*(.+)$', l, re.I)
+        if cm: choices[cm.group(1).upper()] = cm.group(2).strip(); continue
+        cur_q = l if cur_q is None else f"{cur_q} {l}"
+
+    xml = ['<?xml version="1.0" encoding="UTF-8"?>', '<quiz>']
+
+    # Iterate questions with counter for naming
+    for i, q in enumerate(questions, 1):
+        q_name = escape(f"{filename}_{i:04d}")
+        xml += ['  <question type="multichoice">', f'    <name><text>{q_name}</text></name>', '    <questiontext format="html">', f'      <text><![CDATA[{q["text"]}]]></text>', '    </questiontext>', f'    <defaultgrade>{mark}</defaultgrade>', '    <shuffleanswers>1</shuffleanswers>', '    <single>true</single>', '    <answernumbering>abc</answernumbering>']
+        for l, t in q["choices"].items():
+            xml += [f'    <answer fraction="{"100" if l == q["correct"] else "0"}" format="html">', f'      <text><![CDATA[{t}]]></text>', '    </answer>']
+        xml += ['  </question>']
+    return "\n".join(xml + ['</quiz>'])
+
+
+def upload_and_import(path, cat_id, diff, retries=3):
+    global sesskey
+    filename_base = os.path.splitext(os.path.basename(path))[0]
+    xml_name = filename_base + ".xml"
+    tmp_path = os.path.join(os.path.dirname(path), xml_name)
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            with open(tmp_path, "w", encoding="utf-8") as tf:
+
+
+                tf.write(convert_aiken_to_xml(f.read(), {"easy": "1.0", "medium": "2.0"}.get(diff, "1.0"), filename_base))
+    except Exception as e: return log("error", "xml_conversion_failed", file=path, error=str(e)) or False
+    for attempt in range(1, retries + 1):
+        try:
+            r = session.get(f"{BASE_URL}/question/import.php?cmid={MOODLE_CMID}", timeout=30)
+            sesskey = ext(r.text, RE_SESSKEY, sesskey)
+
+            # Extract Moodle draft itemid for file upload
+            draft = ext(r.text, RE_DRAFT_ITEMID)
+            if not draft: 
+                log("warning", "draft_itemid_not_found", attempt=attempt)
+                time.sleep(2**(attempt-1))
+                continue
+            repo = ext(r.text, r'"(\d+)":\{"id":"\1","name":"[^"]*","type":"upload"', "5")
+            ctx = ext(r.text, r'"contextid"\s*:\s*(\d+)', MOODLE_CMID)
+            log("info", "draft_area_obtained", draft_itemid=draft)
+            with open(tmp_path, "rb") as f:
+
+                # Upload file to Moodle draft area
+                up_r = session.post(f"{BASE_URL}/repository/repository_ajax.php?action=upload", data={"sesskey": sesskey, "repo_id": repo, "itemid": draft, "author": MOODLE_USERNAME, "savepath": "/", "title": xml_name, "ctx_id": ctx, "overwrite": "1", "env": "filepicker"}, files={"repo_upload_file": (xml_name, f, "text/xml")}, timeout=30)
+            uj = up_r.json() if "application/json" in up_r.headers.get("Content-Type", "") else {}
+            if uj.get("error"): 
+                log("warning", "draft_upload_error", attempt=attempt, error=uj["error"])
+                time.sleep(2**(attempt-1)); continue
+            log("info", "file_uploaded_to_draft", file=xml_name, draft_itemid=draft)
+
+            # Submit import form with uploaded file
+            r = session.post(f"{BASE_URL}/question/import.php?cmid={MOODLE_CMID}", data={"sesskey": sesskey, "cmid": MOODLE_CMID, "_qf__question_import_form": "1", "cat": cat_id, "category": cat_id, "format": "xml", "newfile": draft, "submitbutton": "Import", "stoponerror": "0", "matchgrades": "error"}, timeout=60)
+            text = r.text.lower()
+            if any(k in text for k in ("successful", "imported", "parsing")):
+                log("info", "import_success", attempt=attempt)
+                try: os.remove(tmp_path)
+                except: pass
+                return True
+            log("warning", "import_response_unexpected", attempt=attempt, preview=r.text[:800])
+            if any(k in text for k in ("invalid sesskey", "session has timed out", "access denied")): return False
+        except Exception as e: log("warning", "import_request_failed", attempt=attempt, error=str(e))
+        time.sleep(2**(attempt-1))
+    try: os.remove(tmp_path)
+    except: pass
+    return False
+
+
+def process_file(path):
+    fname = os.path.basename(path)
+    log("info", "file_processing_started", file=fname)
+    tut, diff = parse_filename(path)
+    if not tut: return log("error", "invalid_filename_format", file=fname) or (False, 0)
+
+    cat_id = find_or_create_category(f"{tut} ({diff.capitalize()})")
+    if not cat_id:
+        log("warning", "using_fallback_category", file=fname, fallback=MOODLE_PARENT_CATEGORY)
+        cat_id = MOODLE_PARENT_CATEGORY
+
+    if not upload_and_import(path, cat_id, diff): return False, 0
+    try:
+        with open(path, 'r', encoding='utf-8') as f: qc = sum(1 for l in f if "ANSWER:" in l)
+    except: qc = 0
+    log("info", "file_processing_completed", file=fname, questions_count=qc)
+    return True, qc
+
+
+def main():
+    p = argparse.ArgumentParser()
+    p.add_argument("--file")
+    args = p.parse_args()
+    req = {"BASE_URL": BASE_URL, "MOODLE_USERNAME": MOODLE_USERNAME, "MOODLE_PASSWORD": MOODLE_PASSWORD, "MOODLE_PARENT_CATEGORY": MOODLE_PARENT_CATEGORY, "QUESTIONBANK_DIR": QUESTIONBANK_DIR}
+    missing = [k for k, v in req.items() if not v]
+    if missing: raise SystemExit(f"Missing: {', '.join(missing)}")
+    log("info", "batch_started")
+    log("info", "login_started")
+    if not moodle_login(): raise SystemExit("Login failed.")
+    files = [args.file] if args.file else sorted(glob.glob(os.path.join(QUESTIONBANK_DIR, "*.txt")))
+    totals = {"proc": 0, "q": 0, "fail": 0}
+    for i, f in enumerate(files):
+        if not os.path.isfile(f): log("error", "file_not_found", file=f); totals["fail"] += 1; continue
+        try:
+            ok, qc = process_file(f)
+            if ok: totals["proc"] += 1; totals["q"] += qc
+            else: totals["fail"] += 1
+        except Exception as e: log("error", "process_file_exception", file=f, error=str(e)); totals["fail"] += 1
+        if i < len(files) - 1: time.sleep(0.5)
+    log("info", "batch_completed")
+    print(f"\n--- Output Summary ---\ntotal_files_processed: {totals['proc']}\ntotal_questions_uploaded: {totals['q']}\nfailed_files: {totals['fail']}")
+
+if __name__ == "__main__": main()