# -*- coding: utf-8 -*-
#
# This file is part of INGInious. See the LICENSE and the COPYRIGHTS files for
# more information about the licensing of this file.
import bson
import json
import logging
from collections import OrderedDict
from flask import session, request, render_template
from natsort import natsorted
from inginious.frontend.tasks import Task
from inginious.frontend.pages.course_admin.utils import INGIniousAdminPage
from inginious.common.exceptions import TaskAlreadyExistsException
from inginious.frontend.task_dispensers import get_task_dispensers
from inginious.frontend.models import UserTask, Submission
[docs]
class CourseTaskListPage(INGIniousAdminPage):
""" List informations about all tasks """
[docs]
def GET_AUTH(self, courseid): # pylint: disable=arguments-differ
""" GET request """
course, __ = self.get_course_and_check_rights(courseid, allow_all_staff=True)
return self.page(course)
[docs]
def POST_AUTH(self, courseid): # pylint: disable=arguments-differ
""" POST request """
course, __ = self.get_course_and_check_rights(courseid, allow_all_staff=False)
errors = []
user_input = request.form
if "task_dispenser" in user_input:
selected_task_dispenser = user_input.get("task_dispenser", "toc")
task_dispenser_class = get_task_dispensers().get(selected_task_dispenser, None)
if task_dispenser_class:
course.set_descriptor_element('task_dispenser', task_dispenser_class.get_id())
course.set_descriptor_element('dispenser_data', {})
course.save()
else:
errors.append(_("Invalid task dispenser"))
elif "migrate_tasks" in user_input:
task_dispenser = course.get_task_dispenser()
try:
data = task_dispenser.import_legacy_tasks()
self.update_dispenser(course, data)
self.clean_task_files(course)
except Exception as e:
errors.append(_("Something wrong happened: ") + str(e))
else:
try:
self.update_dispenser(course, json.loads(user_input["course_structure"]))
except Exception as e:
errors.append(_("Something wrong happened: ") + str(e))
for taskid in json.loads(user_input.get("new_tasks", "[]")):
try:
task_fs = course.get_fs().from_subfolder(taskid)
if task_fs.exists("task.yaml"):
raise TaskAlreadyExistsException("Task with id " + taskid + " already exists.")
t = Task(courseid, taskid, {"name": taskid, "problems": {}, "environment_type": "mcq"})
t.save()
except Exception as ex:
errors.append(_("Couldn't create task {} : ").format(taskid) + str(ex))
for taskid in json.loads(user_input.get("deleted_tasks", "[]")):
try:
t = Task.get(taskid, course.get_fs())
t.delete()
except Exception as ex:
errors.append(_("Couldn't delete task {} : ").format(taskid) + str(ex))
for taskid in json.loads(user_input.get("wiped_tasks", "[]")):
try:
self.wipe_task(courseid, taskid)
except Exception as ex:
errors.append(_("Couldn't wipe task {} : ").format(taskid) + str(ex))
# don't forget to reload the modified course
course, __ = self.get_course_and_check_rights(courseid, allow_all_staff=False)
return self.page(course, errors, not errors)
[docs]
def update_dispenser(self, course, dispenser_data):
""" Update the task dispenser based on dispenser_data """
task_dispenser = course.get_task_dispenser()
data, msg = task_dispenser.check_dispenser_data(dispenser_data)
if data:
course.set_descriptor_element('task_dispenser',task_dispenser.get_id())
course.set_descriptor_element('dispenser_data', data)
course.save()
else:
raise Exception(_("Invalid course structure: ") + msg)
[docs]
def clean_task_files(self, course):
task_dispenser = course.get_task_dispenser()
legacy_fields = task_dispenser.legacy_fields.keys()
for taskid, task in course.get_tasks().items():
task.drop_legacy_fields(legacy_fields)
task.save()
[docs]
def submission_url_generator(self, taskid):
""" Generates a submission url """
return "?format=taskid%2Fusername&tasks=" + taskid
[docs]
def wipe_task(self, courseid, taskid):
""" Wipe the data associated to the taskid from DB"""
for submission in Submission.objects(courseid=courseid, taskid=taskid):
submission.archive.delete()
submission.input.delete()
UserTask.objects(courseid=courseid, taskid=taskid).delete()
Submission.objects(courseid=courseid, taskid=taskid).delete()
logging.getLogger("inginious.webapp.task_edit").info("Task %s/%s wiped.", courseid, taskid)
[docs]
def page(self, course, errors=None, validated=False):
""" Get all data and display the page """
# Load tasks and verify exceptions
files = course.get_readable_tasks()
tasks = {}
if errors is None:
errors = []
tasks_errors = {}
for taskid in files:
try:
tasks[taskid] = course.get_task(taskid)
except Exception as ex:
tasks_errors[taskid] = str(ex)
tasks_data = natsorted([(taskid, {"name": tasks[taskid].get_name(session.language),
"url": self.submission_url_generator(taskid)}) for taskid in tasks],
key=lambda x: x[1]["name"])
tasks_data = OrderedDict(tasks_data)
task_dispensers = get_task_dispensers()
return render_template("course_admin/task_list.html", course=course,
task_dispensers=task_dispensers, tasks=tasks_data, errors=errors,
tasks_errors=tasks_errors, validated=validated)