canglan/SharedClassManager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.2测试修复

Browse Source
This commit is contained in:
canglan
2026-04-10 15:01:21 +08:00
parent 9d89e62b63
commit 81ea44fbab
2 changed files with 78 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
backend/middleware/permission.py
View File

@@ -12,10 +12,16 @@
from fastapi import Request
from typing import List, Optional, Callable, Dict, Any
from functools import wraps
from utils.response import forbidden_response
from utils.database import execute_one
from utils.logger import get_logger
logger = get_logger(__name__)
async def get_current_user(request: Request) -> Dict[str, Any]:
"""获取当前登录用户信息"""
return {
"user_id": getattr(request.state, 'user_id', None),
"username": getattr(request.state, 'username', None),
@@ -24,48 +30,66 @@ async def get_current_user(request: Request) -> Dict[str, Any]:
"role": getattr(request.state, 'role', None)
}
async def get_current_user_id(request: Request) -> int:
"""获取当前用户ID"""
return getattr(request.state, 'user_id', None)
class PermissionChecker:
"""权限检查器"""
@staticmethod
async def get_user_role(user_id: int) -> Optional[str]:
"""获取用户的管理员角色"""
sql = "SELECT role_type FROM admin_roles WHERE user_id = %s LIMIT 1"
result = await execute_one(sql, (user_id,))
return result["role_type"] if result else None
@staticmethod
async def check_is_teacher(user_id: int) -> bool:
"""检查是否为班主任"""
role = await PermissionChecker.get_user_role(user_id)
return role == "班主任"
@staticmethod
async def check_is_monitor(user_id: int) -> bool:
"""检查是否为班长"""
role = await PermissionChecker.get_user_role(user_id)
return role == "班长"
@staticmethod
async def check_is_study_commissioner(user_id: int) -> bool:
"""检查是否为学习委员"""
role = await PermissionChecker.get_user_role(user_id)
return role == "学习委员"
@staticmethod
async def check_is_attendance_rep(user_id: int) -> bool:
"""检查是否为考勤委员"""
role = await PermissionChecker.get_user_role(user_id)
return role == "考勤委员"
@staticmethod
async def check_is_labor_rep(user_id: int) -> bool:
"""检查是否为劳动委员"""
role = await PermissionChecker.get_user_role(user_id)
return role == "劳动委员"
@staticmethod
async def check_can_manage_subjects(user_id: int) -> bool:
"""检查是否可以管理科目（班主任或学习委员）"""
role = await PermissionChecker.get_user_role(user_id)
return role in ["班主任", "学习委员"]
@staticmethod
async def check_can_revoke(user_id: int, record_id: int) -> bool:
"""
检查是否可以撤销扣分记录
班主任：可以撤销任何记录
班长：可以撤销任何记录
其他：只能撤销自己的记录
"""
sql = "SELECT recorder_id FROM conduct_records WHERE record_id = %s"
record = await execute_one(sql, (record_id,))
if not record:
@@ -75,7 +99,9 @@ class PermissionChecker:
return True
return record["recorder_id"] == user_id
def require_auth(func: Callable):
"""需要认证的装饰器"""
@wraps(func)
async def wrapper(*args, **kwargs):
request = kwargs.get('request')
@@ -84,7 +110,9 @@ def require_auth(func: Callable):
return await func(*args, **kwargs)
return wrapper
def require_role(roles: List[str]):
"""需要特定角色的装饰器"""
def decorator(func: Callable):
@wraps(func)
async def wrapper(*args, **kwargs):
@@ -99,24 +127,44 @@ def require_role(roles: List[str]):
return wrapper
return decorator
def require_teacher(func: Callable):
"""需要班主任权限的装饰器"""
@wraps(func)
async def wrapper(*args, **kwargs):
request = kwargs.get('request')
if not request or not hasattr(request.state, 'user_id'):
return forbidden_response("请先登录")
if not await PermissionChecker.check_is_teacher(request.state.user_id):
is_teacher = await PermissionChecker.check_is_teacher(request.state.user_id)
if not is_teacher:
return forbidden_response("需要班主任权限")
return await func(*args, **kwargs)
return wrapper
def require_study_commissioner(func: Callable):
def require_monitor(func: Callable):
"""需要班长权限的装饰器"""
@wraps(func)
async def wrapper(*args, **kwargs):
request = kwargs.get('request')
if not request or not hasattr(request.state, 'user_id'):
return forbidden_response("请先登录")
if not await PermissionChecker.check_is_study_commissioner(request.state.user_id):
is_monitor = await PermissionChecker.check_is_monitor(request.state.user_id)
if not is_monitor:
return forbidden_response("需要班长权限")
return await func(*args, **kwargs)
return wrapper
def require_study_commissioner(func: Callable):
"""需要学习委员权限的装饰器"""
@wraps(func)
async def wrapper(*args, **kwargs):
request = kwargs.get('request')
if not request or not hasattr(request.state, 'user_id'):
return forbidden_response("请先登录")
is_study = await PermissionChecker.check_is_study_commissioner(request.state.user_id)
if not is_study:
return forbidden_response("需要学习委员权限")
return await func(*args, **kwargs)
return wrapper

