101 lines
3.7 KiB
Python
101 lines
3.7 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 |