# =========================================== # 班级操行分管理系统 - 后端服务 # # 开发者: Canglan # 联系方式: admin@sea-studio.top # 版权归属: Sea Network Technology Studio # 许可证: MIT License # # 版权所有 © Sea Network Technology Studio # =========================================== from typing import Dict, Any, Optional from datetime import datetime from models.attendance import AttendanceModel from models.student import StudentModel from models.conduct import ConductModel from middleware.permission import PermissionChecker from config import settings from utils.logger import get_logger logger = get_logger(__name__) class AttendanceService: """考勤服务""" @staticmethod async def add_attendance( student_id: int, date: str, status: str, reason: Optional[str], apply_deduction: bool, recorder_id: int ) -> Dict[str, Any]: """添加考勤记录""" # 检查权限 role = await PermissionChecker.get_user_role(recorder_id) if role not in ["班主任", "考勤委员"]: return {"success": False, "message": "无权进行此操作"} # 检查是否同班级 can_manage = await PermissionChecker.check_can_manage_student(recorder_id, student_id) if not can_manage: return {"success": False, "message": "无权操作该学生"} # 添加考勤记录 attendance_id = await AttendanceModel.create_record( student_id=student_id, date=date, status=status, reason=reason, recorder_id=recorder_id ) if not attendance_id: return {"success": False, "message": "添加考勤记录失败"} # 应用扣分 if apply_deduction and status in ["absent", "late", "leave"]: # 确定扣分数值 if status == "absent": points_change = -settings.DEDUCTION_ATTENDANCE_ABSENT elif status == "late": points_change = -settings.DEDUCTION_ATTENDANCE_LATE else: points_change = -settings.DEDUCTION_ATTENDANCE_LEAVE # 创建扣分记录 student = await StudentModel.get_by_id(student_id) if student: await ConductModel.create_record( student_id=student_id, points_change=points_change, reason=f"考勤异常: {status}", recorder_id=recorder_id, related_type="attendance", related_id=attendance_id ) # 标记已应用扣分 await AttendanceModel.mark_deduction_applied(attendance_id) logger.info(f"用户[{recorder_id}] 添加考勤记录[{attendance_id}] -> {status}") return {"success": True, "message": "考勤记录添加成功"} @staticmethod async def get_records( user_id: int, date: Optional[str] = None, student_id: Optional[int] = None ) -> Dict[str, Any]: """获取考勤记录""" role = await PermissionChecker.get_user_role(user_id) if role in ["班主任", "考勤委员"]: class_id = await PermissionChecker.get_user_class_id(user_id) records = await AttendanceModel.get_class_records( class_id=class_id, date=date, student_id=student_id ) elif student_id: # 查看指定学生 can_manage = await PermissionChecker.check_can_manage_student(user_id, student_id) if not can_manage: return {"error": "无权查看该学生记录"} records = await AttendanceModel.get_student_records(student_id) else: records = [] return {"records": records}