brotherton-erpnext/erpnext/education/utils.py

# Copyright (c) 2015, Frappe Technologies and contributors

import frappe
from frappe import _


class OverlapError(frappe.ValidationError): pass

def validate_overlap_for(doc, doctype, fieldname, value=None):
	"""Checks overlap for specified field.

	:param fieldname: Checks Overlap for this field
	"""

	existing = get_overlap_for(doc, doctype, fieldname, value)
	if existing:
		frappe.throw(_("This {0} conflicts with {1} for {2} {3}").format(doc.doctype, existing.name,
			doc.meta.get_label(fieldname) if not value else fieldname , value or doc.get(fieldname)), OverlapError)

def get_overlap_for(doc, doctype, fieldname, value=None):
	"""Returns overlaping document for specified field.

	:param fieldname: Checks Overlap for this field
	"""

	existing = frappe.db.sql("""select name, from_time, to_time from `tab{0}`
		where `{1}`=%(val)s and schedule_date = %(schedule_date)s and
		(
			(from_time > %(from_time)s and from_time < %(to_time)s) or
			(to_time > %(from_time)s and to_time < %(to_time)s) or
			(%(from_time)s > from_time and %(from_time)s < to_time) or
			(%(from_time)s = from_time and %(to_time)s = to_time))
		and name!=%(name)s and docstatus!=2""".format(doctype, fieldname),
		{
			"schedule_date": doc.schedule_date,
			"val": value or doc.get(fieldname),
			"from_time": doc.from_time,
			"to_time": doc.to_time,
			"name": doc.name or "No Name"
		}, as_dict=True)

	return existing[0] if existing else None


def validate_duplicate_student(students):
	unique_students= []
	for stud in students:
		if stud.student in unique_students:
			frappe.throw(_("Student {0} - {1} appears Multiple times in row {2} & {3}")
				.format(stud.student, stud.student_name, unique_students.index(stud.student)+1, stud.idx))
		else:
			unique_students.append(stud.student)

		return None

# LMS Utils
def get_current_student():
	"""Returns current student from frappe.session.user

	Returns:
		object: Student Document
	"""
	email = frappe.session.user
	if email in ('Administrator', 'Guest'):
		return None
	try:
		student_id = frappe.get_all("Student", {"student_email_id": email}, ["name"])[0].name
		return frappe.get_doc("Student", student_id)
	except (IndexError, frappe.DoesNotExistError):
		return None

def get_portal_programs():
	"""Returns a list of all program to be displayed on the portal
	Programs are returned based on the following logic
		is_published and (student_is_enrolled or student_can_self_enroll)

	Returns:
		list of dictionary: List of all programs and to be displayed on the portal along with access rights
	"""
	published_programs = frappe.get_all("Program", filters={"is_published": True})
	if not published_programs:
		return None

	program_list = [frappe.get_doc("Program", program) for program in published_programs]
	portal_programs = [{'program': program, 'has_access': allowed_program_access(program.name)} for program in program_list if allowed_program_access(program.name) or program.allow_self_enroll]

	return portal_programs

def allowed_program_access(program, student=None):
	"""Returns enrollment status for current student

	Args:
		program (string): Name of the program
		student (object): instance of Student document

	Returns:
		bool: Is current user enrolled or not
	"""
	if has_super_access():
		return True
	if not student:
		student = get_current_student()
	if student and get_enrollment('program', program, student.name):
		return True
	else:
		return False

def get_enrollment(master, document, student):
	"""Gets enrollment for course or program

	Args:
		master (string): can either be program or course
		document (string): program or course name
		student (string): Student ID

	Returns:
		string: Enrollment Name if exists else returns empty string
	"""
	if master == 'program':
		enrollments = frappe.get_all("Program Enrollment", filters={'student':student, 'program': document, 'docstatus': 1})
	if master == 'course':
		enrollments = frappe.get_all("Course Enrollment", filters={'student':student, 'course': document})

	if enrollments:
		return enrollments[0].name
	else:
		return None

