128 lines
4.4 KiB
Python
128 lines
4.4 KiB
Python
from db.connection import MySQLPool
|
||
import bcrypt
|
||
|
||
|
||
class DatabaseManager:
|
||
def __init__(self):
|
||
# 使用MySQLPool初始化数据库连接池
|
||
self.pool = MySQLPool()
|
||
|
||
def fetch(self, query, params=None):
|
||
conn = self.pool.get_connection()
|
||
try:
|
||
cursor = conn.cursor()
|
||
cursor.execute(query, params or ())
|
||
result = cursor.fetchall()
|
||
return result
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
def execute(self, query, params=None):
|
||
conn = self.pool.get_connection()
|
||
try:
|
||
cursor = conn.cursor()
|
||
cursor.execute(query, params or ())
|
||
conn.commit()
|
||
return cursor.rowcount
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
def user_exists(self, phone_number):
|
||
sql = "SELECT 1 FROM user WHERE number=%s LIMIT 1"
|
||
result = self.fetch(sql, (phone_number,))
|
||
return len(result) > 0
|
||
|
||
def insert_user(self, user):
|
||
sql = """
|
||
INSERT INTO user (name, number, password, status)
|
||
VALUES (%s, %s, %s, %s)
|
||
"""
|
||
data = (user.name, user.number, user.password, user.status)
|
||
# print(data)
|
||
return self.execute(sql, data)
|
||
|
||
def valid_login(self, number, password_attempt):
|
||
# SQL查询获取用户的哈希密码，身份和状态
|
||
sql = "SELECT password, status,name FROM user WHERE number=%s LIMIT 1"
|
||
result = self.fetch(sql, (number,))
|
||
print(result)
|
||
if result:
|
||
stored_hash = result[0]['password'] # 假设结果是密码字段
|
||
status = result[0]['status'] # 用户状态
|
||
name = result[0]['name']
|
||
|
||
# 使用bcrypt进行密码验证
|
||
if bcrypt.checkpw(password_attempt.encode('utf-8'), stored_hash.encode('utf-8')):
|
||
# 密码匹配，返回登录成功，身份和状态
|
||
return {'valid': True, 'status': status, 'name': name}
|
||
|
||
# 密码不匹配或用户不存在，返回登录失败
|
||
return {'valid': False}
|
||
|
||
def get_menu(self, role):
|
||
sql = "SELECT menu_name,path FROM menu_items WHERE role=%s ORDER BY `order`"
|
||
result = self.fetch(sql, (role,))
|
||
return result
|
||
|
||
def get_all_courses(self):
|
||
sql = "SELECT course_name, course_code, course_type, credits, description FROM course"
|
||
result = self.fetch(sql)
|
||
return result
|
||
|
||
def get_current_teacher_courses(self, number):
|
||
# 根据电话号码查询user_id
|
||
user_sql = "SELECT user_id FROM user WHERE number = %s"
|
||
user_result = self.fetch(user_sql, (number,))
|
||
|
||
if user_result:
|
||
user_id = user_result[0]['user_id']
|
||
|
||
# 使用联合查询获取该老师的所有课程信息及班级信息
|
||
course_class_sql = """
|
||
SELECT c.course_name, c.course_code, c.credits, cl.class_name, cl.major
|
||
FROM class_courses cc
|
||
JOIN courses c ON cc.course_id = c.course_id
|
||
JOIN classes cl ON cc.class_id = cl.class_id
|
||
WHERE cc.user_id = %s
|
||
"""
|
||
courses_classes_result = self.fetch(course_class_sql, (user_id,))
|
||
|
||
# 组织和返回信息
|
||
if courses_classes_result:
|
||
# print(courses_classes_result)
|
||
return courses_classes_result
|
||
return []
|
||
|
||
def get_course_type(self):
|
||
sql = "SELECT course_name, course_type FROM course"
|
||
result = self.fetch(sql)
|
||
return result
|
||
|
||
def get_announcement_info(self):
|
||
sql = "SELECT course_name, course_type FROM course"
|
||
result = self.fetch(sql)
|
||
return result
|
||
|
||
def insert_teacher(self, teacher):
|
||
sql = "INSERT INTO teacher (name, user_id,teacher_number) VALUES (%s, %s, %s);"
|
||
|
||
data = (teacher.name, teacher.user_id,teacher.teacher_number)
|
||
print(data)
|
||
return self.execute(sql, data)
|
||
|
||
def query_user_id(self, phone_number):
|
||
sql = "SELECT user_id from user WHERE number = %s;"
|
||
data = self.fetch(sql, phone_number)
|
||
|
||
return data[0]['user_id']
|
||
|
||
def insert_student(self, student):
|
||
sql = """
|
||
INSERT INTO student (student_name, student_number, user_id, major_id, class_name)
|
||
VALUES (%s, %s, %s, %s, %s)
|
||
"""
|
||
data = (student.student_name, student.student_number, student.user_id, student.major_id, student.class_name)
|
||
return self.execute(sql, data)
|