erpnext/education/utils.py - Clicks/Forks/ERPNext - Gitiles

 # 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
	# 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