v0.2登录修复
This commit is contained in:
@@ -1,5 +1,5 @@
|
|||||||
# ===========================================
|
# ===========================================
|
||||||
# 班级操行分管理系统 - 后端服务
|
# 班级操行分管理系统 - 管理员角色模型
|
||||||
#
|
#
|
||||||
# 开发者: Canglan
|
# 开发者: Canglan
|
||||||
# 联系方式: admin@sea-studio.top
|
# 联系方式: admin@sea-studio.top
|
||||||
@@ -14,14 +14,13 @@ from utils.database import execute_one, execute_query, execute_insert, execute_u
|
|||||||
|
|
||||||
|
|
||||||
class AdminRoleModel:
|
class AdminRoleModel:
|
||||||
"""管理员角色数据模型"""
|
"""管理员角色数据模型(无班级ID)"""
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
async def get_by_user_id(user_id: int) -> Optional[Dict[str, Any]]:
|
async def get_by_user_id(user_id: int) -> Optional[Dict[str, Any]]:
|
||||||
sql = """
|
sql = """
|
||||||
SELECT ar.*, c.class_name, s.subject_name
|
SELECT ar.*, s.subject_name
|
||||||
FROM admin_roles ar
|
FROM admin_roles ar
|
||||||
LEFT JOIN classes c ON ar.class_id = c.class_id
|
|
||||||
LEFT JOIN subjects s ON ar.subject_id = s.subject_id
|
LEFT JOIN subjects s ON ar.subject_id = s.subject_id
|
||||||
WHERE ar.user_id = %s
|
WHERE ar.user_id = %s
|
||||||
LIMIT 1
|
LIMIT 1
|
||||||
@@ -29,32 +28,36 @@ class AdminRoleModel:
|
|||||||
return await execute_one(sql, (user_id,))
|
return await execute_one(sql, (user_id,))
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
async def get_class_id_by_user(user_id: int) -> Optional[int]:
|
async def get_all() -> List[Dict[str, Any]]:
|
||||||
sql = "SELECT class_id FROM admin_roles WHERE user_id = %s LIMIT 1"
|
|
||||||
result = await execute_one(sql, (user_id,))
|
|
||||||
return result["class_id"] if result else None
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
async def get_by_class(class_id: int) -> List[Dict[str, Any]]:
|
|
||||||
sql = """
|
sql = """
|
||||||
SELECT ar.*, u.real_name, u.username
|
SELECT ar.*, u.real_name, u.username, s.subject_name
|
||||||
FROM admin_roles ar
|
FROM admin_roles ar
|
||||||
JOIN users u ON ar.user_id = u.user_id
|
JOIN users u ON ar.user_id = u.user_id
|
||||||
WHERE ar.class_id = %s
|
LEFT JOIN subjects s ON ar.subject_id = s.subject_id
|
||||||
ORDER BY ar.role_type
|
ORDER BY ar.role_type
|
||||||
"""
|
"""
|
||||||
return await execute_query(sql, (class_id,))
|
return await execute_query(sql)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
async def create(user_id: int, role_type: str, class_id: int, subject_id: int = None) -> int:
|
async def create(user_id: int, role_type: str, subject_id: int = None) -> int:
|
||||||
sql = """
|
sql = """
|
||||||
INSERT INTO admin_roles (user_id, role_type, class_id, subject_id)
|
INSERT INTO admin_roles (user_id, role_type, subject_id)
|
||||||
VALUES (%s, %s, %s, %s)
|
VALUES (%s, %s, %s)
|
||||||
"""
|
"""
|
||||||
return await execute_insert(sql, (user_id, role_type, class_id, subject_id))
|
return await execute_insert(sql, (user_id, role_type, subject_id))
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
async def delete(user_id: int) -> bool:
|
async def delete(user_id: int) -> bool:
|
||||||
sql = "DELETE FROM admin_roles WHERE user_id = %s"
|
sql = "DELETE FROM admin_roles WHERE user_id = %s"
|
||||||
result = await execute_update(sql, (user_id,))
|
result = await execute_update(sql, (user_id,))
|
||||||
return result > 0
|
return result > 0
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
async def update_role(user_id: int, role_type: str, subject_id: int = None) -> bool:
|
||||||
|
sql = """
|
||||||
|
UPDATE admin_roles
|
||||||
|
SET role_type = %s, subject_id = %s
|
||||||
|
WHERE user_id = %s
|
||||||
|
"""
|
||||||
|
result = await execute_update(sql, (role_type, subject_id, user_id))
|
||||||
|
return result > 0
|
||||||
@@ -26,33 +26,28 @@ class AddAdminDebugRequest(BaseModel):
|
|||||||
username: str
|
username: str
|
||||||
password: str
|
password: str
|
||||||
real_name: str
|
real_name: str
|
||||||
role_type: str # 班主任/班长/学习委员/考勤委员/劳动委员
|
role_type: str
|
||||||
subject_id: Optional[int] = None
|
subject_id: Optional[int] = None
|
||||||
|
|
||||||
|
|
||||||
@router.post(settings.DEBUG_PATH)
|
@router.post(settings.DEBUG_PATH)
|
||||||
async def debug_add_admin(request: Request, req: AddAdminDebugRequest):
|
async def debug_add_admin(request: Request, req: AddAdminDebugRequest):
|
||||||
"""
|
|
||||||
调试入口 - 添加第一批管理员
|
|
||||||
注意:此接口仅用于首次部署,使用后建议注释掉此路由
|
|
||||||
"""
|
|
||||||
from models.user import UserModel
|
from models.user import UserModel
|
||||||
existing = await UserModel.get_by_username(req.username)
|
|
||||||
if existing:
|
|
||||||
return error_response(message="用户名已存在")
|
|
||||||
|
|
||||||
# 验证角色类型
|
|
||||||
valid_roles = ["班主任", "班长", "学习委员", "考勤委员", "劳动委员"]
|
valid_roles = ["班主任", "班长", "学习委员", "考勤委员", "劳动委员"]
|
||||||
if req.role_type not in valid_roles:
|
if req.role_type not in valid_roles:
|
||||||
return error_response(message=f"无效的角色类型,可选: {', '.join(valid_roles)}")
|
return error_response(message=f"无效的角色类型,可选: {', '.join(valid_roles)}")
|
||||||
|
|
||||||
# 创建管理员账号
|
existing = await UserModel.get_by_username(req.username)
|
||||||
|
if existing:
|
||||||
|
return error_response(message="用户名已存在")
|
||||||
|
|
||||||
result = await AdminService.add_admin(
|
result = await AdminService.add_admin(
|
||||||
username=req.username,
|
username=req.username,
|
||||||
real_name=req.real_name,
|
real_name=req.real_name,
|
||||||
password=req.password,
|
password=req.password,
|
||||||
role_type=req.role_type,
|
role_type=req.role_type,
|
||||||
operator_id=0 # 系统添加
|
operator_id=0
|
||||||
)
|
)
|
||||||
|
|
||||||
if result["success"]:
|
if result["success"]:
|
||||||
|
|||||||
@@ -10,8 +10,7 @@
|
|||||||
# ===========================================
|
# ===========================================
|
||||||
|
|
||||||
from typing import Dict, Any, List, Optional
|
from typing import Dict, Any, List, Optional
|
||||||
from datetime import datetime
|
from utils.database import execute_query, execute_one
|
||||||
|
|
||||||
from models.user import UserModel
|
from models.user import UserModel
|
||||||
from models.student import StudentModel
|
from models.student import StudentModel
|
||||||
from models.admin_role import AdminRoleModel
|
from models.admin_role import AdminRoleModel
|
||||||
@@ -30,7 +29,7 @@ class AdminService:
|
|||||||
page_size: int = 20,
|
page_size: int = 20,
|
||||||
search: str = None
|
search: str = None
|
||||||
) -> Dict[str, Any]:
|
) -> Dict[str, Any]:
|
||||||
"""获取所有学生列表(单班级)"""
|
"""获取所有学生列表"""
|
||||||
offset = (page - 1) * page_size
|
offset = (page - 1) * page_size
|
||||||
|
|
||||||
sql = """
|
sql = """
|
||||||
@@ -73,10 +72,7 @@ class AdminService:
|
|||||||
operator_id: int,
|
operator_id: int,
|
||||||
initial_points: int = 60
|
initial_points: int = 60
|
||||||
) -> Dict[str, Any]:
|
) -> Dict[str, Any]:
|
||||||
"""
|
"""批量导入学生"""
|
||||||
批量导入学生(单班级版本)
|
|
||||||
初始操行分默认60分
|
|
||||||
"""
|
|
||||||
results = []
|
results = []
|
||||||
success_count = 0
|
success_count = 0
|
||||||
|
|
||||||
@@ -86,66 +82,32 @@ class AdminService:
|
|||||||
parent_phone = student.get("parent_phone", "").strip()
|
parent_phone = student.get("parent_phone", "").strip()
|
||||||
password = student.get("password", "").strip()
|
password = student.get("password", "").strip()
|
||||||
|
|
||||||
# 验证必填字段
|
|
||||||
if not student_no or not name:
|
if not student_no or not name:
|
||||||
results.append({
|
results.append({"student_no": student_no, "success": False, "error": "学号或姓名不能为空"})
|
||||||
"student_no": student_no,
|
|
||||||
"success": False,
|
|
||||||
"error": "学号或姓名不能为空"
|
|
||||||
})
|
|
||||||
continue
|
continue
|
||||||
|
|
||||||
# 验证学号格式
|
|
||||||
if not security.validate_student_no(student_no):
|
if not security.validate_student_no(student_no):
|
||||||
results.append({
|
results.append({"student_no": student_no, "success": False, "error": "学号格式错误"})
|
||||||
"student_no": student_no,
|
|
||||||
"success": False,
|
|
||||||
"error": "学号格式错误(4-20位字母数字组合)"
|
|
||||||
})
|
|
||||||
continue
|
continue
|
||||||
|
|
||||||
# 验证手机号格式(如果有)
|
|
||||||
if parent_phone and not security.validate_phone(parent_phone):
|
if parent_phone and not security.validate_phone(parent_phone):
|
||||||
results.append({
|
results.append({"student_no": student_no, "success": False, "error": "手机号格式错误"})
|
||||||
"student_no": student_no,
|
|
||||||
"success": False,
|
|
||||||
"error": "手机号格式错误"
|
|
||||||
})
|
|
||||||
continue
|
continue
|
||||||
|
|
||||||
# 检查学号是否已存在
|
|
||||||
existing = await StudentModel.get_by_student_no(student_no)
|
existing = await StudentModel.get_by_student_no(student_no)
|
||||||
if existing:
|
if existing:
|
||||||
results.append({
|
results.append({"student_no": student_no, "success": False, "error": "学号已存在"})
|
||||||
"student_no": student_no,
|
|
||||||
"success": False,
|
|
||||||
"error": "学号已存在"
|
|
||||||
})
|
|
||||||
continue
|
continue
|
||||||
|
|
||||||
# 设置初始密码
|
|
||||||
init_password = password if password else "123456"
|
init_password = password if password else "123456"
|
||||||
|
|
||||||
# 验证密码强度(可选)
|
|
||||||
is_valid, msg = security.validate_password_strength(init_password)
|
|
||||||
if not is_valid:
|
|
||||||
results.append({
|
|
||||||
"student_no": student_no,
|
|
||||||
"success": False,
|
|
||||||
"error": f"密码不符合要求: {msg}"
|
|
||||||
})
|
|
||||||
continue
|
|
||||||
|
|
||||||
# 创建学生(初始操行分60分)
|
|
||||||
student_id = await StudentModel.create(
|
student_id = await StudentModel.create(
|
||||||
student_no=student_no,
|
student_no=student_no,
|
||||||
name=name,
|
name=name,
|
||||||
class_id=1, # 单班级,固定为1
|
|
||||||
parent_phone=parent_phone if parent_phone else None,
|
parent_phone=parent_phone if parent_phone else None,
|
||||||
initial_points=initial_points
|
initial_points=initial_points
|
||||||
)
|
)
|
||||||
|
|
||||||
# 创建学生账号
|
|
||||||
await UserModel.create_student(
|
await UserModel.create_student(
|
||||||
username=student_no,
|
username=student_no,
|
||||||
password=init_password,
|
password=init_password,
|
||||||
@@ -153,37 +115,19 @@ class AdminService:
|
|||||||
student_id=student_id
|
student_id=student_id
|
||||||
)
|
)
|
||||||
|
|
||||||
# 创建家长账号(如果有手机号)
|
|
||||||
if parent_phone:
|
if parent_phone:
|
||||||
parent_exists = await UserModel.get_by_username(parent_phone)
|
parent_exists = await UserModel.get_by_username(parent_phone)
|
||||||
if not parent_exists:
|
if not parent_exists:
|
||||||
parent_name = f"{name}家长"
|
|
||||||
await UserModel.create_parent(
|
await UserModel.create_parent(
|
||||||
username=parent_phone,
|
username=parent_phone,
|
||||||
password=init_password,
|
password=init_password,
|
||||||
real_name=parent_name,
|
real_name=f"{name}家长",
|
||||||
student_id=student_id
|
student_id=student_id
|
||||||
)
|
)
|
||||||
else:
|
|
||||||
# 手机号已被占用
|
|
||||||
results.append({
|
|
||||||
"student_no": student_no,
|
|
||||||
"success": True,
|
|
||||||
"student_id": student_id,
|
|
||||||
"warning": f"家长手机号 {parent_phone} 已被其他账号使用,未创建家长账号"
|
|
||||||
})
|
|
||||||
success_count += 1
|
|
||||||
continue
|
|
||||||
|
|
||||||
results.append({
|
results.append({"student_no": student_no, "success": True, "student_id": student_id})
|
||||||
"student_no": student_no,
|
|
||||||
"success": True,
|
|
||||||
"student_id": student_id,
|
|
||||||
"parent_phone": parent_phone if parent_phone else None
|
|
||||||
})
|
|
||||||
success_count += 1
|
success_count += 1
|
||||||
|
logger.info(f"用户[{operator_id}] 导入学生: {student_no} - {name}")
|
||||||
logger.info(f"用户[{operator_id}] 导入学生: {student_no} - {name} (初始分:{initial_points})")
|
|
||||||
|
|
||||||
return {
|
return {
|
||||||
"success": True,
|
"success": True,
|
||||||
@@ -202,29 +146,23 @@ class AdminService:
|
|||||||
initial_points: int = 60
|
initial_points: int = 60
|
||||||
) -> Dict[str, Any]:
|
) -> Dict[str, Any]:
|
||||||
"""新增学生"""
|
"""新增学生"""
|
||||||
# 验证学号格式
|
|
||||||
if not security.validate_student_no(student_no):
|
if not security.validate_student_no(student_no):
|
||||||
return {"success": False, "message": "学号格式错误(4-20位字母数字组合)"}
|
return {"success": False, "message": "学号格式错误"}
|
||||||
|
|
||||||
# 验证手机号格式(如果有)
|
|
||||||
if parent_phone and not security.validate_phone(parent_phone):
|
if parent_phone and not security.validate_phone(parent_phone):
|
||||||
return {"success": False, "message": "手机号格式错误"}
|
return {"success": False, "message": "手机号格式错误"}
|
||||||
|
|
||||||
# 检查学号是否已存在
|
|
||||||
existing = await StudentModel.get_by_student_no(student_no)
|
existing = await StudentModel.get_by_student_no(student_no)
|
||||||
if existing:
|
if existing:
|
||||||
return {"success": False, "message": "学号已存在"}
|
return {"success": False, "message": "学号已存在"}
|
||||||
|
|
||||||
# 创建学生(初始操行分60分)
|
|
||||||
student_id = await StudentModel.create(
|
student_id = await StudentModel.create(
|
||||||
student_no=student_no,
|
student_no=student_no,
|
||||||
name=name,
|
name=name,
|
||||||
class_id=1, # 单班级,固定为1
|
|
||||||
parent_phone=parent_phone if parent_phone else None,
|
parent_phone=parent_phone if parent_phone else None,
|
||||||
initial_points=initial_points
|
initial_points=initial_points
|
||||||
)
|
)
|
||||||
|
|
||||||
# 创建学生账号
|
|
||||||
await UserModel.create_student(
|
await UserModel.create_student(
|
||||||
username=student_no,
|
username=student_no,
|
||||||
password="123456",
|
password="123456",
|
||||||
@@ -232,7 +170,6 @@ class AdminService:
|
|||||||
student_id=student_id
|
student_id=student_id
|
||||||
)
|
)
|
||||||
|
|
||||||
# 创建家长账号
|
|
||||||
if parent_phone:
|
if parent_phone:
|
||||||
parent_exists = await UserModel.get_by_username(parent_phone)
|
parent_exists = await UserModel.get_by_username(parent_phone)
|
||||||
if not parent_exists:
|
if not parent_exists:
|
||||||
@@ -245,7 +182,7 @@ class AdminService:
|
|||||||
|
|
||||||
logger.info(f"用户[{operator_id}] 新增学生: {student_no} - {name}")
|
logger.info(f"用户[{operator_id}] 新增学生: {student_no} - {name}")
|
||||||
|
|
||||||
return {"success": True, "student_id": student_id, "student_no": student_no, "name": name}
|
return {"success": True, "student_id": student_id}
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
async def add_admin(
|
async def add_admin(
|
||||||
@@ -255,29 +192,24 @@ class AdminService:
|
|||||||
role_type: str,
|
role_type: str,
|
||||||
operator_id: int
|
operator_id: int
|
||||||
) -> Dict[str, Any]:
|
) -> Dict[str, Any]:
|
||||||
"""添加管理员(单班级)"""
|
"""添加管理员"""
|
||||||
# 检查用户名是否已存在
|
|
||||||
existing = await UserModel.get_by_username(username)
|
existing = await UserModel.get_by_username(username)
|
||||||
if existing:
|
if existing:
|
||||||
return {"success": False, "message": "用户名已存在"}
|
return {"success": False, "message": "用户名已存在"}
|
||||||
|
|
||||||
# 生成随机密码(如果未提供)
|
|
||||||
if not password:
|
if not password:
|
||||||
password = security.generate_random_password()
|
password = security.generate_random_password()
|
||||||
|
|
||||||
# 创建管理员账号
|
|
||||||
user_id = await UserModel.create_admin(
|
user_id = await UserModel.create_admin(
|
||||||
username=username,
|
username=username,
|
||||||
password=password,
|
password=password,
|
||||||
real_name=real_name
|
real_name=real_name
|
||||||
)
|
)
|
||||||
|
|
||||||
# 分配角色(班级ID固定为1)
|
|
||||||
await AdminRoleModel.create(
|
await AdminRoleModel.create(
|
||||||
user_id=user_id,
|
user_id=user_id,
|
||||||
role_type=role_type,
|
role_type=role_type,
|
||||||
class_id=1, # 单班级,固定为1
|
subject_id=None
|
||||||
subject_id=None # 单班级版本暂不支持科目关联
|
|
||||||
)
|
)
|
||||||
|
|
||||||
logger.info(f"用户[{operator_id}] 添加管理员: {username} ({role_type})")
|
logger.info(f"用户[{operator_id}] 添加管理员: {username} ({role_type})")
|
||||||
@@ -292,7 +224,6 @@ class AdminService:
|
|||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
async def get_admins() -> Dict[str, Any]:
|
async def get_admins() -> Dict[str, Any]:
|
||||||
"""获取管理员列表(单班级)"""
|
"""获取管理员列表"""
|
||||||
admins = await AdminRoleModel.get_by_class(1) # 班级ID固定为1
|
admins = await AdminRoleModel.get_all()
|
||||||
|
|
||||||
return {"admins": admins}
|
return {"admins": admins}
|
||||||
Reference in New Issue
Block a user