219 lines
8.7 KiB
Python
219 lines
8.7 KiB
Python
# ===========================================
|
||
# 班级操行分管理系统 - 学期数据模型
|
||
#
|
||
# 开发者: Canglan
|
||
# 联系方式: admin@sea-studio.top
|
||
# 版权归属: Sea Network Technology Studio
|
||
# 许可证: MIT License
|
||
#
|
||
# 版权所有 © Sea Network Technology Studio
|
||
# ===========================================
|
||
|
||
from typing import Optional, List, Dict, Any
|
||
from utils.database import execute_one, execute_query, execute_insert, execute_update, execute_many
|
||
from utils.logger import get_logger
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
class SemesterModel:
|
||
"""学期数据模型"""
|
||
|
||
@staticmethod
|
||
async def create(
|
||
semester_name: str,
|
||
start_date: str = None,
|
||
end_date: str = None
|
||
) -> int:
|
||
"""创建学期"""
|
||
sql = """
|
||
INSERT INTO semesters (semester_name, start_date, end_date)
|
||
VALUES (%s, %s, %s)
|
||
"""
|
||
return await execute_insert(sql, (semester_name, start_date, end_date))
|
||
|
||
@staticmethod
|
||
async def get_by_id(semester_id: int) -> Optional[Dict[str, Any]]:
|
||
"""根据ID获取学期信息"""
|
||
sql = "SELECT * FROM semesters WHERE semester_id = %s"
|
||
return await execute_one(sql, (semester_id,))
|
||
|
||
@staticmethod
|
||
async def get_all() -> List[Dict[str, Any]]:
|
||
"""获取所有学期列表"""
|
||
sql = """
|
||
SELECT semester_id, semester_name, start_date, end_date,
|
||
is_active, is_archived, created_at
|
||
FROM semesters
|
||
ORDER BY created_at DESC
|
||
"""
|
||
return await execute_query(sql)
|
||
|
||
@staticmethod
|
||
async def get_active() -> Optional[Dict[str, Any]]:
|
||
"""获取当前活跃学期（优先 is_active 标记，降级为日期范围匹配）"""
|
||
fields = "semester_id, semester_name, start_date, end_date, is_active, is_archived, created_at"
|
||
# 第一优先级：is_active 标记
|
||
sql = f"""
|
||
SELECT {fields}
|
||
FROM semesters
|
||
WHERE is_active = 1 AND is_archived = 0
|
||
LIMIT 1
|
||
"""
|
||
result = await execute_one(sql)
|
||
if result:
|
||
return result
|
||
# 第二优先级：日期范围匹配
|
||
# 注：无日期的学期不会自动匹配为活跃学期（需手动激活）
|
||
sql = f"""
|
||
SELECT {fields}
|
||
FROM semesters
|
||
WHERE is_archived = 0 AND start_date <= CURDATE() AND (end_date IS NULL OR end_date >= CURDATE())
|
||
LIMIT 1
|
||
"""
|
||
return await execute_one(sql)
|
||
|
||
@staticmethod
|
||
async def deactivate_all() -> int:
|
||
"""将所有学期设为非活跃"""
|
||
sql = "UPDATE semesters SET is_active = 0 WHERE is_active = 1"
|
||
return await execute_update(sql)
|
||
|
||
@staticmethod
|
||
async def activate(semester_id: int) -> bool:
|
||
"""设为当前活跃学期"""
|
||
sql = """
|
||
UPDATE semesters SET is_active = 1
|
||
WHERE semester_id = %s AND is_archived = 0
|
||
"""
|
||
result = await execute_update(sql, (semester_id,))
|
||
return result > 0
|
||
|
||
@staticmethod
|
||
async def archive(semester_id: int) -> bool:
|
||
"""归档学期"""
|
||
sql = """
|
||
UPDATE semesters SET is_archived = 1, is_active = 0
|
||
WHERE semester_id = %s AND is_archived = 0
|
||
"""
|
||
result = await execute_update(sql, (semester_id,))
|
||
return result > 0
|
||
|
||
@staticmethod
|
||
async def is_archived(semester_id: int) -> bool:
|
||
"""检查学期是否已归档"""
|
||
sql = "SELECT is_archived FROM semesters WHERE semester_id = %s"
|
||
result = await execute_one(sql, (semester_id,))
|
||
if not result:
|
||
return False
|
||
return bool(result['is_archived'])
|
||
|
||
@staticmethod
|
||
async def get_record_semester_id(record_id: int) -> Optional[int]:
|
||
"""获取操行分记录所属的学期ID"""
|
||
sql = "SELECT semester_id FROM conduct_records WHERE record_id = %s"
|
||
result = await execute_one(sql, (record_id,))
|
||
return result['semester_id'] if result else None
|
||
|
||
@staticmethod
|
||
async def get_attendance_stats_by_semester(semester_id: int, start_date: str, end_date: str) -> List[Dict]:
|
||
"""批量查询学期内所有学生的考勤统计"""
|
||
sql = """
|
||
SELECT student_id, status, COUNT(*) as cnt
|
||
FROM attendance_records
|
||
WHERE (semester_id = %s OR (semester_id IS NULL AND `date` BETWEEN %s AND %s))
|
||
GROUP BY student_id, status
|
||
"""
|
||
return await execute_query(sql, (semester_id, start_date, end_date))
|
||
|
||
@staticmethod
|
||
async def get_homework_stats_by_date_range(start_date: str, end_date: str) -> List[Dict]:
|
||
"""通过作业截止日期范围查询所有学生的作业提交统计"""
|
||
sql = """
|
||
SELECT hs.student_id, hs.status, COUNT(*) as cnt
|
||
FROM homework_submissions hs
|
||
JOIN assignments a ON hs.assignment_id = a.assignment_id
|
||
WHERE a.deadline BETWEEN %s AND %s
|
||
GROUP BY hs.student_id, hs.status
|
||
"""
|
||
return await execute_query(sql, (start_date, end_date))
|
||
|
||
|
||
class SemesterArchiveModel:
|
||
"""学期归档快照数据模型"""
|
||
|
||
@staticmethod
|
||
async def batch_create(archives_data: List[Dict]) -> int:
|
||
"""批量创建归档快照"""
|
||
if not archives_data:
|
||
return 0
|
||
sql = """
|
||
INSERT INTO semester_archives
|
||
(semester_id, student_id, student_no, student_name, final_points, rank_position, total_students,
|
||
attendance_present, attendance_absent, attendance_late, attendance_leave,
|
||
homework_submitted, homework_not_submitted, homework_late)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||
"""
|
||
params_list = [
|
||
(
|
||
a['semester_id'], a['student_id'], a['student_no'],
|
||
a['student_name'], a['final_points'],
|
||
a.get('rank_position', 0), a.get('total_students', 0),
|
||
a.get('attendance_present', 0), a.get('attendance_absent', 0),
|
||
a.get('attendance_late', 0), a.get('attendance_leave', 0),
|
||
a.get('homework_submitted', 0), a.get('homework_not_submitted', 0),
|
||
a.get('homework_late', 0)
|
||
)
|
||
for a in archives_data
|
||
]
|
||
return await execute_many(sql, params_list)
|
||
|
||
@staticmethod
|
||
async def delete_by_semester(semester_id: int) -> int:
|
||
"""删除指定学期的所有归档数据（用于归档操作的幂等性）"""
|
||
sql = "DELETE FROM semester_archives WHERE semester_id = %s"
|
||
return await execute_update(sql, (semester_id,))
|
||
|
||
@staticmethod
|
||
async def get_by_semester(semester_id: int) -> List[Dict[str, Any]]:
|
||
"""获取学期的归档数据"""
|
||
sql = """
|
||
SELECT archive_id, semester_id, student_id, student_no,
|
||
student_name, final_points, rank_position, total_students,
|
||
attendance_present, attendance_absent, attendance_late, attendance_leave,
|
||
homework_submitted, homework_not_submitted, homework_late, archived_at
|
||
FROM semester_archives
|
||
WHERE semester_id = %s
|
||
ORDER BY rank_position ASC
|
||
"""
|
||
return await execute_query(sql, (semester_id,))
|
||
|
||
@staticmethod
|
||
async def get_by_semester_and_student(semester_id: int, student_id: int) -> Optional[Dict[str, Any]]:
|
||
"""获取指定学期指定学生的归档数据"""
|
||
sql = """
|
||
SELECT archive_id, semester_id, student_id, student_no,
|
||
student_name, final_points, rank_position, total_students, archived_at
|
||
FROM semester_archives
|
||
WHERE semester_id = %s AND student_id = %s
|
||
"""
|
||
return await execute_one(sql, (semester_id, student_id))
|
||
|
||
@staticmethod
|
||
async def get_by_student(student_id: int) -> List[Dict[str, Any]]:
|
||
"""获取学生在所有已归档学期的数据"""
|
||
sql = """
|
||
SELECT sa.archive_id, sa.semester_id, sa.student_id, sa.student_no,
|
||
sa.student_name, sa.final_points, sa.rank_position,
|
||
sa.total_students, sa.attendance_present, sa.attendance_absent,
|
||
sa.attendance_late, sa.attendance_leave,
|
||
sa.homework_submitted, sa.homework_not_submitted, sa.homework_late,
|
||
sa.archived_at,
|
||
s.semester_name, s.start_date, s.end_date
|
||
FROM semester_archives sa
|
||
JOIN semesters s ON sa.semester_id = s.semester_id
|
||
WHERE sa.student_id = %s
|
||
ORDER BY sa.archived_at DESC
|
||
"""
|
||
return await execute_query(sql, (student_id,))
|