772 lines
26 KiB
Python
772 lines
26 KiB
Python
from sqlalchemy.orm import Session, joinedload
|
|
from sqlalchemy import func, desc
|
|
from typing import Optional
|
|
from datetime import datetime
|
|
from . import models, schemas, auth
|
|
import markdown
|
|
from markdown.extensions.tables import TableExtension
|
|
from markdown.extensions.fenced_code import FencedCodeExtension
|
|
from markdown.extensions.toc import TocExtension
|
|
|
|
def render_markdown_to_html(md_content: str) -> str:
|
|
if not md_content:
|
|
return ""
|
|
extensions = [
|
|
TableExtension(),
|
|
FencedCodeExtension(),
|
|
TocExtension(),
|
|
'nl2br',
|
|
'sane_lists',
|
|
'smarty'
|
|
]
|
|
return markdown.markdown(md_content, extensions=extensions)
|
|
|
|
# User operations
|
|
def get_user_by_email(db: Session, email: str):
|
|
return db.query(models.User).filter(models.User.email == email).first()
|
|
|
|
def create_user(db: Session, user: schemas.UserCreate):
|
|
hashed_password = auth.get_password_hash(user.password)
|
|
db_user = models.User(
|
|
email=user.email,
|
|
hashed_password=hashed_password
|
|
)
|
|
db.add(db_user)
|
|
db.commit()
|
|
db.refresh(db_user)
|
|
return db_user
|
|
|
|
def authenticate_user(db: Session, email: str, password: str):
|
|
user = get_user_by_email(db, email)
|
|
if not user:
|
|
return None
|
|
if not auth.verify_password(password, user.hashed_password):
|
|
return None
|
|
return user
|
|
|
|
# Course operations
|
|
def get_courses(db: Session, skip: int = 0, limit: int = 100, level: Optional[str] = None, author_id: Optional[int] = None, published_only: bool = True):
|
|
query = db.query(models.Course)
|
|
|
|
if level:
|
|
query = query.filter(models.Course.level == level)
|
|
|
|
if author_id:
|
|
query = query.filter(models.Course.author_id == author_id)
|
|
|
|
if published_only:
|
|
query = query.filter(models.Course.status == 'published')
|
|
|
|
return query.order_by(models.Course.created_at.desc()).offset(skip).limit(limit).all()
|
|
|
|
def get_course_by_id(db: Session, course_id: int, include_structure: bool = False):
|
|
course = db.query(models.Course).filter(models.Course.id == course_id).first()
|
|
|
|
if course and include_structure:
|
|
# Eager load all related data
|
|
course.modules # This will trigger lazy loading
|
|
for module in course.modules:
|
|
module.lessons
|
|
for lesson in module.lessons:
|
|
lesson.steps
|
|
|
|
return course
|
|
|
|
def create_course(db: Session, course: schemas.CourseCreate, author_id: int):
|
|
db_course = models.Course(
|
|
title=course.title,
|
|
description=course.description,
|
|
level=course.level,
|
|
duration=course.duration,
|
|
author_id=author_id
|
|
)
|
|
db.add(db_course)
|
|
db.commit()
|
|
db.refresh(db_course)
|
|
return db_course
|
|
|
|
def update_course(db: Session, course_id: int, course_update: schemas.CourseUpdate):
|
|
db_course = get_course_by_id(db, course_id)
|
|
if not db_course:
|
|
return None
|
|
|
|
update_data = course_update.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(db_course, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_course)
|
|
return db_course
|
|
|
|
def delete_course(db: Session, course_id: int):
|
|
db_course = get_course_by_id(db, course_id)
|
|
if not db_course:
|
|
return False
|
|
|
|
db.delete(db_course)
|
|
db.commit()
|
|
return True
|
|
|
|
def create_course_with_content(db: Session, course_data: schemas.CourseCreateRequest, author_id: int):
|
|
"""
|
|
Create a course with content (markdown) and structure (modules, lessons, steps)
|
|
Always creates draft course with real structure in DB
|
|
"""
|
|
# Create basic course with status='draft'
|
|
db_course = models.Course(
|
|
title=course_data.title,
|
|
description=course_data.description,
|
|
level=course_data.level,
|
|
duration=course_data.duration,
|
|
content=course_data.content,
|
|
author_id=author_id,
|
|
status='draft',
|
|
)
|
|
db.add(db_course)
|
|
db.commit()
|
|
db.refresh(db_course)
|
|
|
|
# Extract structure from course_data if provided
|
|
structure = course_data.structure or {}
|
|
modules_data = structure.get('modules', [])
|
|
|
|
# If no structure provided, create default module, lesson and step with content
|
|
if not modules_data:
|
|
# Create default module
|
|
default_module = models.CourseModule(
|
|
course_id=db_course.id,
|
|
title="Введение",
|
|
description="Основное содержание курса",
|
|
order_index=1
|
|
)
|
|
db.add(default_module)
|
|
db.commit()
|
|
db.refresh(default_module)
|
|
|
|
# Create default lesson
|
|
lesson = models.Lesson(
|
|
module_id=default_module.id,
|
|
title="Общее описание",
|
|
order_index=1
|
|
)
|
|
db.add(lesson)
|
|
db.commit()
|
|
db.refresh(lesson)
|
|
|
|
# Create step with content
|
|
step = models.LessonStep(
|
|
lesson_id=lesson.id,
|
|
step_type="theory",
|
|
title=course_data.title,
|
|
content=course_data.content,
|
|
order_index=1
|
|
)
|
|
db.add(step)
|
|
db.commit()
|
|
else:
|
|
# Create modules, lessons and steps from structure
|
|
for module_index, module_data in enumerate(modules_data):
|
|
# Create module
|
|
module = models.CourseModule(
|
|
course_id=db_course.id,
|
|
title=module_data.get('title', f"Модуль {module_index + 1}"),
|
|
description=module_data.get('description', ''),
|
|
duration=module_data.get('duration', ''),
|
|
order_index=module_index + 1
|
|
)
|
|
db.add(module)
|
|
db.commit()
|
|
db.refresh(module)
|
|
|
|
# Create lessons in module
|
|
lessons_data = module_data.get('lessons', [])
|
|
for lesson_index, lesson_data in enumerate(lessons_data):
|
|
lesson = models.Lesson(
|
|
module_id=module.id,
|
|
title=lesson_data.get('title', f"Урок {lesson_index + 1}"),
|
|
description=lesson_data.get('description', ''),
|
|
order_index=lesson_index + 1
|
|
)
|
|
db.add(lesson)
|
|
db.commit()
|
|
db.refresh(lesson)
|
|
|
|
# Create steps in lesson
|
|
steps_data = lesson_data.get('steps', [])
|
|
for step_index, step_data in enumerate(steps_data):
|
|
step = models.LessonStep(
|
|
lesson_id=lesson.id,
|
|
step_type=step_data.get('step_type', 'theory'),
|
|
title=step_data.get('title', f"Шаг {step_index + 1}"),
|
|
content=step_data.get('content', ''),
|
|
order_index=step_index + 1
|
|
)
|
|
db.add(step)
|
|
db.commit()
|
|
|
|
return db_course
|
|
|
|
# Module operations
|
|
def get_module_by_id(db: Session, module_id: int):
|
|
return db.query(models.CourseModule).filter(models.CourseModule.id == module_id).first()
|
|
|
|
def get_course_modules(db: Session, course_id: int):
|
|
return db.query(models.CourseModule).filter(
|
|
models.CourseModule.course_id == course_id
|
|
).order_by(models.CourseModule.order_index).all()
|
|
|
|
def create_module(db: Session, module: schemas.ModuleCreate):
|
|
db_module = models.CourseModule(
|
|
course_id=module.course_id,
|
|
title=module.title,
|
|
description=module.description,
|
|
duration=module.duration,
|
|
order_index=module.order_index
|
|
)
|
|
db.add(db_module)
|
|
db.commit()
|
|
db.refresh(db_module)
|
|
return db_module
|
|
|
|
def update_module(db: Session, module_id: int, module_update: schemas.ModuleUpdate):
|
|
db_module = get_module_by_id(db, module_id)
|
|
if not db_module:
|
|
return None
|
|
|
|
update_data = module_update.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(db_module, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_module)
|
|
return db_module
|
|
|
|
def delete_module(db: Session, module_id: int):
|
|
db_module = get_module_by_id(db, module_id)
|
|
if not db_module:
|
|
return False
|
|
|
|
db.delete(db_module)
|
|
db.commit()
|
|
return True
|
|
|
|
# Lesson operations
|
|
def get_lesson_by_id(db: Session, lesson_id: int):
|
|
return db.query(models.Lesson).filter(models.Lesson.id == lesson_id).first()
|
|
|
|
def get_module_lessons(db: Session, module_id: int):
|
|
return db.query(models.Lesson).filter(
|
|
models.Lesson.module_id == module_id
|
|
).order_by(models.Lesson.order_index).all()
|
|
|
|
def create_lesson(db: Session, lesson: schemas.LessonCreate):
|
|
db_lesson = models.Lesson(
|
|
module_id=lesson.module_id,
|
|
title=lesson.title,
|
|
description=lesson.description,
|
|
order_index=lesson.order_index
|
|
)
|
|
db.add(db_lesson)
|
|
db.commit()
|
|
db.refresh(db_lesson)
|
|
return db_lesson
|
|
|
|
def update_lesson(db: Session, lesson_id: int, lesson_update: schemas.LessonUpdate):
|
|
db_lesson = get_lesson_by_id(db, lesson_id)
|
|
if not db_lesson:
|
|
return None
|
|
|
|
update_data = lesson_update.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(db_lesson, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_lesson)
|
|
return db_lesson
|
|
|
|
def delete_lesson(db: Session, lesson_id: int):
|
|
db_lesson = get_lesson_by_id(db, lesson_id)
|
|
if not db_lesson:
|
|
return False
|
|
|
|
db.delete(db_lesson)
|
|
db.commit()
|
|
return True
|
|
|
|
# Step operations
|
|
def get_step_by_id(db: Session, step_id: int):
|
|
return db.query(models.LessonStep).filter(models.LessonStep.id == step_id).first()
|
|
|
|
def get_lesson_steps(db: Session, lesson_id: int):
|
|
return db.query(models.LessonStep).filter(
|
|
models.LessonStep.lesson_id == lesson_id
|
|
).order_by(models.LessonStep.order_index).all()
|
|
|
|
def create_step(db: Session, step: schemas.StepCreate):
|
|
content_html = render_markdown_to_html(step.content) if step.content else None
|
|
db_step = models.LessonStep(
|
|
lesson_id=step.lesson_id,
|
|
step_type=step.step_type,
|
|
title=step.title,
|
|
content=step.content,
|
|
content_html=content_html,
|
|
file_url=step.file_url,
|
|
order_index=step.order_index
|
|
)
|
|
db.add(db_step)
|
|
db.commit()
|
|
db.refresh(db_step)
|
|
return db_step
|
|
|
|
def update_step(db: Session, step_id: int, step_update: schemas.StepUpdate):
|
|
db_step = get_step_by_id(db, step_id)
|
|
if not db_step:
|
|
return None
|
|
|
|
update_data = step_update.model_dump(exclude_unset=True)
|
|
|
|
if 'content' in update_data:
|
|
update_data['content_html'] = render_markdown_to_html(update_data['content'])
|
|
|
|
for field, value in update_data.items():
|
|
setattr(db_step, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_step)
|
|
return db_step
|
|
|
|
def delete_step(db: Session, step_id: int):
|
|
db_step = get_step_by_id(db, step_id)
|
|
if not db_step:
|
|
return False
|
|
|
|
db.delete(db_step)
|
|
db.commit()
|
|
return True
|
|
|
|
# Stats operations
|
|
def get_stats(db: Session):
|
|
total_users = db.query(func.count(models.User.id)).scalar()
|
|
total_courses = db.query(func.count(models.Course.id)).scalar()
|
|
|
|
# For now, return the latest course as most popular
|
|
# In a real app, you would track course enrollments
|
|
most_popular = db.query(models.Course).order_by(desc(models.Course.created_at)).first()
|
|
most_popular_title = most_popular.title if most_popular else None
|
|
|
|
return {
|
|
"total_users": total_users or 0,
|
|
"total_courses": total_courses or 0,
|
|
"most_popular_course": most_popular_title
|
|
}
|
|
|
|
# Favorite courses operations
|
|
def get_favorite_courses(db: Session, user_id: int):
|
|
return db.query(models.FavoriteCourse).filter(models.FavoriteCourse.user_id == user_id).all()
|
|
|
|
def get_favorite_course(db: Session, user_id: int, course_id: int):
|
|
return db.query(models.FavoriteCourse).filter(
|
|
models.FavoriteCourse.user_id == user_id,
|
|
models.FavoriteCourse.course_id == course_id
|
|
).first()
|
|
|
|
def add_favorite_course(db: Session, user_id: int, course_id: int):
|
|
# Check if already favorited
|
|
existing = get_favorite_course(db, user_id, course_id)
|
|
if existing:
|
|
return existing
|
|
|
|
db_favorite = models.FavoriteCourse(user_id=user_id, course_id=course_id)
|
|
db.add(db_favorite)
|
|
db.commit()
|
|
db.refresh(db_favorite)
|
|
return db_favorite
|
|
|
|
def remove_favorite_course(db: Session, user_id: int, course_id: int):
|
|
favorite = get_favorite_course(db, user_id, course_id)
|
|
if not favorite:
|
|
return False
|
|
|
|
db.delete(favorite)
|
|
db.commit()
|
|
return True
|
|
|
|
# Progress operations
|
|
def get_progress(db: Session, user_id: int, course_id: int):
|
|
return db.query(models.Progress).filter(
|
|
models.Progress.user_id == user_id,
|
|
models.Progress.course_id == course_id
|
|
).first()
|
|
|
|
def get_user_progress(db: Session, user_id: int):
|
|
return db.query(models.Progress).filter(models.Progress.user_id == user_id).all()
|
|
|
|
def create_or_update_progress(db: Session, user_id: int, course_id: int, completed_lessons: int, total_lessons: int):
|
|
progress = get_progress(db, user_id, course_id)
|
|
if progress:
|
|
progress.completed_lessons = completed_lessons
|
|
progress.total_lessons = total_lessons
|
|
else:
|
|
progress = models.Progress(
|
|
user_id=user_id,
|
|
course_id=course_id,
|
|
completed_lessons=completed_lessons,
|
|
total_lessons=total_lessons
|
|
)
|
|
db.add(progress)
|
|
|
|
db.commit()
|
|
db.refresh(progress)
|
|
return progress
|
|
|
|
def delete_progress(db: Session, user_id: int, course_id: int):
|
|
progress = get_progress(db, user_id, course_id)
|
|
if not progress:
|
|
return False
|
|
|
|
db.delete(progress)
|
|
db.commit()
|
|
return True
|
|
|
|
# Course draft operations
|
|
def get_course_drafts(db: Session, user_id: int):
|
|
return db.query(models.CourseDraft).filter(
|
|
models.CourseDraft.user_id == user_id
|
|
).order_by(models.CourseDraft.updated_at.desc()).all()
|
|
|
|
def get_course_draft(db: Session, draft_id: int):
|
|
return db.query(models.CourseDraft).filter(models.CourseDraft.id == draft_id).first()
|
|
|
|
def save_course_draft(db: Session, draft: schemas.CourseDraftCreate, user_id: int):
|
|
# Check if draft exists
|
|
existing = db.query(models.CourseDraft).filter(
|
|
models.CourseDraft.user_id == user_id,
|
|
models.CourseDraft.title == draft.title
|
|
).first()
|
|
|
|
if existing:
|
|
# Update existing draft
|
|
existing.description = draft.description
|
|
existing.level = draft.level
|
|
existing.content = draft.content
|
|
existing.structure = draft.structure or existing.structure
|
|
existing.updated_at = func.now()
|
|
db.commit()
|
|
db.refresh(existing)
|
|
return existing
|
|
else:
|
|
# Create new draft
|
|
db_draft = models.CourseDraft(
|
|
user_id=user_id,
|
|
title=draft.title,
|
|
description=draft.description,
|
|
level=draft.level,
|
|
content=draft.content,
|
|
structure=draft.structure
|
|
)
|
|
db.add(db_draft)
|
|
db.commit()
|
|
db.refresh(db_draft)
|
|
return db_draft
|
|
|
|
def delete_course_draft(db: Session, draft_id: int, user_id: int):
|
|
draft = db.query(models.CourseDraft).filter(
|
|
models.CourseDraft.id == draft_id,
|
|
models.CourseDraft.user_id == user_id
|
|
).first()
|
|
if not draft:
|
|
return False
|
|
|
|
db.delete(draft)
|
|
db.commit()
|
|
return True
|
|
|
|
# Draft-specific queries using Course.status='draft'
|
|
def get_user_drafts(db: Session, user_id: int):
|
|
"""Get all draft courses for a user (status='draft')"""
|
|
return db.query(models.Course).filter(
|
|
models.Course.author_id == user_id,
|
|
models.Course.status == 'draft'
|
|
).order_by(models.Course.updated_at.desc()).all()
|
|
|
|
def get_user_draft(db: Session, user_id: int, draft_id: int, include_structure: bool = False):
|
|
"""Get a specific draft for a user"""
|
|
query = db.query(models.Course).filter(
|
|
models.Course.id == draft_id,
|
|
models.Course.author_id == user_id,
|
|
models.Course.status == 'draft'
|
|
)
|
|
|
|
if include_structure:
|
|
# Предзагрузка связей для полной структуры
|
|
query = query.options(
|
|
joinedload(models.Course.modules).joinedload(models.CourseModule.lessons).joinedload(models.Lesson.steps)
|
|
)
|
|
|
|
return query.first()
|
|
|
|
def publish_course(db: Session, course_id: int, user_id: int):
|
|
"""Publish a draft by changing status to 'published'"""
|
|
draft = get_user_draft(db, user_id, course_id)
|
|
if not draft:
|
|
return None
|
|
|
|
draft.status = 'published'
|
|
db.commit()
|
|
db.refresh(draft)
|
|
return draft
|
|
|
|
def delete_user_draft(db: Session, course_id: int, user_id: int):
|
|
"""Delete a user's draft course"""
|
|
draft = get_user_draft(db, user_id, course_id)
|
|
if not draft:
|
|
return False
|
|
|
|
db.delete(draft)
|
|
db.commit()
|
|
return True
|
|
|
|
# Test operations
|
|
def get_test_questions(db: Session, step_id: int):
|
|
"""Get all questions for a test step"""
|
|
return db.query(models.TestQuestion).filter(
|
|
models.TestQuestion.step_id == step_id
|
|
).order_by(models.TestQuestion.order_index).all()
|
|
|
|
def get_test_question(db: Session, question_id: int):
|
|
"""Get a specific test question"""
|
|
return db.query(models.TestQuestion).options(
|
|
joinedload(models.TestQuestion.answers),
|
|
joinedload(models.TestQuestion.match_items)
|
|
).filter(
|
|
models.TestQuestion.id == question_id
|
|
).first()
|
|
|
|
def create_test_question(db: Session, step_id: int, question: schemas.TestQuestionCreate):
|
|
"""Create a new test question with answers/match items"""
|
|
db_question = models.TestQuestion(
|
|
step_id=step_id,
|
|
question_text=question.question_text,
|
|
question_type=question.question_type,
|
|
points=question.points,
|
|
order_index=question.order_index
|
|
)
|
|
db.add(db_question)
|
|
db.flush() # Flush to get the question ID
|
|
|
|
# Create answers for choice questions
|
|
if question.question_type in ['single_choice', 'multiple_choice']:
|
|
for answer in question.answers:
|
|
db_answer = models.TestAnswer(
|
|
question_id=db_question.id,
|
|
answer_text=answer.answer_text,
|
|
is_correct=answer.is_correct,
|
|
order_index=answer.order_index
|
|
)
|
|
db.add(db_answer)
|
|
|
|
# Create match items for match questions
|
|
if question.question_type == 'match':
|
|
for item in question.match_items:
|
|
db_item = models.TestMatchItem(
|
|
question_id=db_question.id,
|
|
left_text=item.left_text,
|
|
right_text=item.right_text,
|
|
order_index=item.order_index
|
|
)
|
|
db.add(db_item)
|
|
|
|
# Store correct answers for text_answer questions in content field
|
|
if question.question_type == 'text_answer' and question.correct_answers:
|
|
# Store as JSON in question_text temporarily (we'll need a separate field)
|
|
# For now, we'll store it in the database and handle it properly
|
|
pass
|
|
|
|
db.commit()
|
|
db.refresh(db_question)
|
|
return db_question
|
|
|
|
def update_test_question(db: Session, question_id: int, question_update: schemas.TestQuestionUpdate):
|
|
"""Update a test question"""
|
|
db_question = get_test_question(db, question_id)
|
|
if not db_question:
|
|
return None
|
|
|
|
update_data = question_update.model_dump(exclude_unset=True)
|
|
|
|
# Update question fields first
|
|
if 'question_text' in update_data:
|
|
db_question.question_text = update_data['question_text']
|
|
if 'question_type' in update_data:
|
|
db_question.question_type = update_data['question_type']
|
|
if 'points' in update_data:
|
|
db_question.points = update_data['points']
|
|
if 'order_index' in update_data:
|
|
db_question.order_index = update_data['order_index']
|
|
|
|
# Flush to save question type changes before updating answers
|
|
db.flush()
|
|
|
|
# Update answers if provided
|
|
if 'answers' in update_data and db_question.question_type in ['single_choice', 'multiple_choice']:
|
|
# Delete existing answers
|
|
db.query(models.TestAnswer).filter(
|
|
models.TestAnswer.question_id == question_id
|
|
).delete()
|
|
|
|
# Create new answers
|
|
for answer_data in update_data['answers']:
|
|
db_answer = models.TestAnswer(
|
|
question_id=question_id,
|
|
answer_text=answer_data.get('answer_text', '') if isinstance(answer_data, dict) else answer_data.answer_text,
|
|
is_correct=answer_data.get('is_correct', False) if isinstance(answer_data, dict) else answer_data.is_correct,
|
|
order_index=answer_data.get('order_index', 0) if isinstance(answer_data, dict) else answer_data.order_index
|
|
)
|
|
db.add(db_answer)
|
|
|
|
# Update match items if provided
|
|
if 'match_items' in update_data and db_question.question_type == 'match':
|
|
# Delete existing match items
|
|
db.query(models.TestMatchItem).filter(
|
|
models.TestMatchItem.question_id == question_id
|
|
).delete()
|
|
|
|
# Create new match items
|
|
for item_data in update_data['match_items']:
|
|
db_item = models.TestMatchItem(
|
|
question_id=question_id,
|
|
left_text=item_data.get('left_text', '') if isinstance(item_data, dict) else item_data.left_text,
|
|
right_text=item_data.get('right_text', '') if isinstance(item_data, dict) else item_data.right_text,
|
|
order_index=item_data.get('order_index', 0) if isinstance(item_data, dict) else item_data.order_index
|
|
)
|
|
db.add(db_item)
|
|
|
|
db.commit()
|
|
# Re-query to get fresh data with relationships
|
|
db_question = get_test_question(db, question_id)
|
|
return db_question
|
|
|
|
def delete_test_question(db: Session, question_id: int):
|
|
"""Delete a test question"""
|
|
db_question = get_test_question(db, question_id)
|
|
if not db_question:
|
|
return False
|
|
|
|
db.delete(db_question)
|
|
db.commit()
|
|
return True
|
|
|
|
def update_test_settings(db: Session, step_id: int, settings: schemas.TestSettings):
|
|
"""Update test settings for a step"""
|
|
db_step = get_step_by_id(db, step_id)
|
|
if not db_step:
|
|
return None
|
|
|
|
db_step.test_settings = settings.model_dump()
|
|
db.commit()
|
|
db.refresh(db_step)
|
|
return db_step
|
|
|
|
def create_test_attempt(db: Session, user_id: int, step_id: int, answers: dict):
|
|
"""Create a test attempt and calculate score"""
|
|
# Get test settings
|
|
db_step = get_step_by_id(db, step_id)
|
|
if not db_step or not db_step.test_settings:
|
|
return None
|
|
|
|
settings = db_step.test_settings
|
|
|
|
# Get all questions
|
|
questions = get_test_questions(db, step_id)
|
|
if not questions:
|
|
return None
|
|
|
|
# Calculate score
|
|
total_points = 0
|
|
earned_points = 0
|
|
|
|
for question in questions:
|
|
total_points += question.points
|
|
question_answer = answers.get(str(question.id))
|
|
|
|
if not question_answer:
|
|
continue
|
|
|
|
if question.question_type == 'single_choice':
|
|
# Single choice: check if selected answer is correct
|
|
correct_answer = db.query(models.TestAnswer).filter(
|
|
models.TestAnswer.question_id == question.id,
|
|
models.TestAnswer.is_correct == True
|
|
).first()
|
|
|
|
if correct_answer and question_answer == correct_answer.id:
|
|
earned_points += question.points
|
|
|
|
elif question.question_type == 'multiple_choice':
|
|
# Multiple choice: check if all correct answers are selected
|
|
correct_answers = db.query(models.TestAnswer).filter(
|
|
models.TestAnswer.question_id == question.id,
|
|
models.TestAnswer.is_correct == True
|
|
).all()
|
|
|
|
correct_ids = set([a.id for a in correct_answers])
|
|
selected_ids = set(question_answer) if isinstance(question_answer, list) else set()
|
|
|
|
if correct_ids == selected_ids:
|
|
earned_points += question.points
|
|
|
|
elif question.question_type == 'text_answer':
|
|
# Text answer: simple keyword matching (case-insensitive)
|
|
# For better implementation, use fuzzy matching or AI
|
|
# Store correct answers in question content for now
|
|
pass
|
|
|
|
elif question.question_type == 'match':
|
|
# Match: check if all pairs are correct
|
|
match_items = db.query(models.TestMatchItem).filter(
|
|
models.TestMatchItem.question_id == question.id
|
|
).all()
|
|
|
|
all_correct = True
|
|
for item in match_items:
|
|
user_match = question_answer.get(str(item.id))
|
|
if user_match != item.right_text:
|
|
all_correct = False
|
|
break
|
|
|
|
if all_correct:
|
|
earned_points += question.points
|
|
|
|
# Calculate percentage and determine if passed
|
|
score_percentage = (earned_points / total_points * 100) if total_points > 0 else 0
|
|
passed = score_percentage >= settings.get('passing_score', 70)
|
|
|
|
# Create attempt record
|
|
db_attempt = models.TestAttempt(
|
|
user_id=user_id,
|
|
step_id=step_id,
|
|
score=earned_points,
|
|
max_score=total_points,
|
|
passed=passed,
|
|
answers=answers,
|
|
completed_at=datetime.utcnow()
|
|
)
|
|
db.add(db_attempt)
|
|
db.commit()
|
|
db.refresh(db_attempt)
|
|
|
|
return db_attempt
|
|
|
|
def get_test_attempts(db: Session, user_id: int, step_id: int):
|
|
"""Get all test attempts for a user and step"""
|
|
return db.query(models.TestAttempt).filter(
|
|
models.TestAttempt.user_id == user_id,
|
|
models.TestAttempt.step_id == step_id
|
|
).order_by(models.TestAttempt.started_at.desc()).all()
|
|
|
|
def get_test_attempt(db: Session, attempt_id: int):
|
|
"""Get a specific test attempt"""
|
|
return db.query(models.TestAttempt).filter(
|
|
models.TestAttempt.id == attempt_id
|
|
).first()
|