From 4cac65115b61d33a655a3924405a7576ddbcae42 Mon Sep 17 00:00:00 2001 From: Naman Sharma Date: Wed, 18 Mar 2026 11:46:21 +0000 Subject: [PATCH] Add Moodle questions upload script --- moodlle upload script/moodle_config.md | 135 +++++++++++++ moodlle upload script/upload_questions.py | 234 ++++++++++++++++++++++ 2 files changed, 369 insertions(+) create mode 100644 moodlle upload script/moodle_config.md create mode 100644 moodlle upload script/upload_questions.py diff --git a/moodlle upload script/moodle_config.md b/moodlle upload script/moodle_config.md new file mode 100644 index 00000000..9d9d3438 --- /dev/null +++ b/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. + +--- diff --git a/moodlle upload script/upload_questions.py b/moodlle upload script/upload_questions.py new file mode 100644 index 00000000..209d475f --- /dev/null +++ b/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']*>\s*([^<]+)' +RE_CAT_CLEAN = r'( |\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 = ['', ''] + + # Iterate questions with counter for naming + for i, q in enumerate(questions, 1): + q_name = escape(f"{filename}_{i:04d}") + xml += [' ', f' {q_name}', ' ', f' ', ' ', f' {mark}', ' 1', ' true', ' abc'] + for l, t in q["choices"].items(): + xml += [f' ', f' ', ' '] + xml += [' '] + return "\n".join(xml + ['']) + + +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() \ No newline at end of file