学生端选课进课闭环

Html/apps/services/user_db_service.py

@@ -0,0 +1,185 @@
+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
+    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:
+            mongo = current_app.extensions["mongo"]
+            cp = 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))
+            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):
+        mongo = current_app.extensions["mongo"]
+        course_processes = 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):
+        mongo = current_app.extensions["mongo"]
+        cp = mongo.db.course_process.find_one({"user_uuid": self.user_uuid, "material_id": course_id})
+        return cp
+
+    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)
+            mongo = current_app.extensions["mongo"]
+            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:
+                    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)
+            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):
+        """添加新的用户登录信息"""
+        # 检查是否已经有该用户
+        mongo = current_app.extensions["mongo"]
+        if mongo.db.users.find_one({"username": username}) is None:
+            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获取用户信息"""
+        mongo = current_app.extensions["mongo"]
+        if username not in self.Users:
+            self.Users[username] = User(**mongo.db.users.find_one({"username": username}))
+            return self.Users[username]
+        else:
+            return self.Users[username]
+    
+    def has_user(self, username):
+        """检查用户是否存在"""
+        mongo = current_app.extensions["mongo"]
+        return mongo.db.users.find_one({"username": username}) is not None
+    
+    def get_user_is_teacher(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['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):
+        """更新用户的密码"""
+        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):
+        """删除用户"""
+        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())
+
+