98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
# ===========================================
|
|
# 班级操行分管理系统 - 后端服务
|
|
#
|
|
# 开发者: Canglan
|
|
# 联系方式: admin@sea-studio.top
|
|
# 版权归属: Sea Network Technology Studio
|
|
# 许可证: MIT License
|
|
#
|
|
# 版权所有 © Sea Network Technology Studio
|
|
# ===========================================
|
|
|
|
from typing import Optional, Dict, Any, List
|
|
from datetime import datetime
|
|
from utils.database import execute_one, execute_query, execute_insert, execute_update
|
|
|
|
|
|
class AttendanceModel:
|
|
"""考勤数据模型"""
|
|
|
|
@staticmethod
|
|
async def get_student_records(student_id: int, month: str = None) -> List[Dict[str, Any]]:
|
|
sql = """
|
|
SELECT attendance_id, date, status, reason, deduction_applied, created_at
|
|
FROM attendance_records
|
|
WHERE student_id = %s
|
|
"""
|
|
params = [student_id]
|
|
|
|
if month:
|
|
sql += " AND DATE_FORMAT(date, '%%Y-%%m') = %s"
|
|
params.append(month)
|
|
|
|
sql += " ORDER BY date DESC"
|
|
|
|
return await execute_query(sql, tuple(params))
|
|
|
|
@staticmethod
|
|
async def get_class_records(
|
|
class_id: int,
|
|
date: str = None,
|
|
student_id: int = None
|
|
) -> List[Dict[str, Any]]:
|
|
sql = """
|
|
SELECT ar.*, s.name as student_name, s.student_no
|
|
FROM attendance_records ar
|
|
JOIN students s ON ar.student_id = s.student_id
|
|
WHERE s.class_id = %s
|
|
"""
|
|
params = [class_id]
|
|
|
|
if date:
|
|
sql += " AND ar.date = %s"
|
|
params.append(date)
|
|
|
|
if student_id:
|
|
sql += " AND ar.student_id = %s"
|
|
params.append(student_id)
|
|
|
|
sql += " ORDER BY ar.date DESC, s.student_no"
|
|
|
|
return await execute_query(sql, tuple(params))
|
|
|
|
@staticmethod
|
|
async def create_record(
|
|
student_id: int,
|
|
date: str,
|
|
status: str,
|
|
reason: str = None,
|
|
recorder_id: int = None
|
|
) -> int:
|
|
# 检查是否已存在当天记录
|
|
existing = await execute_one(
|
|
"SELECT attendance_id FROM attendance_records WHERE student_id = %s AND date = %s",
|
|
(student_id, date)
|
|
)
|
|
|
|
if existing:
|
|
# 更新已有记录
|
|
sql = """
|
|
UPDATE attendance_records
|
|
SET status = %s, reason = %s, recorder_id = %s
|
|
WHERE student_id = %s AND date = %s
|
|
"""
|
|
await execute_update(sql, (status, reason, recorder_id, student_id, date))
|
|
return existing["attendance_id"]
|
|
else:
|
|
# 插入新记录
|
|
sql = """
|
|
INSERT INTO attendance_records (student_id, date, status, reason, recorder_id)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
"""
|
|
return await execute_insert(sql, (student_id, date, status, reason, recorder_id))
|
|
|
|
@staticmethod
|
|
async def mark_deduction_applied(attendance_id: int) -> bool:
|
|
sql = "UPDATE attendance_records SET deduction_applied = 1 WHERE attendance_id = %s"
|
|
result = await execute_update(sql, (attendance_id,))
|
|
return result > 0 |