123
backend/routes/admin.py
View File

@@ -16,7 +16,6 @@ import json
from middleware.permission import (
get_current_user,
require_teacher,
require_monitor,
PermissionChecker
)
from services.admin_service import AdminService
@@ -27,7 +26,7 @@ from schemas.admin import (
AddPointsRequest, RevokeRequest, AddAdminRequest,
UpdateHomeworkStatusRequest, AddAttendanceRequest
)
from utils.response import success_response, error_response, paginated_response
from utils.response import success_response, error_response
from utils.logger import get_logger
from config import settings
@@ -44,50 +43,30 @@ async def get_students(
page_size: int = Query(20, ge=1, le=100),
search: Optional[str] = None
):
"""
获取所有学生列表（单班级）
"""
"""获取所有学生列表（单班级）"""
user = await get_current_user(request)
result = await AdminService.get_students(
page=page,
page_size=page_size,
search=search
)
result = await AdminService.get_students(page=page, page_size=page_size, search=search)
return success_response(data=result)
@router.post("/students/import")
async def import_students(
request: Request,
file: UploadFile = File(...)
):
"""
批量导入学生（JSON格式）
初始操行分默认为60分
"""
async def import_students(request: Request, file: UploadFile = File(...)):
"""批量导入学生（JSON格式），初始操行分默认为60分"""
user = await get_current_user(request)
# 检查权限（仅班主任）
is_teacher = await PermissionChecker.check_is_teacher(user["user_id"])
if not is_teacher:
return error_response(message="仅班主任可导入学生", code=403)
# 检查文件大小
content = await file.read()
file_size = len(content)
if file_size > settings.MAX_UPLOAD_SIZE:
return error_response(message=f"文件大小不能超过{settings.MAX_UPLOAD_SIZE // 1024 // 1024}MB")
# 检查文件扩展名
filename = file.filename or ""
extension = filename.split('.')[-1].lower() if '.' in filename else ''
if extension not in settings.ALLOWED_EXTENSIONS:
return error_response(message=f"不支持的文件类型，仅支持 {', '.join(settings.ALLOWED_EXTENSIONS)}")
# 解析JSON
try:
data = json.loads(content.decode('utf-8'))
students = data.get("students", [])
@@ -99,13 +78,11 @@ async def import_students(
if not students:
return error_response(message="文件中没有学生数据")
# 导入学生（初始操行分60分）
result = await AdminService.import_students(
students=students,
operator_id=user["user_id"],
initial_points=60
)
return success_response(data=result, message=f"导入完成: 成功{result['success_count']}人，失败{result['failed_count']}人")
@@ -116,11 +93,8 @@ async def add_student(
name: str,
parent_phone: Optional[str] = None
):
"""
新增学生
"""
"""新增学生"""
user = await get_current_user(request)
is_teacher = await PermissionChecker.check_is_teacher(user["user_id"])
if not is_teacher:
return error_response(message="仅班主任可新增学生", code=403)
@@ -132,7 +106,6 @@ async def add_student(
operator_id=user["user_id"],
initial_points=60
)
if result["success"]:
return success_response(data=result, message="学生添加成功")
else:
@@ -143,11 +116,8 @@ async def add_student(
@router.post("/conduct/add")
async def add_conduct_points(request: Request, req: AddPointsRequest):
"""
批量加减分
"""
"""批量加减分"""
user = await get_current_user(request)
result = await ConductService.add_points(
student_ids=req.student_ids,
points_change=req.points_change,
@@ -155,7 +125,6 @@ async def add_conduct_points(request: Request, req: AddPointsRequest):
recorder_id=user["user_id"],
recorder_name=user["username"]
)
if result["success"]:
return success_response(data=result, message="操作成功")
else:
@@ -164,16 +133,12 @@ async def add_conduct_points(request: Request, req: AddPointsRequest):
@router.post("/conduct/revoke")
async def revoke_conduct_record(request: Request, req: RevokeRequest):
"""
撤销扣分记录
"""
"""撤销扣分记录"""
user = await get_current_user(request)
result = await ConductService.revoke_record(
record_id=req.record_id,
revoker_id=user["user_id"]
)
if result["success"]:
return success_response(message="撤销成功")
else:
@@ -189,11 +154,8 @@ async def get_conduct_history(
start_date: Optional[str] = None,
end_date: Optional[str] = None
):
"""
获取操行分历史记录
"""
"""获取操行分历史记录"""
user = await get_current_user(request)
result = await ConductService.get_history(
user_id=user["user_id"],
student_id=student_id,
@@ -202,7 +164,6 @@ async def get_conduct_history(
start_date=start_date,
end_date=end_date
)
return success_response(data=result)
@@ -210,31 +171,26 @@ async def get_conduct_history(
@router.get("/homework/assignments")
async def get_assignments(request: Request):
"""
获取作业列表
"""
"""获取作业列表"""
user = await get_current_user(request)
role = await PermissionChecker.get_user_role(user["user_id"])
if role not in ["班主任", "学习委员"]:
return error_response(message="无权限", code=403)
result = await HomeworkService.get_assignments(user["user_id"])
return success_response(data=result)
@router.get("/homework/submissions/{assignment_id}")
async def get_submissions(request: Request, assignment_id: int):
"""
获取作业提交记录
"""
"""获取作业提交记录"""
user = await get_current_user(request)
role = await PermissionChecker.get_user_role(user["user_id"])
if role not in ["班主任", "学习委员"]:
return error_response(message="无权限", code=403)
result = await HomeworkService.get_submissions(
assignment_id=assignment_id,
user_id=user["user_id"]
)
return success_response(data=result)
@@ -246,15 +202,11 @@ async def create_assignment(
description: Optional[str] = None,
deadline: str = None
):
"""
发布作业（班主任）
"""
"""发布作业（班主任）"""
user = await get_current_user(request)
is_teacher = await PermissionChecker.check_is_teacher(user["user_id"])
if not is_teacher:
return error_response(message="仅班主任可发布作业", code=403)
result = await HomeworkService.create_assignment(
subject_id=subject_id,
title=title,
@@ -262,7 +214,6 @@ async def create_assignment(
deadline=deadline,
created_by=user["user_id"]
)
if result["success"]:
return success_response(data=result, message="作业发布成功")
else:
@@ -271,11 +222,11 @@ async def create_assignment(
@router.put("/homework/submission")
async def update_submission_status(request: Request, req: UpdateHomeworkStatusRequest):
"""更新作业提交状态（班主任或学习委员）"""
user = await get_current_user(request)
role = await PermissionChecker.get_user_role(user["user_id"])
if role not in ["班主任", "学习委员"]:
return error_response(message="无权进行此操作", code=403)
result = await HomeworkService.update_submission_status(
submission_id=req.submission_id,
status=req.status,
@@ -283,7 +234,6 @@ async def update_submission_status(request: Request, req: UpdateHomeworkStatusRe
apply_deduction=req.apply_deduction,
operator_id=user["user_id"]
)
if result["success"]:
return success_response(message="状态更新成功")
else:
@@ -294,11 +244,11 @@ async def update_submission_status(request: Request, req: UpdateHomeworkStatusRe
@router.post("/attendance")
async def add_attendance(request: Request, req: AddAttendanceRequest):
"""
添加考勤记录（考勤委员）
"""
"""添加考勤记录（考勤委员）"""
user = await get_current_user(request)
role = await PermissionChecker.get_user_role(user["user_id"])
if role not in ["班主任", "考勤委员"]:
return error_response(message="无权进行此操作", code=403)
result = await AttendanceService.add_attendance(
student_id=req.student_id,
date=str(req.date),
@@ -307,7 +257,6 @@ async def add_attendance(request: Request, req: AddAttendanceRequest):
apply_deduction=req.apply_deduction,
recorder_id=user["user_id"]
)
if result["success"]:
return success_response(message="考勤记录添加成功")
else:
@@ -320,17 +269,13 @@ async def get_attendance_records(
date: Optional[str] = None,
student_id: Optional[int] = None
):
"""
获取考勤记录
"""
"""获取考勤记录"""
user = await get_current_user(request)
result = await AttendanceService.get_records(
user_id=user["user_id"],
date=date,
student_id=student_id
)
return success_response(data=result)
@@ -338,15 +283,11 @@ async def get_attendance_records(
@router.post("/admin/add")
async def add_admin(request: Request, req: AddAdminRequest):
"""
添加管理员（班主任）
"""
"""添加管理员（班主任）"""
user = await get_current_user(request)
if not await PermissionChecker.check_is_teacher(user["user_id"]):
is_teacher = await PermissionChecker.check_is_teacher(user["user_id"])
if not is_teacher:
return error_response(message="仅班主任可添加管理员", code=403)
# 验证角色类型是否合法
if req.role_type not in ["班长", "学习委员", "考勤委员", "劳动委员"]:
return error_response(message="无效的角色类型", code=400)
result = await AdminService.add_admin(
@@ -356,15 +297,6 @@ async def add_admin(request: Request, req: AddAdminRequest):
role_type=req.role_type,
operator_id=user["user_id"]
)
result = await AdminService.add_admin(
username=req.username,
real_name=req.real_name,
password=req.password,
role_type=req.role_type,
operator_id=user["user_id"]
)
if result["success"]:
return success_response(data=result, message="管理员添加成功")
else:
@@ -373,15 +305,10 @@ async def add_admin(request: Request, req: AddAdminRequest):
@router.get("/admin/list")
async def get_admins(request: Request):
"""
获取管理员列表（班主任）
"""
"""获取管理员列表（班主任）"""
user = await get_current_user(request)
is_teacher = await PermissionChecker.check_is_teacher(user["user_id"])
if not is_teacher:
return error_response(message="仅班主任可查看管理员列表", code=403)
result = await AdminService.get_admins()
return success_response(data=result)