196 lines
9.6 KiB
Python
196 lines
9.6 KiB
Python
"""学生端分组用例和班级快照查询。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
|
|
class GroupOperationsMixin:
|
|
|
|
def class_snapshot(self, class_name: str) -> dict[str, Any]:
|
|
with self.lock:
|
|
groups = self.connection.execute(
|
|
"""SELECT g.id,g.name,g.leader_student_id,g.created_at,
|
|
s.student_no AS leader_student_no,s.name AS leader_name
|
|
FROM groups g LEFT JOIN students s ON s.id=g.leader_student_id
|
|
WHERE g.class_name=? ORDER BY g.id""",
|
|
(class_name,),
|
|
).fetchall()
|
|
group_ids = [row["id"] for row in groups]
|
|
members: dict[int, list[dict[str, Any]]] = {row["id"]: [] for row in groups}
|
|
if group_ids:
|
|
placeholders = ",".join("?" for _ in group_ids)
|
|
member_rows = self.connection.execute(
|
|
f"""SELECT m.group_id,st.id,st.student_no,st.name,st.gender,g.leader_student_id
|
|
FROM group_members m JOIN students st ON st.id=m.student_id
|
|
JOIN groups g ON g.id=m.group_id
|
|
WHERE m.group_id IN ({placeholders})
|
|
ORDER BY CASE WHEN st.id=g.leader_student_id THEN 0 ELSE 1 END,m.joined_at,st.student_no""",
|
|
group_ids,
|
|
).fetchall()
|
|
pending = [dict(row) for row in member_rows]
|
|
for member in pending:
|
|
member["is_leader"] = member["id"] == member["leader_student_id"]
|
|
member.pop("leader_student_id", None)
|
|
members[member["group_id"]].append(member)
|
|
|
|
ungrouped_rows = self.connection.execute(
|
|
"""SELECT id,student_no,name,gender FROM students
|
|
WHERE class_name=? AND id NOT IN (SELECT student_id FROM group_members)
|
|
ORDER BY student_no""",
|
|
(class_name,),
|
|
).fetchall()
|
|
total = self.connection.execute(
|
|
"SELECT COUNT(*) AS n FROM students WHERE class_name=?", (class_name,)
|
|
).fetchone()["n"]
|
|
grouped = self.connection.execute(
|
|
"""SELECT COUNT(*) AS n FROM group_members m
|
|
JOIN students s ON s.id=m.student_id WHERE s.class_name=?""",
|
|
(class_name,),
|
|
).fetchone()["n"]
|
|
|
|
group_data = []
|
|
for row in groups:
|
|
group_data.append(
|
|
{
|
|
**dict(row),
|
|
"locked": self.is_locked(class_name),
|
|
"member_count": len(members[row["id"]]),
|
|
"full": len(members[row["id"]]) >= self.group_limit(),
|
|
"members": members[row["id"]],
|
|
}
|
|
)
|
|
return {
|
|
"class_name": class_name,
|
|
"locked": self.is_locked(class_name),
|
|
"group_limit": self.group_limit(),
|
|
"groups": group_data,
|
|
"ungrouped": [dict(row) for row in ungrouped_rows],
|
|
"statistics": {"total_students": total, "grouped_students": grouped, "group_count": len(group_data)},
|
|
}
|
|
|
|
def create_group(self, student_id: int, group_name: str) -> tuple[bool, str, int | None]:
|
|
student = self.get_student(student_id)
|
|
if student is None:
|
|
return False, "学生不存在", None
|
|
if self.is_locked(student["class_name"]):
|
|
return False, "老师已停止分组操作，当前只能查看", None
|
|
current = self.student_membership(student_id)
|
|
if current is not None:
|
|
return False, "你已加入其他组，不能创建组；请先退出当前组", None
|
|
if student["must_change_password"]:
|
|
return False, "首次登录必须先修改密码", None
|
|
group_name = group_name.strip() or student["name"]
|
|
with self.lock, self.connection:
|
|
count = self.connection.execute(
|
|
"SELECT COUNT(*) AS n FROM groups WHERE class_name=?", (student["class_name"],)
|
|
).fetchone()["n"]
|
|
cursor = self.connection.execute(
|
|
"INSERT INTO groups(class_name,name,leader_student_id) VALUES(?,?,?)",
|
|
(student["class_name"], group_name, student_id),
|
|
)
|
|
group_id = int(cursor.lastrowid)
|
|
self.connection.execute(
|
|
"INSERT INTO group_members(group_id,student_id) VALUES(?,?)",
|
|
(group_id, student_id),
|
|
)
|
|
return True, f"“{group_name}”创建成功，你是组长", group_id
|
|
|
|
def join_group(self, student_id: int, group_id: int, confirmed: bool) -> tuple[bool, str]:
|
|
student = self.get_student(student_id)
|
|
if student is None:
|
|
return False, "学生不存在"
|
|
if student["must_change_password"]:
|
|
return False, "首次登录必须先修改密码"
|
|
if self.is_locked(student["class_name"]):
|
|
return False, "老师已停止分组操作，当前只能查看"
|
|
with self.lock:
|
|
group = self.connection.execute(
|
|
"SELECT * FROM groups WHERE id=?", (group_id,)
|
|
).fetchone()
|
|
current = self.connection.execute(
|
|
"""SELECT m.group_id,g.name,g.leader_student_id
|
|
FROM group_members m JOIN groups g ON g.id=m.group_id WHERE m.student_id=?""",
|
|
(student_id,),
|
|
).fetchone()
|
|
if group is None:
|
|
return False, "目标小组不存在"
|
|
if group["class_name"] != student["class_name"]:
|
|
return False, "不能加入其他班级的小组"
|
|
if current and current["group_id"] == group_id:
|
|
return False, "你已经在这个组里"
|
|
member_count = self.connection.execute(
|
|
"SELECT COUNT(*) AS n FROM group_members WHERE group_id=?", (group_id,)
|
|
).fetchone()["n"]
|
|
group_limit = self.group_limit()
|
|
if member_count >= group_limit:
|
|
return False, f"该组已有 {group_limit} 人，不能再加入新成员"
|
|
if current and not confirmed:
|
|
role = "组长，原组会被解散" if current["leader_student_id"] == student_id else "成员"
|
|
return False, f"你当前是“{current['name']}”的{role}。加入新组前必须退出原组；确认后才会执行。"
|
|
|
|
with self.lock, self.connection:
|
|
if current:
|
|
if current["leader_student_id"] == student_id:
|
|
self.connection.execute("DELETE FROM groups WHERE id=?", (current["group_id"],))
|
|
else:
|
|
self.connection.execute(
|
|
"DELETE FROM group_members WHERE student_id=?", (student_id,)
|
|
)
|
|
self.connection.execute(
|
|
"INSERT INTO group_members(group_id,student_id) VALUES(?,?)",
|
|
(group_id, student_id),
|
|
)
|
|
return True, "加入成功"
|
|
|
|
def leave_group(self, student_id: int, confirmed: bool) -> tuple[bool, str]:
|
|
if self.get_student(student_id) is None:
|
|
return False, "学生不存在"
|
|
current = self.student_membership(student_id)
|
|
if current is None:
|
|
return False, "你当前没有加入任何小组"
|
|
if self.is_locked(current["class_name"]):
|
|
return False, "老师已停止分组操作，当前只能查看"
|
|
is_leader = current["leader_student_id"] == student_id
|
|
if not confirmed:
|
|
if is_leader:
|
|
return False, "你是组长。退出时你创建的组将被解散，所有成员都会变为未分组；确认后才会执行。"
|
|
return False, "退出后可以再加入其他组；确认后才会执行。"
|
|
with self.lock, self.connection:
|
|
if is_leader:
|
|
self.connection.execute("DELETE FROM groups WHERE id=?", (current["group_id"],))
|
|
else:
|
|
self.connection.execute(
|
|
"DELETE FROM group_members WHERE group_id=? AND student_id=?",
|
|
(current["group_id"], student_id),
|
|
)
|
|
return True, "组长已解散小组" if is_leader else "已退出小组"
|
|
|
|
def leader_remove_member(self, leader_student_id: int, member_student_id: int) -> tuple[bool, str]:
|
|
with self.lock, self.connection:
|
|
leader_group = self.connection.execute(
|
|
"""SELECT m.group_id,g.name,g.class_name,g.leader_student_id
|
|
FROM group_members m JOIN groups g ON g.id=m.group_id
|
|
WHERE m.student_id=?""",
|
|
(leader_student_id,),
|
|
).fetchone()
|
|
if leader_group is None or leader_group["leader_student_id"] != leader_student_id:
|
|
return False, "只有组长可以移出组员"
|
|
if leader_student_id == member_student_id:
|
|
return False, "组长不能移出自己；如需退出请解散小组"
|
|
if self.is_locked(leader_group["class_name"]):
|
|
return False, "老师已停止分组操作，当前只能查看"
|
|
member = self.connection.execute(
|
|
"""SELECT st.name
|
|
FROM group_members m JOIN students st ON st.id=m.student_id
|
|
WHERE m.student_id=? AND m.group_id=?""",
|
|
(member_student_id, leader_group["group_id"]),
|
|
).fetchone()
|
|
if member is None:
|
|
return False, "该学生不在你的小组中"
|
|
self.connection.execute(
|
|
"DELETE FROM group_members WHERE student_id=? AND group_id=?",
|
|
(member_student_id, leader_group["group_id"]),
|
|
)
|
|
return True, f"已将 {member['name']} 移出小组"
|