from flask import current_app from bson.binary import Binary import uuid import os # Example usage: import json from datetime import datetime import copy from ..models.material import Material class User: def __init__(self, user_uuid, username, teacher, password, select_course, head_icon, _id, **kwargs): self.user_uuid = user_uuid # User UUID self.username = username self.teacher = teacher self.password = password self.select_course = select_course # List of selected courses self.head_icon = head_icon # URL of the user's head icon self._id = _id self.kwargs = kwargs self.mongo = current_app.extensions["mongo"] def save_chapter_memory(self, course_id, lesson_name, subchapter_title, mem_list, score, is_rebuttal): ''' Save the memory of a chapter into the mongo db ''' try: cp = self.mongo.db.course_process.find_one({"user_uuid": self.user_uuid, "material_id": course_id}) if not cp: return cp['course_last_study_date'] = datetime.now().strftime('%Y-%m-%d') have_lesson = False for _ in range(len(cp['lesson_processs'])): lesson_process, cname, lname = cp['lesson_processs'][_] if lname == lesson_name: have_lesson = True saveed = False for i in range(len(lesson_process)): step = lesson_process[i] if step['title'] == subchapter_title: step['dialog'] = mem_list if is_rebuttal: step['rebuttal_score'] = score else: step['score'] = score step['is_rebuttal'] = is_rebuttal saveed = True break if not saveed: cp['lesson_processs'].append({ 'title': subchapter_title, 'dialog': mem_list, 'score': score, 'is_rebuttal': is_rebuttal, 'rebuttal_score': score if is_rebuttal else 0, }) break if not have_lesson: cp['lesson_processs'].append(([], chapter_name, lesson_name)) self.mongo.db.course_process.update_one({"user_uuid": self.user_uuid, "material_id": course_id}, {"$set": cp}) except Exception as e: print("=-=-=-=-=-=") print(e) def get_all_course_progress(self): course_processes = self.mongo.db.course_process.find({"user_uuid": self.user_uuid}) for cp in course_processes: print(f"Course: {cp['material_id']}") for lessons,chapter_name,lesson_name in cp["lesson_processs"]: print(f" Lesson ID: {lesson_name}") for step in lessons: print(f" Chapter Dialog: {step['dialog']}") print(f" Chapter Score: {step['score']}") print(f" Rebuttal: {'Yes' if step['is_rebuttal'] else 'No'}") if step['is_rebuttal']: print(f" Rebuttal Score: {step['rebuttal_score']}") def get_course_progress(self, course_id): cp = self.mongo.db.course_process.find_one({"user_uuid": self.user_uuid, "material_id": course_id}) return cp def set_course_progress(self, chat_historys, step_scores, material_id, chapter_name, lesson_name): ''' chat_historys: ['', '', ...] 用户对话历史（包括函数调用，也就是右侧栏用户看到的内容） step_scores: [10, 20, ...] 各个步骤的得分 ''' try: cl = self.mongo.db.course_process document = cl.find_one({ 'user_uuid': self.user_uuid, 'material_id': material_id }) if not document: cl.insert_one({ 'user_uuid': self.user_uuid, 'material_id': material_id, 'lesson_processs': [], 'course_put_in_date': datetime.now().strftime('%Y-%m-%d'), 'course_last_study_date': datetime.now().strftime('%Y-%m-%d'), 'course_total_study_time': 0 }) document = cl.find_one({ 'user_uuid': self.user_uuid, 'material_id': material_id }) lesson_processes = document.get('lesson_processs', []) finded = False for process in lesson_processes: if process[1]==chapter_name and process[2]==lesson_name: process[0] = chat_historys if len(process)<=3: process.append(step_scores) else: process[3] = step_scores finded = True if not finded: lesson_processes.append((chat_historys, chapter_name, lesson_name, step_scores)) cl.update_one({'user_uuid': self.user_uuid, 'material_id': material_id}, {'$set': {'lesson_processs': lesson_processes}}) except Exception as e: print("=-=-=-=-=-=") print(e) def select_new_course(self, course_id, course:Material): try: if course_id in self.select_course: return True self.select_course.append(course_id) self.mongo.db.course_process.insert_one({ 'user_uuid': self.user_uuid, 'material_id': course_id, 'lesson_processs':[], 'course_put_in_date':datetime.now().strftime('%Y-%m-%d'), 'course_last_study_date':datetime.now().strftime('%Y-%m-%d'), 'course_total_study_time':0}) for chapter in course.chapters: for lesson in chapter.lessons: self.mongo.db.course_process.update_one({"user_uuid": self.user_uuid, "material_id": course_id}, {"$push": {"lesson_processs": ([], chapter.chapter_name, lesson.lesson_name)}}) self.select_course.append(course_id) self.mongo.db.users.update_one({"user_uuid": self.user_uuid}, {"$push": {"select_course": course_id}}) except Exception as e: print("=-=-=-=-=-=") print(e) return False return True def save_to_db(self): current_app.extensions["mongo"].db.users.update_one({"user_uuid": self.user_uuid}, {"$set": self.__json__()}) def __json__(self): return { 'username': self.username, 'teacher': self.teacher, 'select_course': self.select_course, 'head_icon': self.head_icon, } def to_json(self): return json.dumps(self.__json__(), indent=4) class UserList: def __init__(self): """从mongo中获取用户信息""" self.Users = {} def add_user(self, username, password, teacher=False): """添加新的用户登录信息""" # 检查是否已经有该用户 if current_app.extensions["mongo"].db.users.find_one({"username": username}) is None: current_app.extensions["mongo"].db.users.insert_one({"user_uuid": Binary.from_uuid(uuid.uuid4()), "username": username, "teacher": teacher, "password": password, "select_course": [], "head_icon": ""}) else: print(f"User with ID {username} already exists.") def get_user_by_name(self, username): """根据用户ID获取用户信息""" if username not in self.Users: self.Users[username] = User(**current_app.extensions["mongo"].db.users.find_one({"username": username})) return self.Users[username] else: return self.Users[username] def has_user(self, username): """检查用户是否存在""" return current_app.extensions["mongo"].db.users.find_one({"username": username}) is not None def get_user_is_teacher(self, username): """获取指定 user_id 的用户是否为教师""" if current_app.extensions["mongo"].db.users.find_one({"username": username}) is not None: result = current_app.extensions["mongo"].db.users.find_one({"username": username}) return result['teacher'] return None def get_user_pswd(self, username): """获取指定 user_id 的用户信息""" mongo = current_app.extensions["mongo"] if mongo.db.users.find_one({"username": username}) is not None: result = mongo.db.users.find_one({"username": username}) return result['password'] return None def update_password(self, username, new_password): """更新用户的密码""" mongo = current_app.extensions["mongo"] assert mongo.db.users.find_one({"username": username}) is not None, f"User with ID {username} does not exist." mongo.db.users.update_one({"username": username}, {"$set": {"password": new_password, "teacher": mongo.db.users.find_one({"username": username})['teacher']}}) def delete_user(self, username): """删除用户""" mongo = current_app.extensions["mongo"] assert mongo.db.users.find_one({"username": username}) is not None, f"User with ID {username} does not exist." mongo.db.users.delete_one({"username": username}) def get_all_users(self): """获取所有用户的信息""" mongo = current_app.extensions["mongo"] return list(mongo.db.users.find())