236 lines
8.7 KiB
Python
236 lines
8.7 KiB
Python
"""
|
|
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()
|
|
|