114 lines
4.3 KiB
Python
114 lines
4.3 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 utils.database import execute_one, execute_query, execute_insert, execute_update
|
|
|
|
|
|
class HomeworkModel:
|
|
"""作业数据模型"""
|
|
|
|
@staticmethod
|
|
async def get_all_assignments() -> List[Dict[str, Any]]:
|
|
sql = """
|
|
SELECT a.*, s.subject_name, u.real_name as created_by_name
|
|
FROM assignments a
|
|
JOIN subjects s ON a.subject_id = s.subject_id
|
|
JOIN users u ON a.created_by = u.user_id
|
|
ORDER BY a.deadline ASC, a.created_at DESC
|
|
"""
|
|
return await execute_query(sql)
|
|
|
|
@staticmethod
|
|
async def get_assignments_by_subjects(subject_ids: List[int]) -> List[Dict[str, Any]]:
|
|
if not subject_ids:
|
|
return []
|
|
placeholders = ','.join(['%s'] * len(subject_ids))
|
|
sql = f"""
|
|
SELECT a.*, s.subject_name, u.real_name as created_by_name
|
|
FROM assignments a
|
|
JOIN subjects s ON a.subject_id = s.subject_id
|
|
JOIN users u ON a.created_by = u.user_id
|
|
WHERE a.subject_id IN ({placeholders})
|
|
ORDER BY a.deadline ASC, a.created_at DESC
|
|
"""
|
|
return await execute_query(sql, tuple(subject_ids))
|
|
|
|
@staticmethod
|
|
async def get_student_homework(student_id: int) -> List[Dict[str, Any]]:
|
|
sql = """
|
|
SELECT a.assignment_id, a.title, a.description, a.deadline,
|
|
s.subject_name, hs.status, hs.submit_time, hs.comments, hs.deduction_applied
|
|
FROM assignments a
|
|
JOIN subjects s ON a.subject_id = s.subject_id
|
|
LEFT JOIN homework_submissions hs ON a.assignment_id = hs.assignment_id AND hs.student_id = %s
|
|
ORDER BY a.deadline ASC, a.created_at DESC
|
|
"""
|
|
return await execute_query(sql, (student_id,))
|
|
|
|
@staticmethod
|
|
async def get_submission(submission_id: int) -> Optional[Dict[str, Any]]:
|
|
sql = """
|
|
SELECT hs.*, a.title, a.subject_id, a.assignment_id, s.name as student_name
|
|
FROM homework_submissions hs
|
|
JOIN assignments a ON hs.assignment_id = a.assignment_id
|
|
JOIN students s ON hs.student_id = s.student_id
|
|
WHERE hs.submission_id = %s
|
|
"""
|
|
return await execute_one(sql, (submission_id,))
|
|
|
|
@staticmethod
|
|
async def create_assignment(
|
|
subject_id: int,
|
|
title: str,
|
|
description: str,
|
|
deadline: str,
|
|
created_by: int
|
|
) -> int:
|
|
sql = """
|
|
INSERT INTO assignments (subject_id, title, description, deadline, created_by)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
"""
|
|
assignment_id = await execute_insert(sql, (subject_id, title, description, deadline, created_by))
|
|
|
|
# 为所有学生创建提交记录
|
|
from models.student import StudentModel
|
|
students = await StudentModel.get_all(include_disabled=False)
|
|
|
|
for student in students:
|
|
sql_sub = """
|
|
INSERT INTO homework_submissions (assignment_id, student_id, status)
|
|
VALUES (%s, %s, 'not_submitted')
|
|
"""
|
|
await execute_insert(sql_sub, (assignment_id, student["student_id"]))
|
|
|
|
return assignment_id
|
|
|
|
@staticmethod
|
|
async def update_submission(
|
|
submission_id: int,
|
|
status: str,
|
|
comments: str = None,
|
|
updated_by: int = None
|
|
) -> bool:
|
|
sql = """
|
|
UPDATE homework_submissions
|
|
SET status = %s, comments = %s, updated_by = %s, updated_at = NOW()
|
|
WHERE submission_id = %s
|
|
"""
|
|
result = await execute_update(sql, (status, comments, updated_by, submission_id))
|
|
return result > 0
|
|
|
|
@staticmethod
|
|
async def mark_deduction_applied(submission_id: int) -> bool:
|
|
sql = "UPDATE homework_submissions SET deduction_applied = 1 WHERE submission_id = %s"
|
|
result = await execute_update(sql, (submission_id,))
|
|
return result > 0 |