import os import asyncio import re import json import shutil from sqlmodel import select from sqlmodel.ext.asyncio.session import AsyncSession from app.core.db import async_engine from app.models.models import Topic, Quiz, Question DATA_DIR = "./data" IMPORTED_DIR = os.path.join(DATA_DIR, "imported") def is_option_order_dependent(options): """ Checks if any option in the list contains order-dependent phrasing like 'A and B', 'All of the above', or '(A, B)'. """ pattern = re.compile(r'\b(above|all|none|both|a and|b and|c and|\(a, b\)|\(b, c\))\b', re.IGNORECASE) return any(pattern.search(opt) for opt in options) def js_to_json(js_str): """ Convert a JS object literal string to valid JSON by walking character by character to properly handle strings, unquoted keys, and JS literals. """ result = [] i = 0 length = len(js_str) while i < length: c = js_str[i] # Handle double-quoted strings (pass through, escape inner content for JSON) if c == '"': result.append('"') i += 1 while i < length and js_str[i] != '"': if js_str[i] == '\\': result.append(js_str[i]) i += 1 if i < length: result.append(js_str[i]) i += 1 else: result.append(js_str[i]) i += 1 if i < length: result.append('"') # closing quote i += 1 continue # Handle single-quoted strings → convert to double-quoted if c == "'": result.append('"') i += 1 while i < length and js_str[i] != "'": if js_str[i] == '\\': i += 1 if i < length: # If escaping a single quote, just output the quote without the backslash if js_str[i] == "'": result.append("'") else: result.append('\\') result.append(js_str[i]) i += 1 elif js_str[i] == '"': # Escape double quotes inside converted strings result.append('\\"') i += 1 else: result.append(js_str[i]) i += 1 if i < length: result.append('"') # closing quote i += 1 continue # Handle unquoted identifiers (keys or JS literals like true/false/null) if c.isalpha() or c == '_': word_start = i while i < length and (js_str[i].isalnum() or js_str[i] == '_'): i += 1 word = js_str[word_start:i] # Check what follows (skip whitespace) j = i while j < length and js_str[j] in ' \t': j += 1 if word == 'true': result.append('true') elif word == 'false': result.append('false') elif word == 'null': result.append('null') elif j < length and js_str[j] == ':': # This is an unquoted object key result.append(f'"{word}"') else: # Some other identifier — quote it just in case result.append(f'"{word}"') continue # Skip trailing commas before } or ] if c == ',': # Look ahead past whitespace for } or ] j = i + 1 while j < length and js_str[j] in ' \t\r\n': j += 1 if j < length and js_str[j] in '}]': i += 1 # skip the comma continue result.append(c) i += 1 return ''.join(result) def parse_js_file(filepath): """ Parses a JS file content to extract the quiz object. Strategy: 1. Read file. 2. Remove 'window.something = ' prefix and trailing semicolon. 3. Convert JS object literal to JSON using token-aware conversion. 4. Use json.loads to parse. """ with open(filepath, "r", encoding="utf-8") as f: content = f.read() # Strip assignment (e.g., window.apisBasicQuiz = { ... };) start_idx = content.find('{') end_idx = content.rfind('}') if start_idx == -1 or end_idx == -1: print(f"Error parsing {filepath}: No object found.") return None dict_content = content[start_idx:end_idx+1] json_str = js_to_json(dict_content) try: data = json.loads(json_str) return data except Exception as e: print(f"Failed to parse {filepath} with error: {e}") return None async def import_data(): if not os.path.exists(DATA_DIR): print(f"Data directory {DATA_DIR} does not exist.") return if not os.path.exists(IMPORTED_DIR): os.makedirs(IMPORTED_DIR) async with AsyncSession(async_engine) as session: files = [f for f in os.listdir(DATA_DIR) if f.endswith(".js")] for filename in files: filepath = os.path.join(DATA_DIR, filename) print(f"Processing {filename}...") data = parse_js_file(filepath) if not data: continue # Topic ID from filename (e.g. apis-basic.js -> apis) base_name = filename.removesuffix('.js') topic_id = base_name.split('-')[0] # Capitalize for name, or default to data if available (data usually has quiz name, not topic name) # We need to ensure Topic exists. # The data file structure: # { id: '...', name: '...', icon: '...', questions: [...], ... } # But the icon and generic name often belongs to the TOPIC if these are topic files, # OR they are quiz files. # Looking at apis.js: id: 'apis', name: 'APIs & Web Services', topic: 'APIs' # Looking at apis-basic.js: id: 'apis-basic-extended', name: 'APIs - Basic Extended' # Strategy: # 1. Upsert Topic. # If filename is just `topic.js` (like `apis.js`), maybe that is the "main" topic definition? # Let's assume `apis.js` defined the topic. # But `apis-basic.js` also needs the topic `apis` to exist. # Helper to get topic name if not known: topic_name = topic_id.capitalize() # If we find a "main" file (e.g. apis.js matching topic_id), use its name/icon for the Topic. # Otherwise use defaults. # Check if this file IS the topic definition file (filename == topic_id + ".js") is_topic_def_file = (base_name == topic_id) topic = await session.get(Topic, topic_id) if not topic: # Create default topic, it might be updated later if we find the def file topic = Topic(id=topic_id, name=topic_name, icon="fluent-color:question-circle-24") session.add(topic) await session.commit() # Commit to ensure existence for FKs print(f" Created Topic: {topic_id}") if is_topic_def_file: # Update topic details topic.name = data.get("name", topic_name) topic.icon = data.get("icon", "fluent-color:question-circle-24") session.add(topic) await session.commit() print(f" Updated Topic details from {filename}") # Upsert Quiz quiz_id = data.get("id") quiz_name = data.get("name") # Skip if no ID (shouldn't happen) if not quiz_id: print(f" Skipping {filename}: No quiz ID found.") continue quiz = await session.get(Quiz, quiz_id) if not quiz: quiz = Quiz(id=quiz_id, name=quiz_name, topic_id=topic_id) session.add(quiz) print(f" Created Quiz: {quiz_id}") else: quiz.name = quiz_name quiz.topic_id = topic_id session.add(quiz) print(f" Updated Quiz: {quiz_id}") # Commit quiz to ensure ID exists await session.commit() # Questions questions_data = data.get("questions", []) # Clear existing questions for this quiz to avoid duplication/staleness stmt = select(Question).where(Question.quiz_id == quiz_id) existing_questions = await session.exec(stmt) for q in existing_questions: await session.delete(q) for q_data in questions_data: options = q_data.get("choices", []) shufflable = not is_option_order_dependent(options) question = Question( content=q_data.get("question"), options=options, answer=q_data.get("answer"), difficulty=q_data.get("difficulty"), sub_topic=q_data.get("subTopic"), is_shufflable=shufflable, quiz_id=quiz_id ) session.add(question) await session.commit() print(f" Imported {len(questions_data)} questions for {quiz_id}") # Move the file to the imported directory shutil.move(filepath, os.path.join(IMPORTED_DIR, filename)) print(f" Moved {filename} to {IMPORTED_DIR}") print("Import completed.") if __name__ == "__main__": asyncio.run(import_data())