115 lines
4.3 KiB
Python
115 lines
4.3 KiB
Python
# ===========================================
|
||
# 班级操行分管理系统 - 后端服务
|
||
#
|
||
# 开发者: Canglan
|
||
# 联系方式: admin@sea-studio.top
|
||
# 版权归属: Sea Network Technology Studio
|
||
# 许可证: MIT License
|
||
#
|
||
# 版权所有 © Sea Network Technology Studio
|
||
# ===========================================
|
||
|
||
from utils.database import execute_one, execute_insert, execute_update
|
||
from utils.security import security
|
||
from utils.logger import get_logger
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
class UserModel:
|
||
"""用户数据模型"""
|
||
|
||
@staticmethod
|
||
async def get_by_username(username: str) -> dict:
|
||
"""根据用户名获取用户"""
|
||
sql = """
|
||
SELECT user_id, username, password_hash, real_name, user_type,
|
||
student_id, status, need_change_password, last_login_time, last_login_ip
|
||
FROM users
|
||
WHERE username = %s AND status = 1
|
||
"""
|
||
return await execute_one(sql, (username,))
|
||
|
||
@staticmethod
|
||
async def get_by_user_id(user_id: int) -> dict:
|
||
"""根据用户ID获取用户"""
|
||
sql = """
|
||
SELECT user_id, username, password_hash, real_name, user_type, student_id,
|
||
need_change_password, status
|
||
FROM users
|
||
WHERE user_id = %s
|
||
"""
|
||
return await execute_one(sql, (user_id,))
|
||
|
||
@staticmethod
|
||
async def create_student(username: str, password: str, real_name: str, student_id: int) -> int:
|
||
"""创建学生账号"""
|
||
password_hash = security.sha1_md5_password(password)
|
||
sql = """
|
||
INSERT INTO users (username, password_hash, real_name, user_type, student_id, need_change_password)
|
||
VALUES (%s, %s, %s, 'student', %s, 1)
|
||
"""
|
||
return await execute_insert(sql, (username, password_hash, real_name, student_id))
|
||
|
||
@staticmethod
|
||
async def create_parent(username: str, password: str, real_name: str, student_id: int) -> int:
|
||
"""创建家长账号"""
|
||
password_hash = security.sha1_md5_password(password)
|
||
sql = """
|
||
INSERT INTO users (username, password_hash, real_name, user_type, student_id, need_change_password)
|
||
VALUES (%s, %s, %s, 'parent', %s, 0)
|
||
"""
|
||
return await execute_insert(sql, (username, password_hash, real_name, student_id))
|
||
|
||
@staticmethod
|
||
async def create_admin(username: str, password: str, real_name: str) -> int:
|
||
"""创建管理员账号"""
|
||
password_hash = security.sha1_md5_password(password)
|
||
sql = """
|
||
INSERT INTO users (username, password_hash, real_name, user_type, need_change_password)
|
||
VALUES (%s, %s, %s, 'admin', 1)
|
||
"""
|
||
return await execute_insert(sql, (username, password_hash, real_name))
|
||
|
||
@staticmethod
|
||
async def update_password(user_id: int, new_password: str) -> bool:
|
||
"""更新密码"""
|
||
password_hash = security.sha1_md5_password(new_password)
|
||
sql = """
|
||
UPDATE users
|
||
SET password_hash = %s, need_change_password = 0
|
||
WHERE user_id = %s
|
||
"""
|
||
result = await execute_update(sql, (password_hash, user_id))
|
||
return result > 0
|
||
|
||
@staticmethod
|
||
async def update_last_login(user_id: int, ip: str) -> None:
|
||
"""更新最后登录信息"""
|
||
sql = """
|
||
UPDATE users
|
||
SET last_login_time = NOW(), last_login_ip = %s
|
||
WHERE user_id = %s
|
||
"""
|
||
await execute_update(sql, (ip, user_id))
|
||
|
||
@staticmethod
|
||
async def check_username_exists(username: str) -> bool:
|
||
"""检查用户名是否存在"""
|
||
sql = "SELECT 1 FROM users WHERE username = %s"
|
||
result = await execute_one(sql, (username,))
|
||
return result is not None
|
||
|
||
@staticmethod
|
||
async def update_status(user_id: int, status: int) -> bool:
|
||
"""更新用户状态（0=禁用，1=启用）"""
|
||
sql = "UPDATE users SET status = %s WHERE user_id = %s"
|
||
result = await execute_update(sql, (status, user_id))
|
||
return result > 0
|
||
|
||
@staticmethod
|
||
async def update_real_name(user_id: int, real_name: str) -> bool:
|
||
"""更新用户真实姓名"""
|
||
sql = "UPDATE users SET real_name = %s WHERE user_id = %s"
|
||
result = await execute_update(sql, (real_name, user_id))
|
||
return result > 0 |