@frappe.whitelist()
def enroll_in_program(program_name, student=None):
	"""Enroll student in program

	Args:
		program_name (string): Name of the program to be enrolled into
		student (string, optional): name of student who has to be enrolled, if not
			provided, a student will be created from the current user

	Returns:
		string: name of the program enrollment document
	"""
	if has_super_access():
		return

	if not student == None:
		student = frappe.get_doc("Student", student)
	else:
		# Check if self enrollment in allowed
		program = frappe.get_doc('Program', program_name)
		if not program.allow_self_enroll:
			return frappe.throw(_("You are not allowed to enroll for this course"))

		student = get_current_student()
		if not student:
			student = create_student_from_current_user()

	# Check if student is already enrolled in program
	enrollment = get_enrollment('program', program_name, student.name)
	if enrollment:
		return enrollment

	# Check if self enrollment in allowed
	program = frappe.get_doc('Program', program_name)
	if not program.allow_self_enroll:
		return frappe.throw(_("You are not allowed to enroll for this course"))

	# Enroll in program
	program_enrollment = student.enroll_in_program(program_name)
	return program_enrollment.name

def has_super_access():
	"""Check if user has a role that allows full access to LMS

	Returns:
		bool: true if user has access to all lms content
	"""
	current_user = frappe.get_doc('User', frappe.session.user)
	roles = set([role.role for role in current_user.roles])
	return bool(roles & {'Administrator', 'Instructor', 'Education Manager', 'System Manager', 'Academic User'})

@frappe.whitelist()
def add_activity(course, content_type, content, program):
	if has_super_access():
		return None

	student = get_current_student()
	if not student:
		return frappe.throw(_("Student with email {0} does not exist").format(frappe.session.user), frappe.DoesNotExistError)

	enrollment = get_or_create_course_enrollment(course, program)
	if content_type == 'Quiz':
		return
	else:
		return enrollment.add_activity(content_type, content)

@frappe.whitelist()
def evaluate_quiz(quiz_response, quiz_name, course, program, time_taken):
	import json

	student = get_current_student()

	quiz_response = json.loads(quiz_response)
	quiz = frappe.get_doc("Quiz", quiz_name)
	result, score, status = quiz.evaluate(quiz_response, quiz_name)

	if has_super_access():
		return {'result': result, 'score': score, 'status': status}

	if student:
		enrollment = get_or_create_course_enrollment(course, program)
		if quiz.allowed_attempt(enrollment, quiz_name):
			enrollment.add_quiz_activity(quiz_name, quiz_response, result, score, status, time_taken)
			return {'result': result, 'score': score, 'status': status}
		else:
			return None

@frappe.whitelist()
def get_quiz(quiz_name, course):
	try:
		quiz = frappe.get_doc("Quiz", quiz_name)
		questions = quiz.get_questions()
	except Exception:
		frappe.throw(_("Quiz {0} does not exist").format(quiz_name), frappe.DoesNotExistError)
		return None

	questions = [{
		'name': question.name,
		'question': question.question,
		'type': question.question_type,
		'options': [{'name': option.name, 'option': option.option}
					for option in question.options],
		} for question in questions]

	if has_super_access():
		return {
			'questions': questions,
			'activity': None,
			'is_time_bound': quiz.is_time_bound,
			'duration': quiz.duration
		}

	student = get_current_student()
	course_enrollment = get_enrollment("course", course, student.name)
	status, score, result, time_taken = check_quiz_completion(quiz, course_enrollment)
	return {
		'questions': questions,
		'activity': {'is_complete': status, 'score': score, 'result': result, 'time_taken': time_taken},
		'is_time_bound': quiz.is_time_bound,
		'duration': quiz.duration
	}

def get_topic_progress(topic, course_name, program):
	"""
	Return the porgress of a course in a program as well as the content to continue from.
		:param topic_name:
		:param course_name:
	"""
	student = get_current_student()
	if not student:
		return None
	course_enrollment = get_or_create_course_enrollment(course_name, program)
	progress = student.get_topic_progress(course_enrollment.name, topic)
	if not progress:
		return None
	count = sum([activity['is_complete'] for activity in progress])
	if count == 0:
		return {'completed': False, 'started': False}
	elif count == len(progress):
		return {'completed': True, 'started': True}
	elif count < len(progress):
		return {'completed': False, 'started': True}

