import io import re import os import json import yaml import requests import more_itertools import ruamel.yaml from django.core.exceptions import ObjectDoesNotExist from yaksh.models import Lesson, Course, LearningUnit, LearningModule, Quiz, TableOfContents _HEADER_RE = re.compile(r"^---\s*$") _BLANK_RE = re.compile(r"^\s*$") _JUPYTER_RE = re.compile(r"^meta\s*:\s*$") _LEFTSPACE_RE = re.compile(r"^\s") DATA_SEP = '~#~#~#' def recursive_update(target, update): """ Update recursively a (nested) dictionary with the content of another. Inspired from https://stackoverflow.com/questions/3232943/update-value-of-a-nested-dictionary-of-varying-depth """ for key in update: value = update[key] if value is None: del target[key] elif isinstance(value, dict): target[key] = recursive_update(target.get(key, {}), value) else: target[key] = value return target def _metadata_to_dict(lines): metadata= {} ended = False injupyter = False # jupyter = [] meta_lines = [] for i, line in enumerate(lines): if i == 0 and _HEADER_RE.match(line): continue if (i > 0) and _HEADER_RE.match(line): ended = True if _JUPYTER_RE.match(line): injupyter = True elif line and not _LEFTSPACE_RE.match(line): injupyter = False if injupyter: meta_lines.append(line) if ended: if meta_lines: recursive_update(metadata, yaml.safe_load("\n".join(meta_lines))) return metadata def _display_content_to_dict(lines): ended = False display_content = [] for i, line in enumerate(lines): if (i > 0) and _HEADER_RE.match(line): ended = True continue if ended: display_content.append(line) return ''.join(display_content) class LessonData(): type = "lesson" display_content_field = 'description' def __init__(self, obj, order, course_id): toc_file = TableOfContents.objects.get_all_tocs_as_yaml( course_id, obj.id, '{0}_lesson_toc.yaml'.format(obj.id) ) self.data = { "id": obj.id, "name": obj.name, "active": obj.active, "order": order, } if toc_file: self.data.update( {"toc": toc_file,} ) self.display_content = obj.description if obj.description else None class QuizData(): type = "quiz" display_content_field = 'instructions' def __init__(self, obj, order): self.data = { "id": obj.id, "description": obj.description, "active": obj.active, "start_date_time": obj.start_date_time, "end_date_time": obj.end_date_time, "duration": obj.duration, "pass_criteria": obj.pass_criteria, "attempts_allowed": obj.attempts_allowed, "allow_skip": obj.allow_skip, "view_answerpaper": obj.view_answerpaper, "order": order, } self.display_content = obj.instructions if obj.instructions else None class UnitData(): @classmethod def get_lesson_or_quiz_data(cls, units, course_id): return [ LessonData(unit_obj.lesson, unit_obj.order, course_id) if unit_obj.lesson else QuizData(unit_obj.quiz, unit_obj.order) for unit_obj in units ] @classmethod def md_to_dict(cls, lines): all_unit_data = list( more_itertools.split_at( lines, lambda x: x.strip('\n') == DATA_SEP ) ) learning_units = [] module_data = None for unit_lines in all_unit_data: if unit_lines: metadata = _metadata_to_dict(unit_lines) display_content = _display_content_to_dict(unit_lines) clean_display_content = display_content.strip('\n') try: unit_data = metadata.get('meta', {}).get('data', {}) order = unit_data.pop('order', 0) unit_type = metadata.get('meta', {}).get('type', None) unit_cls = LessonData if unit_type == 'lesson' else QuizData except KeyError: print("[ERROR] Data not found. Please check the MD file") unit_data.update( {getattr(unit_cls,'display_content_field'): clean_display_content,} ) if unit_type == 'lesson': learning_unit_data = { "order": order, "type": unit_type, "quiz": None, "lesson": unit_data, } learning_units.append(learning_unit_data) elif unit_type == 'quiz': learning_unit_data = { "order": order, "type": unit_type, "quiz": unit_data, "lesson": None, } learning_units.append(learning_unit_data) elif unit_type == 'module': module_data = ModuleData.md_to_dict(unit_lines) return {'module': module_data, 'learning_units': learning_units} @classmethod def md_to_dict_from_file(cls, file): with open(file, 'r') as f: lines = f.readlines() return cls.md_to_dict(lines) class ModuleData(): type = "module" display_content_field = 'description' def __init__(self, obj, course_id): self.data = { "id": obj.id, "name": obj.name, "order": obj.order, "active":obj.active, } self.display_content = obj.description if obj.description else None self.units = self.set_lessons(obj.get_learning_units(), course_id) def set_lessons(self, units, course_id): return UnitData.get_lesson_or_quiz_data(units, course_id) @classmethod def md_to_dict(cls, lines): metadata = _metadata_to_dict(lines) display_content = _display_content_to_dict(lines) clean_display_content = display_content.strip('\n') try: data = metadata.get('meta', {}).get('data', {}) except KeyError: print("[ERROR] Data not found. Please check the MD file") data.update({cls.display_content_field: clean_display_content}) return data @classmethod def md_to_dict_from_file(cls, file): with open(file, 'r') as f: lines = f.readlines() return cls.md_to_dict(lines) class CourseData(): type = "course" display_content_field = "instructions" def __init__(self, obj): self.data = { "id": obj.id, "name": obj.name, "enrollment": obj.enrollment, "active": obj.active, "start_enroll_time": obj.start_enroll_time, "end_enroll_time": obj.end_enroll_time, } self.display_content = obj.instructions if obj.instructions else None self.modules = self.set_modules(obj.get_learning_modules(), obj.id) def set_modules(self, modules, course_id): return [ModuleData(module_obj, course_id) for module_obj in modules] @classmethod def md_to_dict(cls, file): with open(file, 'r') as f: lines = f.readlines() metadata = _metadata_to_dict(lines) display_content = _display_content_to_dict(lines) clean_display_content = display_content.strip('\n') try: data = metadata.get('meta', {}).get('data', {}) except KeyError: print("[ERROR] Data not found. Please check the MD file") data.update({cls.display_content_field: clean_display_content}) return data def create_header(data, dtype): header = [] metadata = { "type": dtype, "data": data } yaml=ruamel.yaml.YAML() yaml.default_flow_style = False io_obj = io.StringIO() yaml.dump({"meta": metadata}, io_obj) if metadata['data'].get('id', None): raw_yaml_data = yaml.load(io_obj.getvalue()) raw_yaml_data['meta']['data'].yaml_add_eol_comment('Do Not Change This Value', 'id') io_obj = io.StringIO() yaml.dump(raw_yaml_data, io_obj) header.extend(io_obj.getvalue().splitlines()) header = ["---"] + header + ["---"] return header def get_course_data(course_id): course_obj = Course.objects.get(id=course_id) return CourseData(course_obj) def _create_clean_file_name(name, ext): clean_name = re.sub(r'[^\w\s-]', '', name).strip().lower() return re.sub(r'[-\s]+', '_', clean_name) + '.' + ext def _get_file_name_from_object(content_object, ext): name = ' '.join( [content_object.type, content_object.data.get('name')] ) return _create_clean_file_name(name, ext) def create_md(content_object, file_name=None, multiple_obj=False): file_content = [] if not file_name: dest_file_name = _get_file_name_from_object(content_object, 'md') else: dest_file_name = file_name data = content_object.data dtype = content_object.type seperator = '\n{0}\n'.format(DATA_SEP) if multiple_obj else '' header = create_header(data, dtype) if content_object.display_content: file_content.append(content_object.display_content) file_content = ( '\n'.join(header) + '\n\n' + '\n'.join(file_content) + seperator ) with open(dest_file_name, 'a+') as f: f.write(file_content) return dest_file_name def write_course_to_file(course_id): # Create the course md file course = get_course_data(course_id) course_file = create_md(course) course_map = {'course': course_file, 'modules': []} # Create the modules and lessons md files for module in course.modules: mod_file = create_md(module, multiple_obj=True) for lesson in module.units: create_md(lesson, mod_file, multiple_obj=True) course_map['modules'].append( {'file': mod_file,} ) with open('toc.yml', 'w') as f: f.write(yaml.safe_dump(course_map)) def convert_md_to_dict(toc, user): course_file = toc.get('course') course_data = CourseData.md_to_dict(course_file) module_obj_list = [] for module in toc.get('modules', []): module_file = module.get('file') module_data = UnitData.md_to_dict_from_file( module_file ).get('module') module_id = module_data.get('id', None) if module_id: mod_created = False module_obj = LearningModule.objects.get(id=module_id) module_obj.__dict__.update(module_data) module_obj.save() else: mod_created = True module_data.update({'creator': user}) module_obj = LearningModule.objects.create( **module_data ) unit_file = module.get('units', None) unit_list = UnitData.md_to_dict_from_file(module_file).get('learning_units') unit_obj_list = [] for unit in unit_list: unit_type = unit.get('type') lesson_or_quiz_obj = None if unit_type == 'lesson': lq_data = unit.pop('lesson') lq_data.update({'creator': user}) if lq_data.get('id'): # Lesson already exists lesson_or_quiz_obj = Lesson.objects.get(id=lq_data.get('id')) lesson_or_quiz_obj.__dict__.update(lq_data) lesson_or_quiz_obj.save() toc_file = lq_data.get('toc', None) if toc_file: with open(toc_file, 'r') as tocf: toc_data = ruamel.yaml.safe_load_all(tocf.read()) results = TableOfContents.objects.add_contents( course_data.get('id'), lesson_or_quiz_obj.id , user, toc_data) for status, msg in results: if status == False: raise Exception(msg) else: lesson_or_quiz_obj = Lesson.objects.create( **lq_data, ) unit['lesson'] = lesson_or_quiz_obj toc_file = lq_data.get('toc', None) if toc_file: with open(toc_file, 'r') as tocf: toc_data = ruamel.yaml.safe_load_all(tocf.read()) results = TableOfContents.objects.add_contents( course_data.get('id'), lesson_or_quiz_obj.id , user, toc_data) for status, msg in results: if status == False: raise Exception(msg) else: lq_data = unit.pop('quiz') lq_data.update({'creator': user}) if lq_data.get('id'): # Quiz already exists lesson_or_quiz_obj = Quiz.objects.get(id=lq_data.get('id')) lesson_or_quiz_obj.__dict__.update(lq_data) lesson_or_quiz_obj.save() else: lesson_or_quiz_obj = Quiz.objects.create( **lq_data, ) unit['quiz'] = lesson_or_quiz_obj if not lq_data.get('id'): unit_obj, unit_created = LearningUnit.objects.create( **unit ), True else: lesson_or_quiz_class_map = { 'lesson': Lesson, 'quiz': Quiz, } unit_created = False lq_obj_id = lq_data.get('id') lq_obj = lesson_or_quiz_class_map.get(unit_type).objects.get(id=lq_obj_id) m_units = module_obj.learning_unit.values_list('id', flat=True) lq_units = lq_obj.learningunit_set.values_list('id', flat=True) unit_id = list(set(m_units) & set(lq_units))[0] unit_obj = LearningUnit.objects.get(id=unit_id) unit_obj.__dict__.update({'order': unit.get('order', 0)}) unit_obj.save() if unit_created: unit_obj_list.append(unit_obj) module_obj.learning_unit.add(*unit_obj_list) if mod_created: module_obj_list.append(module_obj) if course_data.get('id', None): course_obj = Course.objects.get(id=course_data.get('id')) course_obj.__dict__.update(course_data) course_obj.save() else: course_data.update({'creator': user}) course_obj, course_created = Course.objects.create( **course_data ), True course_obj.learning_module.add(*module_obj_list) return course_obj def check_data(toc): course_file = toc.get('course') course_data = CourseData.md_to_dict(course_file) course_id = course_data.get('id') module_id_list = [] for data_elem in toc.get('modules', []): _file = data_elem.get('file') _data = UnitData.md_to_dict_from_file(_file) module_id = _data.get('module', None).get('id', None) if module_id: module_id_list.append(module_id) lesson_id_list = [] quiz_id_list = [] for unit in _data.get('learning_units'): unit_id = unit.get('id', None) unit_type = unit.get('type', None) if unit_id: if unit_type == 'lesson': lesson_id_list.append(unit_id) else: quiz_id_list.append(unit_id) try: if not has_relationship(module_id, 'learning_module', lesson_id_list, unit_type): msg = "Lesson IDs used in metadata do not belong to current course, Kindly inspect and reupload" return False, msg if not has_relationship(module_id, 'learning_module', quiz_id_list, unit_type): msg = "Quiz IDs used in metadata do not belong to current course, Kindly inspect and reupload" return False, msg except ObjectDoesNotExist: msg = "Object does not exist in DB" return False, msg try: if not has_relationship(course_id, 'course', module_id_list): msg = "Module IDs used in metadata do not belong to current course, Kindly inspect and reupload" return False, msg if has_duplicate_id(course_id, 'course', module_id_list): msg = "Modules metadata contains duplicate IDs, Kindly inspect and reupload" return False, msg except ObjectDoesNotExist: msg = "Object does not exist in DB" return False, msg return True, 'File check successful' def get_parent_child_data_from_db(parent_id, parent_type, child_id_list, child_type=None): if parent_type == 'learning_module': mod_obj = LearningModule.objects.get(id=parent_id) relationship_id_list = [ e for e in mod_obj.learning_unit.order_by( "order" ).values_list( child_type + '__id', flat=True, ) if e != None ] return relationship_id_list elif parent_type == 'course': course_obj = Course.objects.get(id=parent_id) relationship_id_list = course_obj.get_learning_modules().values_list( 'id', flat=True, ) return relationship_id_list def has_duplicate_id(parent_id, parent_type, child_id_list, child_type=None): relationship_id_list = get_parent_child_data_from_db( parent_id, parent_type, child_id_list, child_type ) if len(child_id_list) != len(set(relationship_id_list)): return True # duplicates exist else: return False # duplicates do not exist def has_relationship(parent_id, parent_type, child_id_list, child_type=None): relationship_id_list = get_parent_child_data_from_db( parent_id, parent_type, child_id_list, child_type ) for _id in child_id_list: if _id not in relationship_id_list: return False return True def read_toc(file): with open(file, 'r') as f: toc = yaml.load(f, Loader=yaml.FullLoader) return toc def upload_course(user): toc = read_toc('toc.yml') status, msg = check_data(toc) course_data = convert_md_to_dict(toc, user) return status, msg