137 lines
6.1 KiB
Python
137 lines
6.1 KiB
Python
"""教师端班级和分组管理。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
|
|
|
|
class TeacherGroupAdminMixin:
|
|
|
|
def switch_class(self, class_name: str) -> tuple[bool, str]:
|
|
with self.lock:
|
|
row = self.connection.execute(
|
|
"SELECT COUNT(*) AS n FROM students WHERE class_name=?", (class_name,)
|
|
).fetchone()
|
|
if row["n"] == 0:
|
|
return False, "班级不存在"
|
|
self.set_setting("active_class", class_name)
|
|
return True, f"已切换到 {class_name} 班，仅该班学生可登录"
|
|
|
|
def set_locked(self, class_name: str, locked: bool) -> tuple[bool, str]:
|
|
with self.lock:
|
|
exists = self.connection.execute(
|
|
"SELECT 1 FROM students WHERE class_name=? LIMIT 1", (class_name,)
|
|
).fetchone()
|
|
if not exists:
|
|
return False, "班级不存在"
|
|
with self.lock, self.connection:
|
|
self.connection.execute(
|
|
"""INSERT INTO class_controls(class_name,locked) VALUES(?,?)
|
|
ON CONFLICT(class_name) DO UPDATE SET locked=excluded.locked""",
|
|
(class_name, int(locked)),
|
|
)
|
|
return True, ("已停止学生操作" if locked else "已恢复学生操作")
|
|
|
|
def _require_teacher_group(self, class_name: str) -> tuple[bool, str, sqlite3.Row | None]:
|
|
if not any(item["class_name"] == class_name for item in self.class_list()):
|
|
return False, "班级不存在", None
|
|
return True, "", None
|
|
|
|
def teacher_create_group(self, class_name: str, name: str) -> tuple[bool, str]:
|
|
if not self._require_teacher_group(class_name)[0]:
|
|
return False, "班级不存在"
|
|
name = name.strip() or "未命名小组"
|
|
with self.lock, self.connection:
|
|
self.connection.execute(
|
|
"INSERT INTO groups(class_name,name,leader_student_id) VALUES(?,?,NULL)",
|
|
(class_name, name),
|
|
)
|
|
return True, f"小组“{name}”已创建"
|
|
|
|
def teacher_rename_group(self, group_id: int, name: str) -> tuple[bool, str]:
|
|
name = name.strip()
|
|
if not name:
|
|
return False, "组名不能为空"
|
|
with self.lock, self.connection:
|
|
cursor = self.connection.execute("UPDATE groups SET name=? WHERE id=?", (name, group_id))
|
|
if cursor.rowcount == 0:
|
|
return False, "小组不存在"
|
|
return True, "组名已保存"
|
|
|
|
def teacher_set_leader(self, group_id: int, student_id: int | None) -> tuple[bool, str]:
|
|
with self.lock, self.connection:
|
|
group = self.connection.execute("SELECT * FROM groups WHERE id=?", (group_id,)).fetchone()
|
|
if group is None:
|
|
return False, "小组不存在"
|
|
if student_id is None:
|
|
self.connection.execute(
|
|
"UPDATE groups SET leader_student_id=NULL WHERE id=?", (group_id,)
|
|
)
|
|
return True, "已取消组长"
|
|
member = self.connection.execute(
|
|
"SELECT 1 FROM group_members WHERE group_id=? AND student_id=?",
|
|
(group_id, student_id),
|
|
).fetchone()
|
|
if not member:
|
|
return False, "组长必须是该组成员"
|
|
self.connection.execute(
|
|
"UPDATE groups SET leader_student_id=? WHERE id=?", (student_id, group_id)
|
|
)
|
|
return True, "组长已设置"
|
|
|
|
def teacher_add_member(self, group_id: int, student_id: int) -> tuple[bool, str]:
|
|
with self.lock, self.connection:
|
|
group = self.connection.execute("SELECT * FROM groups WHERE id=?", (group_id,)).fetchone()
|
|
student = self.connection.execute("SELECT * FROM students WHERE id=?", (student_id,)).fetchone()
|
|
if group is None or student is None:
|
|
return False, "小组或学生不存在"
|
|
if group["class_name"] != student["class_name"]:
|
|
return False, "不能跨班调整"
|
|
current_group = self.connection.execute(
|
|
"SELECT group_id FROM group_members WHERE student_id=?", (student_id,)
|
|
).fetchone()
|
|
if current_group and current_group["group_id"] == group_id:
|
|
return False, "学生已在该组"
|
|
count = self.connection.execute(
|
|
"SELECT COUNT(*) AS n FROM group_members WHERE group_id=?", (group_id,)
|
|
).fetchone()["n"]
|
|
group_limit = self.group_limit()
|
|
if count >= group_limit:
|
|
return False, f"该组已有 {group_limit} 人；请先移出一人"
|
|
if current_group:
|
|
self.connection.execute(
|
|
"DELETE FROM group_members WHERE student_id=?", (student_id,)
|
|
)
|
|
self.connection.execute(
|
|
"UPDATE groups SET leader_student_id=NULL WHERE id=? AND leader_student_id=?",
|
|
(current_group["group_id"], student_id),
|
|
)
|
|
self.connection.execute(
|
|
"INSERT INTO group_members(group_id,student_id) VALUES(?,?)",
|
|
(group_id, student_id),
|
|
)
|
|
return True, "已加入小组"
|
|
|
|
def teacher_remove_member(self, student_id: int) -> tuple[bool, str]:
|
|
with self.lock, self.connection:
|
|
membership = self.connection.execute(
|
|
"SELECT * FROM group_members WHERE student_id=?", (student_id,)
|
|
).fetchone()
|
|
if not membership:
|
|
return False, "学生当前未分组"
|
|
self.connection.execute(
|
|
"DELETE FROM group_members WHERE student_id=?", (student_id,)
|
|
)
|
|
self.connection.execute(
|
|
"""UPDATE groups SET leader_student_id=NULL
|
|
WHERE leader_student_id=?""", (student_id,)
|
|
)
|
|
return True, "已移出小组"
|
|
|
|
def teacher_disband_group(self, group_id: int) -> tuple[bool, str]:
|
|
with self.lock, self.connection:
|
|
cursor = self.connection.execute("DELETE FROM groups WHERE id=?", (group_id,))
|
|
if cursor.rowcount == 0:
|
|
return False, "小组不存在"
|
|
return True, "小组已解散"
|