# =========================================== # 班级操行分管理系统 - 后端服务 # # 开发者: 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 models.user import UserModel from middleware.permission import PermissionChecker from config import settings from utils.logger import get_logger logger = get_logger(__name__) # 考勤状态中文映射 ATTENDANCE_STATUS_MAP = { "absent": "缺勤", "late": "迟到", "leave": "请假" } class AttendanceService: """考勤服务""" @staticmethod async def add_attendance( student_id: int, date: str, status: str, reason: Optional[str], apply_deduction: bool, recorder_id: int, custom_deduction: Optional[int] = None ) -> Dict[str, Any]: """添加考勤记录""" # 检查权限 role = await PermissionChecker.get_user_role(recorder_id) if role not in ["班主任", "考勤委员"]: 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 custom_deduction is not None: points_change = -custom_deduction elif 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: # 获取操作人姓名 user = await UserModel.get_by_user_id(recorder_id) recorder_name = user.get("real_name", "班主任") if user else "班主任" # 使用中文状态 # 使用中文状态 status_text = ATTENDANCE_STATUS_MAP.get(status, status) await ConductModel.create_record( student_id=student_id, points_change=points_change, reason=f"考勤:{status_text}", recorder_id=recorder_id, recorder_name=recorder_name, related_type="attendance", related_id=attendance_id ) # 更新学生总分 await StudentModel.update_total_points(student_id, points_change) # 标记已应用扣分 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 ["班主任", "考勤委员"]: records = await AttendanceModel.get_class_records( date=date, student_id=student_id ) elif student_id: # 管理员可查看指定学生 records = await AttendanceModel.get_student_records(student_id) else: records = [] return {"records": records}