145 lines
5.5 KiB
Python
145 lines
5.5 KiB
Python
# ===========================================
|
||
# 班级操行分管理系统 - 后端服务
|
||
#
|
||
# 开发者: 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,
|
||
slot: str = 'morning'
|
||
) -> Dict[str, Any]:
|
||
"""添加考勤记录"""
|
||
# 校验时段
|
||
if slot not in ('morning', 'afternoon', 'evening'):
|
||
return {"success": False, "message": "无效的考勤时段"}
|
||
# 校验状态
|
||
if status not in ('present', 'absent', 'late', 'leave'):
|
||
return {"success": False, "message": "无效的考勤状态"}
|
||
# 校验自定义扣分范围
|
||
if custom_deduction is not None and (custom_deduction < 1 or custom_deduction > 20):
|
||
return {"success": False, "message": "自定义扣分必须在1-20之间"}
|
||
|
||
# 检查权限
|
||
role = await PermissionChecker.get_user_role(recorder_id)
|
||
if role not in ["班主任", "考勤委员"]:
|
||
return {"success": False, "message": "无权进行此操作"}
|
||
|
||
# 考勤委员扣分上限
|
||
if role == "考勤委员" and apply_deduction and status in ["absent", "late"]:
|
||
if custom_deduction is not None and custom_deduction > settings.ATTENDANCE_REP_MAX_POINTS:
|
||
return {"success": False, "message": f"考勤委员单次扣分上限为{settings.ATTENDANCE_REP_MAX_POINTS}分"}
|
||
|
||
# 添加考勤记录
|
||
attendance_id = await AttendanceModel.create_record(
|
||
student_id=student_id,
|
||
date=date,
|
||
status=status,
|
||
reason=reason,
|
||
recorder_id=recorder_id,
|
||
slot=slot
|
||
)
|
||
|
||
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
|
||
|
||
# 扣分为0时跳过（如请假不扣分）
|
||
if points_change == 0:
|
||
logger.info(f"用户[{recorder_id}] 添加考勤记录[{attendance_id}] -> {status} (不扣分)")
|
||
return {"success": True, "message": "考勤记录添加成功（不扣分）"}
|
||
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,
|
||
slot: Optional[str] = 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,
|
||
slot=slot
|
||
)
|
||
elif student_id:
|
||
# 管理员可查看指定学生
|
||
records = await AttendanceModel.get_student_records(student_id)
|
||
else:
|
||
records = []
|
||
|
||
return {"records": records} |