def get_course_progress(course, program):
	"""
	Return the porgress of a course in a program as well as the content to continue from.
		:param topic_name:
		:param course_name:
	"""
	course_progress = []
	for course_topic in course.topics:
		topic = frappe.get_doc("Topic", course_topic.topic)
		progress = get_topic_progress(topic, course.name, program)
		if progress:
			course_progress.append(progress)
	if course_progress:
		number_of_completed_topics = sum([activity['completed'] for activity in course_progress])
		total_topics = len(course_progress)
		if total_topics == 1:
			return course_progress[0]
		if number_of_completed_topics == 0:
			return {'completed': False, 'started': False}
		if number_of_completed_topics == total_topics:
			return {'completed': True, 'started': True}
		if number_of_completed_topics < total_topics:
			return {'completed': False, 'started': True}

	return None

def get_program_progress(program):
	program_progress = []
	if not program.courses:
		return None
	for program_course in program.courses:
		course = frappe.get_doc("Course", program_course.course)
		progress = get_course_progress(course, program.name)
		if progress:
			progress['name'] = course.name
			progress['course'] = course.course_name
			program_progress.append(progress)

	if program_progress:
		return program_progress

	return None

def get_program_completion(program):
	topics = frappe.db.sql("""select `tabCourse Topic`.topic, `tabCourse Topic`.parent
	from `tabCourse Topic`,
		 `tabProgram Course`
	where `tabCourse Topic`.parent = `tabProgram Course`.course
			and `tabProgram Course`.parent = %s""", program.name)

	progress = []
	for topic in topics:
		topic_doc = frappe.get_doc('Topic', topic[0])
		topic_progress = get_topic_progress(topic_doc, topic[1], program.name)
		if topic_progress:
			progress.append(topic_progress)

	if progress:
		number_of_completed_topics = sum([activity['completed'] for activity in progress if activity])
		total_topics = len(progress)
		try:
			return int((float(number_of_completed_topics)/total_topics)*100)
		except ZeroDivisionError:
			return 0

	return 0

def create_student_from_current_user():
	user = frappe.get_doc("User", frappe.session.user)

	student = frappe.get_doc({
		"doctype": "Student",
		"first_name": user.first_name,
		"last_name": user.last_name,
		"student_email_id": user.email,
		"user": frappe.session.user
		})

	student.save(ignore_permissions=True)
	return student

def get_or_create_course_enrollment(course, program):
	student = get_current_student()
	course_enrollment = get_enrollment("course", course, student.name)
	if not course_enrollment:
		program_enrollment = get_enrollment('program', program.name, student.name)
		if not program_enrollment:
			frappe.throw(_("You are not enrolled in program {0}").format(program))
			return
		return student.enroll_in_course(course_name=course, program_enrollment=get_enrollment('program', program.name, student.name))
	else:
		return frappe.get_doc('Course Enrollment', course_enrollment)

def check_content_completion(content_name, content_type, enrollment_name):
	activity = frappe.get_all("Course Activity", filters={'enrollment': enrollment_name, 'content_type': content_type, 'content': content_name})
	if activity:
		return True
	else:
		return False

def check_quiz_completion(quiz, enrollment_name):
	attempts = frappe.get_all("Quiz Activity",
		filters={
			'enrollment': enrollment_name,
			'quiz': quiz.name
		},
		fields=["name", "activity_date", "score", "status", "time_taken"]
	)
	status = False if quiz.max_attempts == 0 else bool(len(attempts) >= quiz.max_attempts)
	score = None
	result = None
	time_taken = None
	if attempts:
		if quiz.grading_basis == 'Last Highest Score':
			attempts = sorted(attempts, key = lambda i: int(i.score), reverse=True)
		score = attempts[0]['score']
		result = attempts[0]['status']
		time_taken = attempts[0]['time_taken']
		if result == 'Pass':
			status = True
	return status, score, result, time_taken