From 916b0bac4a79fc8deffdc3e8a36f0f31fc5bb7ff Mon Sep 17 00:00:00 2001 From: canglan Date: Fri, 10 Apr 2026 16:05:56 +0800 Subject: [PATCH] =?UTF-8?q?v0.2=E7=99=BB=E5=BD=95=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/models/admin_role.py | 39 ++++++------ backend/routes/debug.py | 17 ++--- backend/services/admin_service.py | 101 +++++------------------------- backend/services/auth_service.py | 2 +- 4 files changed, 44 insertions(+), 115 deletions(-) diff --git a/backend/models/admin_role.py b/backend/models/admin_role.py index aa33835..723ac3a 100644 --- a/backend/models/admin_role.py +++ b/backend/models/admin_role.py @@ -1,5 +1,5 @@ # =========================================== -# 班级操行分管理系统 - 后端服务 +# 班级操行分管理系统 - 管理员角色模型 # # 开发者: Canglan # 联系方式: admin@sea-studio.top @@ -14,14 +14,13 @@ from utils.database import execute_one, execute_query, execute_insert, execute_u class AdminRoleModel: - """管理员角色数据模型""" + """管理员角色数据模型（无班级ID）""" @staticmethod async def get_by_user_id(user_id: int) -> Optional[Dict[str, Any]]: sql = """ - SELECT ar.*, c.class_name, s.subject_name + SELECT ar.*, s.subject_name 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 WHERE ar.user_id = %s LIMIT 1 @@ -29,32 +28,36 @@ class AdminRoleModel: return await execute_one(sql, (user_id,)) @staticmethod - async def get_class_id_by_user(user_id: int) -> Optional[int]: - 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]]: + async def get_all() -> List[Dict[str, Any]]: sql = """ - SELECT ar.*, u.real_name, u.username + SELECT ar.*, u.real_name, u.username, s.subject_name FROM admin_roles ar 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 """ - return await execute_query(sql, (class_id,)) + return await execute_query(sql) @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 = """ - INSERT INTO admin_roles (user_id, role_type, class_id, subject_id) - VALUES (%s, %s, %s, %s) + INSERT INTO admin_roles (user_id, role_type, subject_id) + 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 async def delete(user_id: int) -> bool: sql = "DELETE FROM admin_roles WHERE user_id = %s" result = await execute_update(sql, (user_id,)) + 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 \ No newline at end of file diff --git a/backend/routes/debug.py b/backend/routes/debug.py index a6aad12..cf3d54c 100644 --- a/backend/routes/debug.py +++ b/backend/routes/debug.py @@ -26,33 +26,28 @@ class AddAdminDebugRequest(BaseModel): username: str password: str real_name: str - role_type: str # 班主任/班长/学习委员/考勤委员/劳动委员 + role_type: str subject_id: Optional[int] = None @router.post(settings.DEBUG_PATH) async def debug_add_admin(request: Request, req: AddAdminDebugRequest): - """ - 调试入口 - 添加第一批管理员 - 注意：此接口仅用于首次部署，使用后建议注释掉此路由 - """ from models.user import UserModel - existing = await UserModel.get_by_username(req.username) - if existing: - return error_response(message="用户名已存在") - # 验证角色类型 valid_roles = ["班主任", "班长", "学习委员", "考勤委员", "劳动委员"] if req.role_type not in 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( username=req.username, real_name=req.real_name, password=req.password, role_type=req.role_type, - operator_id=0 # 系统添加 + operator_id=0 ) if result["success"]: diff --git a/backend/services/admin_service.py b/backend/services/admin_service.py index 352d6b9..c9ce11b 100644 --- a/backend/services/admin_service.py +++ b/backend/services/admin_service.py @@ -10,8 +10,7 @@ # =========================================== 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.student import StudentModel from models.admin_role import AdminRoleModel @@ -30,7 +29,7 @@ class AdminService: page_size: int = 20, search: str = None ) -> Dict[str, Any]: - """获取所有学生列表（单班级）""" + """获取所有学生列表""" offset = (page - 1) * page_size sql = """ @@ -73,10 +72,7 @@ class AdminService: operator_id: int, initial_points: int = 60 ) -> Dict[str, Any]: - """ - 批量导入学生（单班级版本） - 初始操行分默认60分 - """ + """批量导入学生""" results = [] success_count = 0 @@ -86,66 +82,32 @@ class AdminService: parent_phone = student.get("parent_phone", "").strip() password = student.get("password", "").strip() - # 验证必填字段 if not student_no or not name: - results.append({ - "student_no": student_no, - "success": False, - "error": "学号或姓名不能为空" - }) + results.append({"student_no": student_no, "success": False, "error": "学号或姓名不能为空"}) continue - # 验证学号格式 if not security.validate_student_no(student_no): - results.append({ - "student_no": student_no, - "success": False, - "error": "学号格式错误（4-20位字母数字组合）" - }) + results.append({"student_no": student_no, "success": False, "error": "学号格式错误"}) continue - # 验证手机号格式（如果有） if parent_phone and not security.validate_phone(parent_phone): - results.append({ - "student_no": student_no, - "success": False, - "error": "手机号格式错误" - }) + results.append({"student_no": student_no, "success": False, "error": "手机号格式错误"}) continue - # 检查学号是否已存在 existing = await StudentModel.get_by_student_no(student_no) if existing: - results.append({ - "student_no": student_no, - "success": False, - "error": "学号已存在" - }) + results.append({"student_no": student_no, "success": False, "error": "学号已存在"}) continue - # 设置初始密码 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_no=student_no, name=name, - class_id=1, # 单班级，固定为1 parent_phone=parent_phone if parent_phone else None, initial_points=initial_points ) - # 创建学生账号 await UserModel.create_student( username=student_no, password=init_password, @@ -153,37 +115,19 @@ class AdminService: student_id=student_id ) - # 创建家长账号（如果有手机号） if parent_phone: parent_exists = await UserModel.get_by_username(parent_phone) if not parent_exists: - parent_name = f"{name}家长" await UserModel.create_parent( username=parent_phone, password=init_password, - real_name=parent_name, + real_name=f"{name}家长", 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({ - "student_no": student_no, - "success": True, - "student_id": student_id, - "parent_phone": parent_phone if parent_phone else None - }) + results.append({"student_no": student_no, "success": True, "student_id": student_id}) success_count += 1 - - logger.info(f"用户[{operator_id}] 导入学生: {student_no} - {name} (初始分:{initial_points})") + logger.info(f"用户[{operator_id}] 导入学生: {student_no} - {name}") return { "success": True, @@ -202,29 +146,23 @@ class AdminService: initial_points: int = 60 ) -> Dict[str, Any]: """新增学生""" - # 验证学号格式 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): return {"success": False, "message": "手机号格式错误"} - # 检查学号是否已存在 existing = await StudentModel.get_by_student_no(student_no) if existing: return {"success": False, "message": "学号已存在"} - # 创建学生（初始操行分60分） student_id = await StudentModel.create( student_no=student_no, name=name, - class_id=1, # 单班级，固定为1 parent_phone=parent_phone if parent_phone else None, initial_points=initial_points ) - # 创建学生账号 await UserModel.create_student( username=student_no, password="123456", @@ -232,7 +170,6 @@ class AdminService: student_id=student_id ) - # 创建家长账号 if parent_phone: parent_exists = await UserModel.get_by_username(parent_phone) if not parent_exists: @@ -245,7 +182,7 @@ class AdminService: 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 async def add_admin( @@ -255,29 +192,24 @@ class AdminService: role_type: str, operator_id: int ) -> Dict[str, Any]: - """添加管理员（单班级）""" - # 检查用户名是否已存在 + """添加管理员""" existing = await UserModel.get_by_username(username) if existing: return {"success": False, "message": "用户名已存在"} - # 生成随机密码（如果未提供） if not password: password = security.generate_random_password() - # 创建管理员账号 user_id = await UserModel.create_admin( username=username, password=password, real_name=real_name ) - # 分配角色（班级ID固定为1） await AdminRoleModel.create( user_id=user_id, role_type=role_type, - class_id=1, # 单班级，固定为1 - subject_id=None # 单班级版本暂不支持科目关联 + subject_id=None ) logger.info(f"用户[{operator_id}] 添加管理员: {username} ({role_type})") @@ -292,7 +224,6 @@ class AdminService: @staticmethod async def get_admins() -> Dict[str, Any]: - """获取管理员列表（单班级）""" - admins = await AdminRoleModel.get_by_class(1) # 班级ID固定为1 - + """获取管理员列表""" + admins = await AdminRoleModel.get_all() return {"admins": admins} \ No newline at end of file diff --git a/backend/services/auth_service.py b/backend/services/auth_service.py index 01cf62a..a0b52a3 100644 --- a/backend/services/auth_service.py +++ b/backend/services/auth_service.py @@ -64,7 +64,7 @@ class AuthService: if user["user_type"] == "admin": admin_role = await AdminRoleModel.get_by_user_id(user["user_id"]) role = admin_role["role_type"] if admin_role else None - + # 生成Token token = jwt_handler.create_token( user_id=user["user_id"],