186 lines
7.8 KiB
Python
186 lines
7.8 KiB
Python
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())
|
|
|
|
|