""" File-based storage utilities for exam system. """ import json import os from datetime import datetime from pathlib import Path from typing import Dict, List, Optional from django.conf import settings class StorageManager: """Manage file-based storage for exams, attempts, and progress.""" def __init__(self): self.input_dir = settings.INPUT_DIR self.attempts_dir = settings.ATTEMPTS_DIR self.output_dir = settings.OUTPUT_DIR self.progress_dir = settings.PROGRESS_DIR self.manifest_file = settings.MANIFEST_FILE def _read_json(self, path: Path) -> Dict: """Read JSON file safely.""" if not path.exists(): return {} with open(path, 'r', encoding='utf-8') as f: return json.load(f) def _write_json(self, path: Path, data: Dict): """Write JSON file atomically.""" path.parent.mkdir(parents=True, exist_ok=True) temp_path = path.with_suffix('.tmp') with open(temp_path, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) temp_path.replace(path) # Manifest operations def get_manifest(self) -> Dict: """Load manifest.json.""" if not self.manifest_file.exists(): manifest = {"version": "1.0.0", "exams": [], "users": {}} self._write_json(self.manifest_file, manifest) return manifest return self._read_json(self.manifest_file) def save_manifest(self, manifest: Dict): """Save manifest.json.""" self._write_json(self.manifest_file, manifest) # Exam operations def list_exams(self, published_only: bool = True) -> List[Dict]: """List all exams from manifest.""" manifest = self.get_manifest() exams = manifest.get('exams', []) if published_only: exams = [e for e in exams if e.get('published', False)] return exams def get_exam(self, exam_id: str) -> Optional[Dict]: """Load exam JSON by ID.""" manifest = self.get_manifest() exam_info = next((e for e in manifest.get('exams', []) if e['examId'] == exam_id), None) if not exam_info: return None # Support both old flat structure and new hierarchical structure exam_path = self.input_dir / exam_info.get('path', f"{exam_id}.json") if not exam_path.exists(): # Fallback to flat structure exam_path = self.input_dir / f"{exam_id}.json" if not exam_path.exists(): return None return self._read_json(exam_path) def is_exam_finished(self, user_id: str, exam_id: str, allow_retake: bool = False) -> bool: """Check if user has finished exam.""" if allow_retake: return False manifest = self.get_manifest() users = manifest.get('users', {}) user_data = users.get(user_id, {}) return exam_id in user_data.get('finished', []) # Attempt operations def get_attempt_path(self, user_id: str, exam_id: str, attempt_id: str, subject: str = None, month: str = None) -> Path: """Get path for attempt file with hierarchical structure.""" if subject and month: # New hierarchical structure return self.attempts_dir / user_id / subject / month / exam_id / f"{attempt_id}.json" else: # Old flat structure (backward compatibility) return self.attempts_dir / user_id / exam_id / f"{attempt_id}.json" def get_active_attempt(self, user_id: str, exam_id: str) -> Optional[Dict]: """Get active attempt for user/exam.""" attempt_dir = self.attempts_dir / user_id / exam_id if not attempt_dir.exists(): return None # Find most recent in_progress attempt attempts = [] for attempt_file in attempt_dir.glob('*.json'): attempt_data = self._read_json(attempt_file) if attempt_data.get('status') == 'in_progress': attempts.append(attempt_data) if not attempts: return None # Return most recent attempts.sort(key=lambda x: x.get('updatedAt', ''), reverse=True) return attempts[0] def create_attempt(self, user_id: str, exam_id: str, exam: Dict) -> Dict: """Create new attempt with hierarchical structure.""" timestamp = datetime.utcnow() timestamp_str = timestamp.isoformat() + 'Z' month_str = timestamp.strftime('%Y-%m') attempt_id = f"attempt-{timestamp.strftime('%Y%m%d-%H%M%S')}" # Get subject and month from exam subject = exam.get('subject', 'unknown') attempt = { "attemptId": attempt_id, "userId": user_id, "examId": exam_id, "subject": subject, "month": month_str, "examVersion": exam.get('metadata', {}).get('version', '1.0.0'), "status": "in_progress", "startedAt": timestamp_str, "updatedAt": timestamp_str, "answers": [] } attempt_path = self.get_attempt_path(user_id, exam_id, attempt_id, subject, month_str) self._write_json(attempt_path, attempt) # Update manifest manifest = self.get_manifest() users = manifest.setdefault('users', {}) user_data = users.setdefault(user_id, {'active': [], 'finished': []}) if exam_id not in user_data['active']: user_data['active'].append(exam_id) self.save_manifest(manifest) return attempt def save_attempt(self, attempt: Dict): """Save attempt (autosave) with hierarchical structure.""" user_id = attempt['userId'] exam_id = attempt['examId'] attempt_id = attempt['attemptId'] subject = attempt.get('subject') month = attempt.get('month') attempt['updatedAt'] = datetime.utcnow().isoformat() + 'Z' attempt_path = self.get_attempt_path(user_id, exam_id, attempt_id, subject, month) self._write_json(attempt_path, attempt) def submit_attempt(self, attempt: Dict, exam: Dict) -> str: """Submit attempt and create output bundle with hierarchical structure.""" attempt['status'] = 'submitted' attempt['submittedAt'] = datetime.utcnow().isoformat() + 'Z' self.save_attempt(attempt) # Create output bundle bundle = { "exam": exam, "attempt": attempt } # Get username from user_id from django.contrib.auth import get_user_model User = get_user_model() try: if attempt['userId'].isdigit(): user = User.objects.get(id=int(attempt['userId'])) username = user.username else: username = attempt['userId'] # guest users except: username = attempt['userId'] # Organize by username/subject/month subject = attempt.get('subject', exam.get('subject', 'unknown')) month = attempt.get('month', datetime.utcnow().strftime('%Y-%m')) output_path = self.output_dir / username / subject / month / f"{exam['examId']}_{attempt['attemptId']}.json" self._write_json(output_path, bundle) # Mark as finished attempt['status'] = 'finished' self.save_attempt(attempt) # Update manifest manifest = self.get_manifest() users = manifest.setdefault('users', {}) user_data = users.setdefault(attempt['userId'], {'active': [], 'finished': []}) if exam['examId'] in user_data['active']: user_data['active'].remove(exam['examId']) if exam['examId'] not in user_data['finished']: user_data['finished'].append(exam['examId']) self.save_manifest(manifest) return str(output_path) # Progress operations def get_progress(self, user_id: str) -> Dict: """Get user progress.""" progress_path = self.progress_dir / f"{user_id}.json" if not progress_path.exists(): return {"userId": user_id, "exams": {}} return self._read_json(progress_path) def update_progress(self, user_id: str, exam_id: str, percent: float): """Update progress for exam.""" progress = self.get_progress(user_id) progress['exams'][exam_id] = { "percentComplete": percent, "lastSavedAt": datetime.utcnow().isoformat() + 'Z' } progress_path = self.progress_dir / f"{user_id}.json" self._write_json(progress_path, progress) # Global storage instance storage = StorageManager()