# python standard library from contextlib import contextmanager # `with` statement in db sessions import logging from os import path, sys # user installed libraries try: import bcrypt from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker import networkx as nx import yaml except ImportError: logger.critical('Python package missing. See README.md for instructions.') sys.exit(1) # this project from models import Student, Answer, Topic, StudentTopic from knowledge import Knowledge from questions import QFactory from tools import load_yaml # setup logger for this module logger = logging.getLogger(__name__) # ============================================================================ # LearnApp - application logic # ============================================================================ class LearnApp(object): def __init__(self, conffile='demo/demo.yaml'): # online students self.online = {} # build dependency graph self.build_dependency_graph(conffile) # FIXME # connect to database and check registered students self.db_setup(self.depgraph.graph['database']) # FIXME # add topics from depgraph to the database self.db_add_topics() # ------------------------------------------------------------------------ def login(self, uid, try_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 or already loggeg in hashedtry = bcrypt.hashpw(try_pw.encode('utf-8'), student.password) if hashedtry != student.password: logger.info(f'User "{uid}" wrong password.') return False # wrong password # success tt = s.query(StudentTopic).filter(StudentTopic.student_id == uid) state = {t.topic_id: t.level for t in tt} self.online[uid] = { 'name': student.name, 'number': student.id, 'state': Knowledge(self.depgraph, state), } logger.info(f'User "{uid}" logged in') return True # ------------------------------------------------------------------------ # logout def logout(self, uid): state = self.online[uid]['state'].state # dict {node:level,...} # save state to database with self.db_session(autoflush=False) as s: # update existing associations and remove from state dict for a in s.query(StudentTopic).filter_by(student_id=uid): if a.topic_id in state: a.level = state.pop(a.topic_id) # update s.add(a) # insert the remaining ones u = s.query(Student).get(uid) for n,l in state.items(): a = StudentTopic(level=l) t = s.query(Topic).get(n) if t is None: # create if topic doesn't exist yet t = Topic(id=n) a.topic = t u.topics.append(a) s.add(a) del self.online[uid] logger.info(f'User "{uid}" logged out') # ------------------------------------------------------------------------ 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 # ------------------------------------------------------------------------ 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_title(self): return self.depgraph.graph['title'] # ------------------------------------------------------------------------ def get_topic_name(self, ref): return self.depgraph.node[ref]['name'] # ------------------------------------------------------------------------ def get_current_public_dir(self, uid): topic = self.online[uid]['state'].get_current_topic() p = self.depgraph.graph['path'] return path.join(p, topic, 'public') # ------------------------------------------------------------------------ # check answer and if correct returns new question, otherise returns None def check_answer(self, uid, answer): knowledge = self.online[uid]['state'] current_question = knowledge.check_answer(answer) if current_question is not None: logger.debug('check_answer: saving answer to db ...') with self.db_session() as s: s.add(Answer( ref=current_question['ref'], grade=current_question['grade'], starttime=str(current_question['start_time']), finishtime=str(current_question['finish_time']), student_id=uid)) s.commit() return knowledge.new_question() # ------------------------------------------------------------------------ # Receives a set of topics (strings like "math/algebra"), # and recursively adds dependencies to the dependency graph def build_dependency_graph(self, config_file): logger.debug(f'LearnApp.build_dependency_graph("{config_file}")') # Load configuration file try: with open(config_file, 'r') as f: logger.info(f'Loading configuration file "{config_file}"') config = yaml.load(f) except FileNotFoundError as e: logger.error(f'File not found: "{config_file}"') sys.exit(1) # config file parsed 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, list): # if prop is a list, we assume it's just a list of dependencies g.add_edges_from((d,ref) for d in attr) elif isinstance(attr, dict): g.node[ref]['name'] = attr.get('name', ref) g.add_edges_from((d,ref) for d in attr.get('deps', [])) elif isinstance(attr, str): g.node[ref]['name'] = attr # iterate over topics and create question factories for ref in g.nodes_iter(): g.node[ref].setdefault('name', ref) fullpath = path.expanduser(path.join(prefix, ref)) if path.isdir(fullpath): filename = path.join(fullpath, "questions.yaml") else: logger.error(f'build_dependency_graph: "{fullpath}" is not a directory') if path.isfile(filename): logger.info(f'Loading questions from "{filename}"') questions = load_yaml(filename, default=[]) for q in questions: q['path'] = fullpath g.node[ref]['factory'] = [QFactory(q) for q in questions] else: g.node[ref]['factory'] = [] logger.error(f'build_dependency_graph: "{filename}" does not exist') self.depgraph = g return g # ------------------------------------------------------------------------ def db_add_topics(self): with self.db_session() as s: tt = [t[0] for t in s.query(Topic.id)] # db list of topics nn = self.depgraph.nodes_iter() # topics in the graph s.add_all([Topic(id=n) for n in nn if n not in tt]) # ------------------------------------------------------------------------ # setup and check database def db_setup(self, db): logger.debug(f'LearnApp.db_setup("{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() except Exception as e: logger.critical(f'Database "{db}" not usable.') sys.exit(1) else: logger.info(f'Database has {n} students registered.') # ------------------------------------------------------------------------ # 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()