# python standard library from contextlib import contextmanager # `with` statement in db sessions import logging from os import path, sys from datetime import datetime # user installed libraries import bcrypt from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker import networkx as nx # this project from models import Student, Answer, Topic, StudentTopic from knowledge import StudentKnowledge from factory import QFactory from tools import load_yaml # setup logger for this module logger = logging.getLogger(__name__) class LearnAppException(Exception): pass # ============================================================================ # LearnApp - application logic # ============================================================================ class LearnApp(object): def __init__(self, config_file): # state of online students self.online = {} config = load_yaml(config_file) # connect to database and checks for registered students self.db_setup(config['database']) # dependency graph shared by all students self.deps = build_dependency_graph(config) # add topics from dependency graph to the database, if missing self.db_add_missing_topics(self.deps.nodes()) # ------------------------------------------------------------------------ # login # ------------------------------------------------------------------------ def login(self, uid, pw): with self.db_session() as s: student = s.query(Student).filter(Student.id == uid).one_or_none() if student is None: logger.info(f'User "{uid}" does not exist!') return False # student does not exist if bcrypt.checkpw(pw.encode('utf-8'), student.password): if uid in self.online: logger.warning(f'User "{uid}" already logged in, overwriting state') tt = s.query(StudentTopic).filter(StudentTopic.student_id == uid) state = {t.topic_id: { 'level': t.level, 'date': datetime.strptime(t.date, "%Y-%m-%d %H:%M:%S.%f") } for t in tt} self.online[uid] = { 'number': student.id, 'name': student.name, 'state': StudentKnowledge(self.deps, state=state) } logger.info(f'User "{uid}" logged in successfully') return True else: logger.info(f'User "{uid}" wrong password!') return False # ------------------------------------------------------------------------ # logout # ------------------------------------------------------------------------ def logout(self, uid): del self.online[uid] logger.info(f'User "{uid}" logged out') # ------------------------------------------------------------------------ # change_password # ------------------------------------------------------------------------ def change_password(self, uid, pw): if not pw: return False with self.db_session() as s: u = s.query(Student).get(uid) u.password = bcrypt.hashpw(pw.encode('utf-8'), bcrypt.gensalt()) logger.info(f'User "{uid}" changed password') return True # ------------------------------------------------------------------------ # checks answer (updating student state) and returns grade. # ------------------------------------------------------------------------ def check_answer(self, uid, answer): knowledge = self.online[uid]['state'] grade = knowledge.check_answer(answer) if knowledge.get_current_question() is None: # finished topic, save into database finished_topic = knowledge.get_current_topic() level = knowledge.get_topic_level(finished_topic) date = str(knowledge.get_topic_date(finished_topic)) finished_questions = knowledge.get_finished_questions() logger.info(f'User "{uid}" finished "{finished_topic}"') with self.db_session(autoflush=False) as s: # save topic a = s.query(StudentTopic).filter_by(student_id=uid, topic_id=finished_topic).one_or_none() if a is None: # insert new studenttopic into database t = s.query(Topic).get(finished_topic) a = StudentTopic(level=level, date=date, topic=t) u = s.query(Student).get(uid) u.topics.append(a) else: # update studenttopic in database a.level = level a.date = date s.add(a) # save answered questions from finished_questions list s.add_all([ Answer( ref=q['ref'], grade=q['grade'], starttime=str(q['start_time']), finishtime=str(q['finish_time']), student_id=uid, topic_id=finished_topic) for q in finished_questions]) return grade # ------------------------------------------------------------------------ # Start new topic # ------------------------------------------------------------------------ def start_topic(self, uid, topic): try: ok = self.online[uid]['state'].init_topic(topic) except KeyError as e: logger.warning(f'User "{uid}" denied nonexistent "{topic}"') raise e else: if ok: logger.info(f'User "{uid}" started "{topic}"') else: logger.warning(f'User "{uid}" denied locked "{topic}"') return ok # ------------------------------------------------------------------------ # Fill db table 'Topic' with topics from the graph if not already there. # ------------------------------------------------------------------------ def db_add_missing_topics(self, nn): with self.db_session() as s: tt = [t[0] for t in s.query(Topic.id)] # db list of topics missing_topics = [Topic(id=n) for n in nn if n not in tt] if missing_topics: s.add_all(missing_topics) logger.info(f'Added {len(missing_topics)} new topics to the database') # ------------------------------------------------------------------------ # setup and check database # ------------------------------------------------------------------------ def db_setup(self, db): logger.info(f'Checking database "{db}":') engine = create_engine(f'sqlite:///{db}', echo=False) self.Session = sessionmaker(bind=engine) try: with self.db_session() as s: n = s.query(Student).count() m = s.query(Topic).count() q = s.query(Answer).count() except Exception as e: logger.critical(f'Database "{db}" not usable!') sys.exit(1) else: logger.info(f'{n:6} students') logger.info(f'{m:6} topics') logger.info(f'{q:6} answers') # ------------------------------------------------------------------------ # helper to manage db sessions using the `with` statement, for example # with self.db_session() as s: s.query(...) # ------------------------------------------------------------------------ @contextmanager def db_session(self, **kw): session = self.Session(**kw) try: yield session session.commit() except Exception as e: session.rollback() raise e finally: session.close() # ======================================================================== # methods that do not change state (pure functions) # ======================================================================== # ------------------------------------------------------------------------ def get_student_name(self, uid): return self.online[uid].get('name', '') # ------------------------------------------------------------------------ def get_student_state(self, uid): return self.online[uid]['state'].get_knowledge_state() # ------------------------------------------------------------------------ def get_student_progress(self, uid): return self.online[uid]['state'].get_topic_progress() # ------------------------------------------------------------------------ def get_student_question(self, uid): return self.online[uid]['state'].get_current_question() # dict # ------------------------------------------------------------------------ def get_student_question_type(self, uid): return self.online[uid]['state'].get_current_question()['type'] # ------------------------------------------------------------------------ def get_student_topic(self, uid): return self.online[uid]['state'].get_current_topic() # str # ------------------------------------------------------------------------ def get_title(self): return self.deps.graph['title'] # ------------------------------------------------------------------------ def get_topic_name(self, ref): return self.deps.node[ref]['name'] # ------------------------------------------------------------------------ def get_current_public_dir(self, uid): topic = self.online[uid]['state'].get_current_topic() p = self.deps.graph['path'] return path.join(p, topic, 'public') # ============================================================================ # Builds a digraph. # # First, topics such as `computer/mips/exceptions` are added as nodes # together with dependencies. Then, questions are loaded to a factory. # # g.graph['path'] base path where topic directories are located # g.graph['title'] title defined in the configuration YAML # g.graph['database'] sqlite3 database file to use # # Nodes are the topic references e.g. 'my/topic' # g.node['my/topic']['name'] name of the topic # g.node['my/topic']['questions'] list of question refs defined in YAML # g.node['my/topic']['factory'] dict with question factories # ---------------------------------------------------------------------------- def build_dependency_graph(config={}): logger.info('Building topic dependency graph.') # create graph prefix = config.get('path', '.') title = config.get('title', '') database = config.get('database', 'students.db') g = nx.DiGraph(path=prefix, title=title, database=database) # iterate over topics and build graph topics = config.get('topics', {}) for ref,attr in topics.items(): g.add_node(ref) if isinstance(attr, dict): g.node[ref]['name'] = attr.get('name', ref) g.node[ref]['questions'] = attr.get('questions', []) g.add_edges_from((d,ref) for d in attr.get('deps', [])) # iterate over topics and create question factories logger.info('Loading:') for tref in g.nodes(): tnode = g.node[tref] # current node (topic) fullpath = path.expanduser(path.join(prefix, tref)) filename = path.join(fullpath, 'questions.yaml') loaded_questions = load_yaml(filename, default=[]) # list # if questions not in configuration then load all, preserving order if not tnode['questions']: tnode['questions'] = [q['ref'] for q in loaded_questions] # make questions factory (without repeating same question) tnode['factory'] = {} for q in loaded_questions: if q['ref'] in tnode['questions']: q['path'] = fullpath tnode['factory'][q['ref']] = QFactory(q) logger.info(f'{len(tnode["questions"]):6} from {tref}') return g