# =========================================== # 班级操行分管理系统 - 后端服务 # # 开发者: Canglan # 联系方式: admin@sea-studio.top # 版权归属: Sea Network Technology Studio # 许可证: MIT License # # 版权所有 © Sea Network Technology Studio # =========================================== from typing import Dict, Any, List, Optional from datetime import datetime from models.student import StudentModel from models.conduct import ConductModel from models.user import UserModel from middleware.permission import PermissionChecker from config import settings from utils.logger import get_logger logger = get_logger(__name__) class ConductService: """操行分服务""" @staticmethod async def add_points( student_ids: List[int], points_change: int, reason: str, recorder_id: int, recorder_name: str ) -> Dict[str, Any]: """ 批量加减分 """ # 验证分值 if points_change == 0: return {"success": False, "message": "分值不能为0"} # 获取操作人角色 role = await PermissionChecker.get_user_role(recorder_id) # 权限验证 if role == "班主任": # 班主任无限制 pass elif role == "班长": # 班长限制 ±5分 if points_change > settings.MONITOR_MAX_ADD or points_change < settings.MONITOR_MAX_SUBTRACT: return {"success": False, "message": f"班长单次只能加减{settings.MONITOR_MAX_ADD}分以内"} elif role == "劳动委员": # 劳动委员可加减分，±LABOR_REP_MAX_POINTS以内 if abs(points_change) > settings.LABOR_REP_MAX_POINTS: return {"success": False, "message": f"劳动委员单次只能加减{settings.LABOR_REP_MAX_POINTS}分以内"} elif role == "志愿委员": # 志愿委员只能加分，上限VOLUNTEER_REP_MAX_POINTS if points_change < 0: return {"success": False, "message": "志愿委员只能加分"} if points_change > settings.VOLUNTEER_REP_MAX_POINTS: return {"success": False, "message": f"志愿委员单次最多加{settings.VOLUNTEER_REP_MAX_POINTS}分"} elif role == "学习委员": # 学习委员可加减分，±STUDY_COMMISSIONER_MAX_POINTS以内 if abs(points_change) > settings.STUDY_COMMISSIONER_MAX_POINTS: return {"success": False, "message": f"学习委员单次只能加减{settings.STUDY_COMMISSIONER_MAX_POINTS}分以内"} elif role == "考勤委员": # 考勤委员只能扣分，上限ATTENDANCE_REP_MAX_POINTS if points_change > 0: return {"success": False, "message": "考勤委员只能进行扣分操作"} if abs(points_change) > settings.ATTENDANCE_REP_MAX_POINTS: return {"success": False, "message": f"考勤委员单次最多扣{settings.ATTENDANCE_REP_MAX_POINTS}分"} else: return {"success": False, "message": "无权进行此操作"} # 批量处理 success_count = 0 fail_count = 0 details = [] for student_id in student_ids: try: # 检查学生是否存在 student = await StudentModel.get_by_id(student_id) if not student: details.append({"student_id": student_id, "error": "学生不存在"}) fail_count += 1 continue # 创建记录 record_id = await ConductModel.create_record( student_id=student_id, points_change=points_change, reason=reason, recorder_id=recorder_id, recorder_name=recorder_name ) # 更新学生总分 await StudentModel.update_total_points(student_id, points_change) details.append({"student_id": student_id, "success": True, "record_id": record_id}) success_count += 1 logger.info(f"用户[{recorder_id}] 对学生[{student_id}] 进行 {points_change} 分操作") except Exception as e: details.append({"student_id": student_id, "error": str(e)}) fail_count += 1 return { "success": fail_count == 0, "success_count": success_count, "fail_count": fail_count, "details": details } @staticmethod async def revoke_record(record_id: int, revoker_id: int) -> Dict[str, Any]: """撤销扣分记录""" # 检查权限 can_revoke = await PermissionChecker.check_can_revoke(revoker_id, record_id) if not can_revoke: return {"success": False, "message": "无权撤销此记录"} # 先获取原记录信息（用于恢复分数） record = await ConductModel.get_record_by_id(record_id) if not record: return {"success": False, "message": "记录不存在"} # 归档后班主任仍可撤销/修改记录（任务需求#8） # 归档操作本身不可逆，但归档数据可由班主任修改 # 撤销记录 result = await ConductModel.revoke_record(record_id, revoker_id) if result: # 反向恢复学生总分 await StudentModel.update_total_points(record["student_id"], -record["points_change"]) logger.info(f"用户[{revoker_id}] 撤销了记录[{record_id}]") return { "success": True, "message": "撤销成功", "record": { "student_id": record["student_id"], "recorder_name": record.get("recorder_name", "未知"), "points_change": record["points_change"], "reason": record.get("reason", "") } } else: return {"success": False, "message": "撤销失败"} @staticmethod async def get_history( user_id: int, student_id: Optional[int] = None, page: int = 1, page_size: int = 20, start_date: Optional[str] = None, end_date: Optional[str] = None, grouped: bool = False ) -> Dict[str, Any]: """获取历史记录""" # 空字符串转为None if start_date == "": start_date = None if end_date == "": end_date = None role = await PermissionChecker.get_user_role(user_id) offset = (page - 1) * page_size # 班主任/班长/志愿委员可查看全班 if role in ["班主任", "班长", "志愿委员"]: if grouped: return await ConductModel.get_grouped_records( student_id=student_id, start_date=start_date, end_date=end_date, page=page, page_size=page_size ) records = await ConductModel.get_all_records( limit=page_size, offset=offset, start_date=start_date, end_date=end_date, student_id=student_id ) # 获取总数 from utils.database import execute_one count_conditions = ["cr.is_revoked = 0"] count_params = [] if student_id: count_conditions.append("cr.student_id = %s") count_params.append(student_id) if start_date: count_conditions.append("DATE(cr.created_at) >= %s") count_params.append(start_date) if end_date: count_conditions.append("DATE(cr.created_at) <= %s") count_params.append(end_date) count_where = " AND ".join(count_conditions) count_sql = f""" SELECT COUNT(*) as total FROM conduct_records cr JOIN students s ON cr.student_id = s.student_id WHERE {count_where} """ total_result = await execute_one(count_sql, tuple(count_params)) total = total_result["total"] if total_result else 0 elif student_id: # 管理员可查看指定学生 records = await ConductModel.get_student_records( student_id=student_id, limit=page_size, offset=offset ) total = len(await ConductModel.get_student_records(student_id, limit=10000)) else: # 查看自己提交的记录 records = await ConductModel.get_records_by_recorder( recorder_id=user_id, limit=page_size, offset=offset ) total = len(await ConductModel.get_records_by_recorder(user_id, limit=10000)) return { "records": records, "page": page, "page_size": page_size, "total": total, "total_pages": (total + page_size - 1) // page_size }