commit c6e9a82fe5fedff90620947c600827b51d107659 Author: Shuming Liu Date: Tue Sep 8 10:06:45 2026 +0800 first commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a6c4911 --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +*.pyc +selfgroup.db +*.db +.upload-roster.xlsx +.DS_Store diff --git a/2025mt.xlsx b/2025mt.xlsx new file mode 100644 index 0000000..2f1a1ba Binary files /dev/null and b/2025mt.xlsx differ diff --git a/README.md b/README.md new file mode 100644 index 0000000..5167958 --- /dev/null +++ b/README.md @@ -0,0 +1,90 @@ +# 学生自助分组系统 + +这是一个基于 B/S 架构的本地/校园网分组系统。项目使用 Python 标准库实现，不需要安装第三方依赖；数据库为 SQLite，学生名单来源为 `2025mt.xlsx`。 + +## 功能 + +- 学生用学号登录，首次登录密码也是学号，并强制修改密码。 +- 学生登录后显示姓名、学号和登录状态。 +- 学生可创建组并自动成为组长；其他学生可选择加入。 +- 每组最多 6 人，包括组长。 +- 已加入其他组的学生不能创建新组；组长加入新组时，需确认解散原组；普通成员切换组时需确认退出原组。 +- 学生端和教师端都能直接看到全部组及成员的学号、姓名，无需点开详情。 +- 教师可切换“当前开放登录班级”，未开放班级不能登录。 +- 教师可停止/恢复学生分组操作。 +- 停止后学生只能查看；教师仍可手工建组、改名、设组长、加入、移出、解散。 +- 教师可导出当前班分组为 `.xlsx`。 + +## 启动 + +在本目录执行： + +```bash +python3 app.py +``` + +默认访问地址： + +- 学生端： +- 教师端： + +如果学生要从其他电脑/手机访问，请把 `127.0.0.1` 换成老师电脑的局域网 IP，例如： + +```text +http://192.168.1.20:8000/ +``` + +可自定义端口和数据库： + +```bash +PORT=8000 GROUP_DB=selfgroup.db python3 app.py +``` + +## 教师账号 + +默认账号： + +```text +账号：teacher +密码：teacher123 +``` + +正式使用前建议改用环境变量设置： + +```bash +TEACHER_USER=yourname TEACHER_PASSWORD=your-strong-password python3 app.py +``` + +注意：教师密码只在首次创建数据库时写入。如果数据库已存在，修改环境变量后还需要删除旧数据库或直接更新数据库中 `settings` 表的 `teacher_password`。 + +## 学生名单 + +服务第一次启动时，如果没有数据库，会自动读取当前目录的 `2025mt.xlsx`。表格至少需要以下表头： + +```text +班级 学号 姓名 +``` + +也兼容读取：性别、录取编号、备注。 + +如果换了名单，可以在教师后台上传新的 `.xlsx`。导入会重建学生账号并解散全部已有分组。 + +## 数据文件 + +- `selfgroup.db`：学生、密码、分组、班级状态和登录会话。 +- `2025mt.xlsx`：首次启动导入的原始名单。 + +## 建议部署方式 + +1. 在教师电脑上运行 `python3 app.py`。 +2. 查看教师电脑局域网 IP。 +3. 将学生端网址发到班级群。 +4. 教师打开 `/teacher`，切换到要分组的班级。 +5. 分组完成后点击“停止学生操作”。 +6. 检查并手工调整分组，然后导出 Excel。 + +## 注意 + +- 学生密码使用 PBKDF2 加盐哈希保存，不以明文保存。 +- 系统适合局域网使用；如果在公网部署，请使用 HTTPS 反向代理，并务必修改教师默认密码。 +- SQLite 适合班级规模的小型并发使用，不建议多名教师同时频繁写数据库。 diff --git a/app.py b/app.py new file mode 100644 index 0000000..d422e00 --- /dev/null +++ b/app.py @@ -0,0 +1,1136 @@ +#!/usr/bin/env python3 +"""学生自助分组系统 - B/S 架构，仅使用 Python 标准库。""" + +from __future__ import annotations + +import hashlib +import http.server +import hmac +import json +import logging +import os +import secrets +import sqlite3 +import threading +import time +import urllib.parse +from http import HTTPStatus +from pathlib import Path +from typing import Any + +from xlsxio import export_filename, read_xlsx_rows, write_xlsx + + +BASE_DIR = Path(__file__).resolve().parent +DB_PATH = Path(os.environ.get("GROUP_DB", BASE_DIR / "selfgroup.db")) +XLSX_PATH = Path(os.environ.get("GROUP_XLSX", BASE_DIR / "2025mt.xlsx")) +STATIC_DIR = BASE_DIR / "static" +TEACHER_USER = os.environ.get("TEACHER_USER", "teacher") +TEACHER_PASSWORD = os.environ.get("TEACHER_PASSWORD", "teacher123") +GROUP_LIMIT = 6 +SESSION_DAYS = 7 + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") +LOG = logging.getLogger("group-system") + + +def hash_password(password: str, salt: bytes | None = None) -> str: + salt = salt or secrets.token_bytes(16) + digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, 180_000) + return salt.hex() + "$" + digest.hex() + + +def verify_password(password: str, stored: str | None) -> bool: + if not stored or "$" not in stored: + return False + try: + salt_hex, digest_hex = stored.split("$", 1) + salt = bytes.fromhex(salt_hex) + expected = bytes.fromhex(digest_hex) + except ValueError: + return False + actual = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, 180_000) + return hmac.compare_digest(actual, expected) + + +class GroupStore: + def __init__(self, path: Path): + self.path = path + self.lock = threading.RLock() + self.connection = sqlite3.connect(str(path), check_same_thread=False) + self.connection.row_factory = sqlite3.Row + self.connection.execute("PRAGMA foreign_keys = ON") + self.init_database() + + def init_database(self) -> None: + with self.lock, self.connection: + self.connection.executescript( + """ + CREATE TABLE IF NOT EXISTS students ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + class_name TEXT NOT NULL, + student_no TEXT NOT NULL UNIQUE, + name TEXT NOT NULL, + gender TEXT DEFAULT '', + admission_no TEXT DEFAULT '', + remark TEXT DEFAULT '', + password_hash TEXT NOT NULL, + must_change_password INTEGER NOT NULL DEFAULT 1 + ); + CREATE TABLE IF NOT EXISTS groups ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + class_name TEXT NOT NULL, + name TEXT NOT NULL, + leader_student_id INTEGER REFERENCES students(id) ON DELETE SET NULL, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP + ); + CREATE TABLE IF NOT EXISTS group_members ( + group_id INTEGER NOT NULL REFERENCES groups(id) ON DELETE CASCADE, + student_id INTEGER NOT NULL UNIQUE REFERENCES students(id) ON DELETE CASCADE, + joined_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP + ); + CREATE TABLE IF NOT EXISTS class_controls ( + class_name TEXT PRIMARY KEY, + locked INTEGER NOT NULL DEFAULT 0 + ); + CREATE TABLE IF NOT EXISTS settings ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS sessions ( + token_hash TEXT PRIMARY KEY, + role TEXT NOT NULL, + username TEXT NOT NULL, + student_id INTEGER REFERENCES students(id) ON DELETE CASCADE, + created_at INTEGER NOT NULL, + expires_at INTEGER NOT NULL + ); + """ + ) + self.connection.execute( + "INSERT OR IGNORE INTO settings(key,value) VALUES('active_class','1')" + ) + self.connection.execute( + "INSERT OR IGNORE INTO settings(key,value) VALUES('teacher_username',?)", + (TEACHER_USER,), + ) + self.connection.execute( + "INSERT OR IGNORE INTO settings(key,value) VALUES('teacher_password',?)", + (hash_password(TEACHER_PASSWORD),), + ) + + def import_roster(self, path: Path) -> tuple[int, int]: + rows = read_xlsx_rows(path) + students = [] + errors = [] + seen = set() + for index, row in enumerate(rows, 2): + student_no = str(row.get("学号", "")).strip() + name = str(row.get("姓名", "")).strip() + class_name = str( + row.get("班级", row.get("班别", row.get("行政班", ""))) + ).strip() + gender = str(row.get("性别", "")).strip() + admission_no = str(row.get("录取编号", "")).strip() + remark = str(row.get("备注", "")).strip() + if not student_no or not name or not class_name: + errors.append(f"第 {index} 行：学号、姓名、班级均不能为空") + continue + if student_no in seen: + errors.append(f"第 {index} 行：学号 {student_no} 重复") + continue + seen.add(student_no) + students.append( + (class_name, student_no, name, gender, admission_no, remark) + ) + if errors: + raise ValueError("；".join(errors[:8])) + if not students: + raise ValueError("Excel 中没有可导入的学生") + + with self.lock, self.connection: + self.connection.executescript( + "DELETE FROM sessions; DELETE FROM group_members; DELETE FROM groups; DELETE FROM students;" + ) + self.connection.executemany( + """INSERT INTO students + (class_name,student_no,name,gender,admission_no,remark,password_hash,must_change_password) + VALUES(?,?,?,?,?,?,?,1)""", + [ + ( + item[0], + item[1], + item[2], + item[3], + item[4], + item[5], + hash_password(item[1]), + ) + for item in students + ], + ) + active = self.get_setting("active_class", conn=self.connection) or students[0][0] + if not any(item[0] == active for item in students): + active = students[0][0] + self.connection.execute( + "UPDATE settings SET value=? WHERE key='active_class'", (active,) + ) + return len(students), len({item[0] for item in students}) + + def get_setting(self, key: str, conn: sqlite3.Connection | None = None) -> str | None: + conn = conn or self.connection + row = conn.execute("SELECT value FROM settings WHERE key=?", (key,)).fetchone() + return None if row is None else row["value"] + + def set_setting(self, key: str, value: str) -> None: + with self.lock, self.connection: + self.connection.execute( + "INSERT INTO settings(key,value) VALUES(?,?) " + "ON CONFLICT(key) DO UPDATE SET value=excluded.value", + (key, value), + ) + + def class_list(self) -> list[dict[str, Any]]: + with self.lock: + rows = self.connection.execute( + """SELECT class_name, COUNT(*) AS student_count + FROM students GROUP BY class_name + ORDER BY CAST(class_name AS REAL), class_name""" + ).fetchall() + active = self.get_setting("active_class") + locked = { + row["class_name"]: bool(row["locked"]) + for row in self.connection.execute("SELECT class_name,locked FROM class_controls") + } + return [ + { + "class_name": row["class_name"], + "student_count": row["student_count"], + "locked": bool(locked.get(row["class_name"], False)), + "active": row["class_name"] == active, + } + for row in rows + ] + + def active_class(self) -> str: + return self.get_setting("active_class") or "" + + def is_locked(self, class_name: str) -> bool: + with self.lock: + row = self.connection.execute( + "SELECT locked FROM class_controls WHERE class_name=?", (class_name,) + ).fetchone() + return bool(row and row["locked"]) + + def create_session(self, role: str, username: str, student_id: int | None = None) -> str: + token = secrets.token_urlsafe(32) + token_hash = hashlib.sha256(token.encode()).hexdigest() + now = int(time.time()) + with self.lock, self.connection: + self.connection.execute( + "INSERT INTO sessions(token_hash,role,username,student_id,created_at,expires_at) VALUES(?,?,?,?,?,?)", + (token_hash, role, username, student_id, now, now + SESSION_DAYS * 86400), + ) + self.connection.execute( + "DELETE FROM sessions WHERE expires_at < ?", (now,) + ) + return token + + def get_session(self, token: str | None) -> dict[str, Any] | None: + if not token: + return None + token_hash = hashlib.sha256(token.encode()).hexdigest() + now = int(time.time()) + with self.lock: + row = self.connection.execute( + """SELECT s.token_hash,s.role,s.username,s.student_id,s.expires_at, + st.class_name,st.student_no,st.name,st.must_change_password + FROM sessions s LEFT JOIN students st ON st.id=s.student_id + WHERE s.token_hash=? AND s.expires_at>?""", + (token_hash, now), + ).fetchone() + if row is None: + return None + return dict(row) + + def destroy_session(self, token: str | None) -> None: + if not token: + return + token_hash = hashlib.sha256(token.encode()).hexdigest() + with self.lock, self.connection: + self.connection.execute("DELETE FROM sessions WHERE token_hash=?", (token_hash,)) + + def login_student(self, student_no: str, password: str) -> tuple[dict[str, Any] | None, str]: + student_no = student_no.strip() + active = self.active_class() + with self.lock: + student = self.connection.execute( + "SELECT * FROM students WHERE student_no=?", (student_no,) + ).fetchone() + if student is None or not verify_password(password, student["password_hash"]): + return None, "学号或密码不正确" + if student["class_name"] != active: + return None, f"当前仅开放 {active} 班登录，{student['class_name']} 班暂不能登录" + token = self.create_session("student", student["student_no"], student["id"]) + return { + "token": token, + "student": { + "id": student["id"], + "name": student["name"], + "student_no": student["student_no"], + "class_name": student["class_name"], + "must_change_password": bool(student["must_change_password"]), + }, + }, "登录成功" + + def login_teacher(self, username: str, password: str) -> tuple[dict[str, Any] | None, str]: + with self.lock: + expected_user = self.get_setting("teacher_username") or TEACHER_USER + expected_password = self.get_setting("teacher_password") + if username.strip() != expected_user or not verify_password(password, expected_password): + return None, "教师账号或密码不正确" + token = self.create_session("teacher", expected_user, None) + return {"token": token, "teacher": {"username": expected_user}}, "登录成功" + + def change_student_password( + self, session: dict[str, Any], old_password: str, new_password: str + ) -> tuple[bool, str]: + student_id = session.get("student_id") + if not student_id: + return False, "学生会话无效" + old_password = old_password or "" + new_password = new_password or "" + if len(new_password) < 8: + return False, "新密码至少 8 位" + if new_password == old_password: + return False, "新密码不能与原密码相同" + with self.lock: + student = self.connection.execute( + "SELECT * FROM students WHERE id=?", (student_id,) + ).fetchone() + if student is None or not verify_password(old_password, student["password_hash"]): + return False, "原密码不正确" + if new_password == student["student_no"]: + return False, "新密码不能设置为学号" + with self.connection: + self.connection.execute( + "UPDATE students SET password_hash=?, must_change_password=0 WHERE id=?", + (hash_password(new_password), student_id), + ) + token_hash = hashlib.sha256(session["_token"].encode()).hexdigest() + self.connection.execute( + "DELETE FROM sessions WHERE student_id=? AND token_hash<>?", + (student_id, token_hash), + ) + return True, "密码修改成功" + + def get_student(self, student_id: int) -> dict[str, Any] | None: + with self.lock: + row = self.connection.execute("SELECT * FROM students WHERE id=?", (student_id,)).fetchone() + return dict(row) if row else None + + def student_membership(self, student_id: int) -> dict[str, Any] | None: + with self.lock: + row = 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=?""", + (student_id,), + ).fetchone() + return dict(row) if row else None + + 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"]]) >= GROUP_LIMIT, + "members": members[row["id"]], + } + ) + return { + "class_name": class_name, + "locked": self.is_locked(class_name), + "group_limit": 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 "未命名小组" + 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"] + 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 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"] + 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, "小组已解散" + + def export_groups(self, class_name: str | None = None) -> tuple[bytes, str]: + with self.lock: + if class_name: + class_filter = "WHERE g.class_name=?" + params: tuple[Any, ...] = (class_name,) + else: + class_filter = "" + params = () + class_name = None + group_rows = self.connection.execute( + f"""SELECT g.id,g.class_name,g.name,s1.student_no leader_no,s1.name leader_name + FROM groups g LEFT JOIN students s1 ON s1.id=g.leader_student_id + {class_filter} ORDER BY g.class_name,g.id""", + params, + ).fetchall() + member_rows = self.connection.execute( + f"""SELECT m.group_id,s.student_no,s.name,s.gender,s.class_name + FROM group_members m JOIN groups g ON g.id=m.group_id + JOIN students s ON s.id=m.student_id + {class_filter} + ORDER BY g.class_name,g.id,CASE WHEN s.id=g.leader_student_id THEN 0 ELSE 1 END,s.student_no""", + params, + ).fetchall() + ungrouped_rows = self.connection.execute( + f"""SELECT s.class_name,s.student_no,s.name,s.gender + FROM students s WHERE s.id NOT IN (SELECT student_id FROM group_members) + {'AND s.class_name=?' if class_name else ''} + ORDER BY s.class_name,s.student_no""", + ((class_name,) if class_name else ()), + ).fetchall() + + by_group: dict[int, list[sqlite3.Row]] = {} + for row in member_rows: + by_group.setdefault(row["group_id"], []).append(row) + summary_headers = ["序号", "班级", "组名", "组内序号", "学号", "姓名", "性别", "是否组长"] + summary_records = [] + row_number = 0 + for group in group_rows: + for member_number, member in enumerate(by_group.get(group["id"], []), 1): + row_number += 1 + summary_records.append( + { + "序号": row_number, + "班级": member["class_name"], + "组名": group["name"], + "组内序号": member_number, + "学号": member["student_no"], + "姓名": member["name"], + "性别": member["gender"], + "是否组长": "是" if member["student_no"] == group["leader_no"] else "否", + } + ) + if not summary_records: + summary_records.append({ + "序号": "", "班级": "", "组名": "暂无分组", "组内序号": "", + "学号": "", "姓名": "", "性别": "", "是否组长": "", + }) + + ungrouped_headers = ["序号", "班级", "学号", "姓名", "性别", "说明"] + ungrouped_records = [ + { + "序号": index, + "班级": row["class_name"], + "学号": row["student_no"], + "姓名": row["name"], + "性别": row["gender"], + "说明": "未分组", + } + for index, row in enumerate(ungrouped_rows, 1) + ] + if not ungrouped_records: + ungrouped_records.append({ + "序号": "", "班级": "", "学号": "", "姓名": "", "性别": "", "说明": "全部学生均已分组" + }) + + title = f"{class_name}班" if class_name else "全部班级" + xlsx = write_xlsx( + [ + (f"{title}分组", summary_headers, summary_records), + ("未分组学生", ungrouped_headers, ungrouped_records), + ] + ) + return xlsx, export_filename("学生分组", class_name) + + +STORE = GroupStore(DB_PATH) + + +def json_response(status: int, payload: Any, cookies: list[dict[str, Any]] | None = None): + from http.server import BaseHTTPRequestHandler + + def send(handler: BaseHTTPRequestHandler): + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + handler.send_response(status) + handler.send_header("Content-Type", "application/json; charset=utf-8") + handler.send_header("Cache-Control", "no-store") + handler.send_header("Content-Length", str(len(body))) + for cookie in cookies or []: + handler.send_header("Set-Cookie", cookie["value"]) + handler.end_headers() + handler.wfile.write(body) + + return send + + +def make_session_cookie(token: str) -> str: + return f"GROUP_SESSION={token}; Path=/; HttpOnly; SameSite=Lax; Max-Age={SESSION_DAYS * 86400}" + + +def clear_session_cookie() -> str: + return "GROUP_SESSION=; Path=/; HttpOnly; SameSite=Lax; Max-Age=0" + + +class GroupRequestHandler: + server_version = "StudentGroupSystem/1.0" + + def __init__(self, handler): + self.handler = handler + + def cookies(self) -> dict[str, str]: + header = self.handler.headers.get("Cookie", "") + result = {} + for part in header.split(";"): + if "=" in part: + key, value = part.strip().split("=", 1) + result[key] = urllib.parse.unquote(value) + return result + + def token(self) -> str | None: + return self.cookies().get("GROUP_SESSION") + + def session(self) -> dict[str, Any] | None: + result = STORE.get_session(self.token()) + if result: + result["_token"] = self.token() + return result + + def read_json(self) -> dict[str, Any]: + length = int(self.handler.headers.get("Content-Length", "0") or 0) + if length <= 0: + return {} + if length > 1024 * 1024: + raise ValueError("请求体过大") + raw = self.handler.rfile.read(length) + return json.loads(raw.decode("utf-8")) + + def require_teacher(self): + session = self.session() + if not session or session["role"] != "teacher": + return None, json_response(HTTPStatus.UNAUTHORIZED, {"ok": False, "message": "请先登录教师后台"}) + return session, None + + def require_student(self): + session = self.session() + if not session or session["role"] != "student" or not session["student_id"]: + return None, json_response(HTTPStatus.UNAUTHORIZED, {"ok": False, "message": "请先登录"}) + return session, None + + def handle_api(self, action: str, method: str): + try: + return self.dispatch_api(action, method) + except (ValueError, sqlite3.IntegrityError, sqlite3.OperationalError) as exc: + LOG.warning("API error %s %s: %s", method, action, exc) + return json_response(HTTPStatus.BAD_REQUEST, {"ok": False, "message": str(exc)}) + except Exception: + LOG.exception("Unexpected API error") + return json_response(HTTPStatus.INTERNAL_SERVER_ERROR, {"ok": False, "message": "服务器内部错误"}) + + def dispatch_api(self, action: str, method: str): + if action == "login" and method == "POST": + data = self.read_json() + result, message = STORE.login_student(data.get("username", ""), data.get("password", "")) + if not result: + return json_response(HTTPStatus.UNAUTHORIZED, {"ok": False, "message": message}) + return json_response( + HTTPStatus.OK, + {"ok": True, "message": message, "user": result["student"]}, + [{"value": make_session_cookie(result["token"])}], + ) + + if action == "teacher-login" and method == "POST": + data = self.read_json() + result, message = STORE.login_teacher(data.get("username", ""), data.get("password", "")) + if not result: + return json_response(HTTPStatus.UNAUTHORIZED, {"ok": False, "message": message}) + return json_response( + HTTPStatus.OK, + {"ok": True, "message": message, "user": result["teacher"]}, + [{"value": make_session_cookie(result["token"])}], + ) + + if action == "logout" and method == "POST": + STORE.destroy_session(self.token()) + return json_response(HTTPStatus.OK, {"ok": True, "message": "已退出"}, [{"value": clear_session_cookie()}]) + + if action == "state" and method == "GET": + session = self.session() + if not session: + return json_response(HTTPStatus.OK, {"ok": True, "authenticated": False}) + active = STORE.active_class() + common = { + "authenticated": True, + "role": session["role"], + "active_class": active, + "classes": STORE.class_list() if session["role"] == "teacher" else [], + } + if session["role"] == "teacher": + selected = self.handler.path.split("class=", 1)[1].split("&", 1)[0] if "class=" in self.handler.path else active + selected = urllib.parse.unquote(selected) + if selected and selected != active: + # Teacher can preview any class, but active_class remains explicit. + return json_response( + HTTPStatus.OK, + { + **common, + "view_class": selected, + "snapshot": STORE.class_snapshot(selected), + }, + ) + return json_response( + HTTPStatus.OK, + { + **common, + "view_class": active, + "teacher": {"username": session["username"]}, + "snapshot": STORE.class_snapshot(active), + }, + ) + + student = STORE.get_student(session["student_id"]) + if student is None: + return json_response(HTTPStatus.OK, {"ok": True, "authenticated": False}) + membership = STORE.student_membership(student["id"]) + snapshot = STORE.class_snapshot(student["class_name"]) + return json_response( + HTTPStatus.OK, + { + **common, + "user": { + "id": student["id"], + "name": student["name"], + "student_no": student["student_no"], + "class_name": student["class_name"], + "must_change_password": bool(student["must_change_password"]), + }, + "membership": membership, + "snapshot": snapshot, + }, + ) + + if action == "change-password" and method == "POST": + session, error = self.require_student() + if error: + return error + data = self.read_json() + ok, message = STORE.change_student_password( + session, data.get("old_password", ""), data.get("new_password", "") + ) + status = HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST + return json_response(status, {"ok": ok, "message": message}) + + if action == "groups" and method == "POST": + session, error = self.require_student() + if error: + return error + student = STORE.get_student(session["student_id"]) + if student and student["must_change_password"]: + return json_response(HTTPStatus.FORBIDDEN, {"ok": False, "message": "首次登录必须先修改密码"}) + data = self.read_json() + ok, message, _ = STORE.create_group(session["student_id"], data.get("name", "")) + return json_response( + HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, + {"ok": ok, "message": message}, + ) + + if action == "join" and method == "POST": + session, error = self.require_student() + if error: + return error + data = self.read_json() + ok, message = STORE.join_group( + session["student_id"], int(data.get("group_id", 0)), bool(data.get("confirmed")) + ) + return json_response( + HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, + {"ok": ok, "message": message, "require_confirm": (not ok and "确认后才会执行" in message)}, + ) + + if action == "leave" and method == "POST": + session, error = self.require_student() + if error: + return error + data = self.read_json() + ok, message = STORE.leave_group(session["student_id"], bool(data.get("confirmed"))) + return json_response( + HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, + {"ok": ok, "message": message, "require_confirm": (not ok and "确认后才会执行" in message)}, + ) + + if action == "teacher/class" and method == "POST": + _, error = self.require_teacher() + if error: + return error + data = self.read_json() + ok, message = STORE.switch_class(str(data.get("class_name", ""))) + return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {"ok": ok, "message": message}) + + if action == "teacher/lock" and method == "POST": + _, error = self.require_teacher() + if error: + return error + data = self.read_json() + class_name = str(data.get("class_name", STORE.active_class())) + ok, message = STORE.set_locked(class_name, bool(data.get("locked"))) + return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {"ok": ok, "message": message}) + + if action == "teacher/create-group" and method == "POST": + _, error = self.require_teacher() + if error: + return error + data = self.read_json() + ok, message = STORE.teacher_create_group( + str(data.get("class_name", STORE.active_class())), data.get("name", "") + ) + return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {"ok": ok, "message": message}) + + if action == "teacher/rename-group" and method == "POST": + _, error = self.require_teacher() + if error: + return error + data = self.read_json() + ok, message = STORE.teacher_rename_group(int(data.get("group_id", 0)), data.get("name", "")) + return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {"ok": ok, "message": message}) + + if action == "teacher/set-leader" and method == "POST": + _, error = self.require_teacher() + if error: + return error + data = self.read_json() + student_id = data.get("student_id") + ok, message = STORE.teacher_set_leader( + int(data.get("group_id", 0)), int(student_id) if student_id else None + ) + return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {"ok": ok, "message": message}) + + if action == "teacher/add-member" and method == "POST": + _, error = self.require_teacher() + if error: + return error + data = self.read_json() + ok, message = STORE.teacher_add_member(int(data.get("group_id", 0)), int(data.get("student_id", 0))) + return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {"ok": ok, "message": message}) + + if action == "teacher/remove-member" and method == "POST": + _, error = self.require_teacher() + if error: + return error + data = self.read_json() + ok, message = STORE.teacher_remove_member(int(data.get("student_id", 0))) + return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {"ok": ok, "message": message}) + + if action == "teacher/disband" and method == "POST": + _, error = self.require_teacher() + if error: + return error + data = self.read_json() + ok, message = STORE.teacher_disband_group(int(data.get("group_id", 0))) + return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {"ok": ok, "message": message}) + + if action == "teacher/import" and method == "POST": + _, error = self.require_teacher() + if error: + return error + length = int(self.handler.headers.get("Content-Length", "0") or 0) + if length <= 0 or length > 10 * 1024 * 1024: + return json_response(HTTPStatus.BAD_REQUEST, {"ok": False, "message": "Excel 文件大小无效"}) + raw = self.handler.rfile.read(length) + temp_path = BASE_DIR / ".upload-roster.xlsx" + temp_path.write_bytes(raw) + try: + student_count, class_count = STORE.import_roster(temp_path) + finally: + temp_path.unlink(missing_ok=True) + return json_response( + HTTPStatus.OK, + {"ok": True, "message": f"导入成功：{student_count} 名学生，{class_count} 个班。所有原分组已重置。"}, + ) + + if action == "teacher/export" and method == "GET": + _, error = self.require_teacher() + if error: + return error + query = urllib.parse.parse_qs(urllib.parse.urlparse(self.handler.path).query) + class_name = query.get("class", [None])[0] + if class_name: + class_name = urllib.parse.unquote(class_name) + content, filename = STORE.export_groups(class_name) + + def send(handler): + handler.send_response(HTTPStatus.OK) + handler.send_header("Content-Type", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") + quoted = urllib.parse.quote(filename) + handler.send_header("Content-Disposition", f"attachment; filename*=UTF-8''{quoted}") + handler.send_header("Content-Length", str(len(content))) + handler.send_header("Cache-Control", "no-store") + handler.end_headers() + handler.wfile.write(content) + + return send + + return json_response(HTTPStatus.NOT_FOUND, {"ok": False, "message": "接口不存在"}) + + +def route(handler): + parsed = urllib.parse.urlparse(handler.path) + path = parsed.path + request = GroupRequestHandler(handler) + if path.startswith("/api/"): + result = request.handle_api(path[len("/api/"):], handler.command) + if callable(result): + result(handler) + return + + if path in ("/", "/index.html", "/student"): + return serve_file(handler, STATIC_DIR / "index.html") + if path in ("/teacher", "/admin"): + return serve_file(handler, STATIC_DIR / "teacher.html") + safe_path = (STATIC_DIR / urllib.parse.unquote(path).lstrip("/")).resolve() + try: + safe_path.relative_to(STATIC_DIR.resolve()) + except ValueError: + return send_text(handler, HTTPStatus.NOT_FOUND, "Not Found") + return serve_file(handler, safe_path) + + +def serve_file(handler, path: Path): + from http.server import SimpleHTTPRequestHandler + + path = path if path.is_absolute() else STATIC_DIR / path + if not path.is_file(): + return send_text(handler, HTTPStatus.NOT_FOUND, "Not Found") + suffix = path.suffix.lower() + if suffix in (".html", ".htm"): + guess = "text/html; charset=utf-8" + elif suffix == ".css": + guess = "text/css; charset=utf-8" + elif suffix in (".js", ".mjs"): + guess = "text/javascript; charset=utf-8" + elif suffix == ".svg": + guess = "image/svg+xml" + elif suffix == ".png": + guess = "image/png" + else: + guess = "application/octet-stream" + content = path.read_bytes() + handler.send_response(HTTPStatus.OK) + handler.send_header("Content-Type", guess or "application/octet-stream") + handler.send_header("Content-Length", str(len(content))) + handler.send_header("Cache-Control", "no-cache") + handler.end_headers() + handler.wfile.write(content) + + +def send_text(handler, status: int, text: str): + body = text.encode("utf-8") + handler.send_response(status) + handler.send_header("Content-Type", "text/plain; charset=utf-8") + handler.send_header("Content-Length", str(len(body))) + handler.end_headers() + handler.wfile.write(body) + + +class Handler(http.server.BaseHTTPRequestHandler): + def do_GET(self): + route(self) + + def do_POST(self): + route(self) + + def do_HEAD(self): + route(self) + + def log_message(self, format_string, *args): + LOG.info("%s - %s", self.address_string(), format_string % args) + + +def main() -> None: + DB_PATH.parent.mkdir(parents=True, exist_ok=True) + if not any(STORE.class_list()): + if XLSX_PATH.exists(): + student_count, class_count = STORE.import_roster(XLSX_PATH) + LOG.info("已导入 %s： %d 名学生 / %d 个班", XLSX_PATH, student_count, class_count) + else: + LOG.warning("未找到 %s，请先通过教师后台上传学生名单", XLSX_PATH) + + host = os.environ.get("HOST", "0.0.0.0") + port = int(os.environ.get("PORT", "8000")) + server = http.server.ThreadingHTTPServer((host, port), Handler) + LOG.info("学生分组系统已启动：http://0.0.0.0:%d", port) + LOG.info("学生入口：http://<老师电脑IP>:%d/ 教师入口：http://<老师电脑IP>:%d/teacher", port, port) + try: + server.serve_forever() + except KeyboardInterrupt: + LOG.info("服务停止") + finally: + server.server_close() + + +if __name__ == "__main__": + main() diff --git a/start.sh b/start.sh new file mode 100755 index 0000000..bf87bbf --- /dev/null +++ b/start.sh @@ -0,0 +1,4 @@ +#!/bin/sh +set -eu +cd "$(dirname "$0")" +exec python3 app.py diff --git a/static/index.html b/static/index.html new file mode 100644 index 0000000..9a95b89 --- /dev/null +++ b/static/index.html @@ -0,0 +1,91 @@ + + + + + + 学生自助分组 + + + +
+
组学生自助分组系统
+ +
+ +
+ + + +
+ + + + + + + diff --git a/static/shared.js b/static/shared.js new file mode 100644 index 0000000..7f092e9 --- /dev/null +++ b/static/shared.js @@ -0,0 +1,53 @@ +async function api(path, options = {}) { + const config = { + method: options.method || 'GET', + headers: {}, + credentials: 'same-origin' + }; + if (options.json !== undefined) { + config.headers['Content-Type'] = 'application/json'; + config.body = JSON.stringify(options.json); + } else if (options.body) { + config.body = options.body; + } + const response = await fetch(`/api/${path}`, config); + let data; + try { + data = await response.json(); + } catch (_) { + data = { ok: false, message: '服务器响应异常，请刷新后重试' }; + } + if (!response.ok && !data.message) data.message = `请求失败（${response.status}）`; + return data; +} + +function toast(message, isError = false) { + let element = document.querySelector('#toast'); + if (!element) { + element = document.createElement('div'); + element.id = 'toast'; + element.className = 'toast'; + document.body.appendChild(element); + } + element.textContent = message; + element.classList.toggle('error', Boolean(isError)); + element.classList.add('show'); + clearTimeout(element.timer); + element.timer = setTimeout(() => element.classList.remove('show'), 3800); +} + +function text(value) { + return String(value ?? '').replace(/[&<>"']/g, ch => ({ + '&': '&', '<': '<', '>': '>', '"': '"', "'": ''' + }[ch])); +} + +function requireConfirm(message) { + return window.confirm(message); +} + +function setButtonsDisabled(root, disabled) { + root.querySelectorAll('button[data-working]').forEach(button => { + button.disabled = disabled || button.dataset.alwaysDisabled === '1'; + }); +} diff --git a/static/student.js b/static/student.js new file mode 100644 index 0000000..d34aafe --- /dev/null +++ b/static/student.js @@ -0,0 +1,211 @@ +const state = { user: null, membership: null, snapshot: null }; + +const el = id => document.getElementById(id); + +function renderUnauthenticated() { + el('loginArea').classList.remove('hidden'); + el('mainArea').classList.add('hidden'); + el('userToolbar').classList.add('hidden'); + el('passwordModal').classList.add('hidden'); +} + +function renderStatistics() { + const stats = state.snapshot.statistics; + const remaining = state.snapshot.statistics.total_students - state.snapshot.statistics.grouped_students; + el('statistics').innerHTML = [ + { label: '班级总人数', value: stats.total_students }, + { label: '已分组人数', value: stats.grouped_students }, + { label: '未分组人数', value: remaining } + ].map(item => `
${text(item.value)}${item.label}
`).join(''); +} + +function renderIdentity() { + const user = state.user; + el('userChip').textContent = `${user.class_name}班 · ${user.name} · ${user.student_no}`; + el('identity').innerHTML = [ + ['姓名', user.name], ['学号', user.student_no], ['班级', `${user.class_name}班`], ['登录状态', '已登录'] + ].map(row => `${text(row[0])}${text(row[1])}`).join(''); + el('classStatus').innerHTML = [ + `${text(user.class_name)}班`, + `${state.snapshot.locked ? '老师已停止操作' : '分组进行中'}` + ].join(''); +} + +function renderMyGroup() { + const target = el('myGroupAction'); + const membership = state.membership; + if (!membership) { + target.innerHTML = ` +
+ +
+ + +
+

创建后你是组长。每组成员（包括组长）最多 ${state.snapshot.group_limit} 人。已加入别人的组时不能创建。

+
`; + el('createForm').addEventListener('submit', submitCreate); + return; + } + + const isLeader = membership.leader_student_id === state.user.id; + target.innerHTML = ` +
+ 当前组：${text(membership.name)} + ${isLeader ? '我是组长' : '我是组员'} +
+
+ +
+

${isLeader + ? '组长退出会解散整个组，所有成员都会变为未分组。' + : '退出后可以加入其他组。已加入其他组时，不能创建新组。'}

`; + el('leaveButton').addEventListener('click', submitLeave); +} + +function memberLine(member) { + return `${text(member.student_no)} · ${text(member.name)}${member.is_leader ? ' 组长' : ''}`; +} + +function renderGroups() { + const grid = el('groupsGrid'); + if (!state.snapshot.groups.length) { + grid.innerHTML = '
当前班级还没有小组。你可以创建第一个组。
'; + return; + } + grid.innerHTML = state.snapshot.groups.map(group => { + const own = state.membership && state.membership.group_id === group.id; + const locked = state.snapshot.locked; + const full = group.full; + let disabledReason = ''; + if (locked) disabledReason = '老师已停止分组操作，当前只能查看分组结果'; + else if (full) disabledReason = `该组已有 ${state.snapshot.group_limit} 人，不能再加入新成员`; + return ` +
+
+
${text(group.name)}
+ ${group.member_count}/${state.snapshot.group_limit} 人 +
+
    + ${group.members.length ? group.members.map(member => `
  • ${memberLine(member)}
    • `).join('') : '
  • 暂无成员
    • '} +
+ ${own ? '我的当前组' + : ``} +
`; + }).join(''); + grid.querySelectorAll('button[data-group]').forEach(button => button.addEventListener('click', () => { + if (button.dataset.reason) { + toast(button.dataset.reason, true); + return; + } + submitJoin(Number(button.dataset.group)); + })); +} + +function renderState() { + if (!state.user) return renderUnauthenticated(); + el('loginArea').classList.add('hidden'); + el('mainArea').classList.remove('hidden'); + el('userToolbar').classList.remove('hidden'); + renderStatistics(); + renderIdentity(); + renderMyGroup(); + renderGroups(); + if (state.user.must_change_password) { + el('passwordModal').classList.remove('hidden'); + } else { + el('passwordModal').classList.add('hidden'); + } +} + +async function loadState() { + const result = await api('state'); + if (!result.ok || !result.authenticated) return renderUnauthenticated(); + state.user = result.user; + state.membership = result.membership; + state.snapshot = result.snapshot; + renderState(); +} + +async function submitLogin(event) { + event.preventDefault(); + setButtonsDisabled(el('loginForm'), true); + const result = await api('login', { + method: 'POST', + json: { username: el('username').value.trim(), password: el('password').value } + }); + setButtonsDisabled(el('loginForm'), false); + el('loginNotice').textContent = result.ok ? '' : result.message; + if (result.ok) { + toast(result.message); + await loadState(); + } +} + +async function submitCreate(event) { + event.preventDefault(); + const result = await api('groups', { method: 'POST', json: { name: el('groupName').value } }); + toast(result.message, !result.ok); + if (result.ok) await loadState(); +} + +async function submitJoin(groupId, confirmed = false) { + const result = await api('join', { method: 'POST', json: { group_id: groupId, confirmed } }); + if (!result.ok && result.require_confirm) { + if (requireConfirm(result.message)) return submitJoin(groupId, true); + toast('已取消加入'); + return; + } + toast(result.message, !result.ok); + if (result.ok) await loadState(); +} + +async function submitLeave(confirmed = false) { + const result = await api('leave', { method: 'POST', json: { confirmed } }); + if (!result.ok && result.require_confirm) { + if (requireConfirm(result.message)) return submitLeave(true); + toast('已取消操作'); + return; + } + toast(result.message, !result.ok); + if (result.ok) await loadState(); +} + +async function submitPassword(prefix) { + const oldInput = el(`${prefix}Old`); + const newInput = el(`${prefix}New`); + const confirmInput = el(`${prefix}Confirm`); + const form = prefix === 'first' ? el('firstPasswordForm') : el('passwordForm'); + const notice = prefix === 'first' ? el('firstPasswordNotice') : null; + if (newInput.value !== confirmInput.value) { + if (notice) notice.textContent = '两次输入的新密码不一致'; + else toast('两次输入的新密码不一致', true); + return; + } + setButtonsDisabled(form, true); + const result = await api('change-password', { + method: 'POST', + json: { old_password: oldInput.value, new_password: newInput.value } + }); + setButtonsDisabled(form, false); + if (notice) notice.textContent = result.ok ? '' : result.message; + toast(result.message, !result.ok); + if (result.ok) { + form.reset(); + await loadState(); + } +} + +async function logout() { + await api('logout', { method: 'POST' }); + state.user = null; + renderUnauthenticated(); +} + +el('loginForm').addEventListener('submit', submitLogin); +el('logoutButton').addEventListener('click', logout); +el('logoutFromModal').addEventListener('click', logout); +el('passwordForm').addEventListener('submit', event => { event.preventDefault(); submitPassword(''); }); +el('firstPasswordForm').addEventListener('submit', event => { event.preventDefault(); submitPassword('first'); }); + +loadState(); diff --git a/static/style.css b/static/style.css new file mode 100644 index 0000000..eae27dd --- /dev/null +++ b/static/style.css @@ -0,0 +1,156 @@ +:root { + --bg: #f3f6fb; + --panel: #ffffff; + --text: #18212f; + --muted: #64748b; + --primary: #2563eb; + --primary-dark: #1d4ed8; + --success: #059669; + --danger: #dc2626; + --warning: #d97706; + --border: #dbe2ea; + --shadow: 0 10px 30px rgba(15, 23, 42, .08); +} + +* { box-sizing: border-box; } +html, body { margin: 0; min-height: 100%; } +body { + color: var(--text); + font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", "PingFang SC", "Microsoft YaHei", sans-serif; + background: radial-gradient(circle at top left, #e0edff 0, var(--bg) 38%); +} + +.topbar { + min-height: 64px; + padding: 14px max(20px, env(safe-area-inset-left)); + display: flex; + align-items: center; + justify-content: space-between; + gap: 16px; + background: rgba(255,255,255,.92); + border-bottom: 1px solid var(--border); + position: sticky; + top: 0; + z-index: 10; +} +.brand { display: flex; align-items: center; gap: 10px; font-weight: 750; font-size: 19px; } +.brand-mark { + width: 34px; height: 34px; border-radius: 9px; color: white; + display: grid; place-items: center; background: linear-gradient(135deg,#2563eb,#7c3aed); +} +.toolbar { display: flex; align-items: center; justify-content: flex-end; gap: 8px; flex-wrap: wrap; } + +.page { max-width: 1280px; margin: 24px auto 60px; padding: 0 20px; } +.login-wrap { min-height: calc(100vh - 80px); display: grid; place-items: center; } +.login-card, .panel { + width: min(430px, calc(100vw - 32px)); + background: var(--panel); border: 1px solid var(--border); + border-radius: 16px; padding: 28px; box-shadow: var(--shadow); +} +.panel { width: 100%; margin-bottom: 18px; } +.title { margin: 0 0 6px; font-size: 24px; } +.subtitle { margin: 0 0 22px; color: var(--muted); line-height: 1.6; } + +label { display: block; margin: 14px 0 6px; font-size: 14px; font-weight: 650; } +input, select { + width: 100%; height: 42px; border: 1px solid var(--border); border-radius: 8px; + padding: 0 12px; font-size: 15px; color: var(--text); background: #fff; outline: none; +} +input:focus, select:focus { border-color: var(--primary); box-shadow: 0 0 0 3px rgba(37,99,235,.13); } +button { + height: 38px; border: 0; border-radius: 8px; padding: 0 14px; font-size: 14px; + cursor: pointer; color: white; background: var(--primary); transition: .15s; +} +button:hover:not(:disabled) { background: var(--primary-dark); transform: translateY(-1px); } +button:disabled { cursor: not-allowed; opacity: .5; transform: none; } +button.full { width: 100%; height: 42px; margin-top: 18px; font-weight: 650; } +button.ghost { color: var(--text); background: #eef2f7; } +button.success { background: var(--success); } +button.danger { background: var(--danger); } +button.warning { background: var(--warning); } +button.small { height: 30px; padding: 0 8px; font-size: 12px; white-space: nowrap; } + +.hidden { display: none !important; } +.notice { + min-height: 20px; margin-top: 12px; font-size: 14px; color: var(--danger); + text-align: center; white-space: pre-wrap; +} +.hint { color: var(--muted); font-size: 13px; line-height: 1.6; } +.status-row { display: flex; gap: 8px; flex-wrap: wrap; align-items: center; } +.chip { + display: inline-flex; align-items: center; min-height: 28px; padding: 3px 10px; + border-radius: 999px; background: #eef2f7; color: #334155; font-size: 13px; white-space: nowrap; +} +.chip.active { background: #dcfce7; color: #166534; } +.chip.locked { background: #ffedd5; color: #9a3412; } +.chip.leader { background: #dbeafe; color: #1e40af; } +.chip.member { background: #f1f5f9; color: #475569; } +.chip.full { background: #fee2e2; color: #991b1b; } +.section-head { display: flex; justify-content: space-between; align-items: center; gap: 14px; margin-bottom: 14px; flex-wrap: wrap; } +h1,h2,h3 { line-height: 1.25; } +h2 { margin: 0; font-size: 20px; } +h3 { margin: 0; font-size: 17px; } +.class-tabs { display: flex; gap: 8px; flex-wrap: wrap; } +.class-tab { + height: 34px; padding: 0 13px; border-radius: 8px; background: #fff; + border: 1px solid var(--border); color: #334155; +} +.class-tab.active { border-color: transparent; background: var(--primary); color: #fff; } + +.grid { + display: grid; grid-template-columns: repeat(auto-fill,minmax(290px,1fr)); gap: 16px; +} +.group-card { + background: var(--panel); border: 1px solid var(--border); border-radius: 14px; + padding: 15px; box-shadow: 0 5px 16px rgba(15,23,42,.05); align-self: start; +} +.group-card.own { border-color: var(--primary); box-shadow: 0 8px 24px rgba(37,99,235,.14); } +.group-top { display: flex; justify-content: space-between; align-items: center; gap: 8px; margin-bottom: 12px; } +.group-title { font-weight: 750; overflow-wrap: anywhere; } +.member-list { margin: 0 0 12px; padding: 0; list-style: none; } +.member-list li { + min-height: 42px; padding: 7px 0; border-top: 1px solid #edf2f7; + display: flex; align-items: center; justify-content: space-between; gap: 8px; +} +.member-list li:first-child { border-top: 0; } +.member-name { min-width: 0; overflow-wrap: anywhere; } +.member-actions { display: flex; gap: 5px; flex-shrink: 0; } +.empty { color: var(--muted); text-align: center; padding: 24px 0; } +.form-line { display: flex; gap: 8px; align-items: center; } +.form-line input, .form-line select { flex: 1; min-width: 0; } +.actions { display: flex; gap: 8px; flex-wrap: wrap; } +.statistics { display: grid; grid-template-columns: repeat(3,1fr); gap: 10px; margin-bottom: 16px; } +.statistic { background: #fff; border: 1px solid var(--border); border-radius: 12px; padding: 13px; } +.statistic strong { display: block; font-size: 24px; } +.statistic span { color: var(--muted); font-size: 13px; } +.two-col { display: grid; grid-template-columns: 340px 1fr; gap: 18px; align-items: start; } +.identity { display: grid; grid-template-columns: auto 1fr; gap: 6px 14px; margin: 14px 0 18px; } +.identity span:first-child { color: var(--muted); } + +.modal-mask { + position: fixed; inset: 0; z-index: 100; display: grid; place-items: center; + background: rgba(15,23,42,.55); padding: 18px; backdrop-filter: blur(3px); +} +.modal { + width: min(460px,100%); background: #fff; border-radius: 15px; padding: 26px; box-shadow: var(--shadow); +} +.modal-actions { display: grid; grid-template-columns: 1fr 1fr; gap: 10px; margin-top: 20px; } +.toast { + position: fixed; left: 50%; bottom: 30px; transform: translateX(-50%) translateY(20px); + max-width: min(640px, calc(100vw - 32px)); padding: 13px 18px; border-radius: 10px; + background: #0f172a; color: #fff; box-shadow: var(--shadow); z-index: 200; opacity: 0; + transition: .2s; pointer-events: none; font-size: 14px; line-height: 1.5; +} +.toast.show { opacity: 1; transform: translateX(-50%) translateY(0); } +.toast.error { background: #b91c1c; } +.loading { color: var(--muted); text-align: center; padding: 40px; } + +@media (max-width: 800px) { + .page { margin-top: 14px; } + .two-col { grid-template-columns: 1fr; } + .statistics { grid-template-columns: 1fr; } + .topbar { padding: 12px 14px; } + .brand span { display: none; } + .group-card { padding: 13px; } + .member-actions button { padding: 0 6px; } +} diff --git a/static/teacher.html b/static/teacher.html new file mode 100644 index 0000000..ff2ba05 --- /dev/null +++ b/static/teacher.html @@ -0,0 +1,75 @@ + + + + + + 分组管理后台 + + + +
+
师分组管理后台
+ +
+ +
+ + + +
+ + + + + diff --git a/static/teacher.js b/static/teacher.js new file mode 100644 index 0000000..58494bb --- /dev/null +++ b/static/teacher.js @@ -0,0 +1,225 @@ +const state = { classes: [], snapshot: null, activeClass: '', username: '' }; +const el = id => document.getElementById(id); + +function renderUnauthenticated() { + el('loginArea').classList.remove('hidden'); + el('mainArea').classList.add('hidden'); + el('teacherToolbar').classList.add('hidden'); +} + +function renderState() { + el('loginArea').classList.add('hidden'); + el('mainArea').classList.remove('hidden'); + el('teacherToolbar').classList.remove('hidden'); + el('teacherChip').textContent = `教师：${state.username}`; + renderClassTabs(); + renderStatistics(); + renderGroups(); + renderUngrouped(); +} + +function renderClassTabs() { + el('classTabs').innerHTML = state.classes.map(item => ` + + `).join(''); + const current = state.classes.find(item => item.class_name === state.activeClass); + const locked = current ? current.locked : false; + el('managementStatus').innerHTML = [ + `开放登录：${text(state.activeClass)}班`, + `${locked ? '已停止学生操作' : '学生可正常操作'}`, + `${state.snapshot.statistics.group_count} 个小组` + ].join(''); + el('lockButton').textContent = locked ? '恢复学生操作' : '停止学生操作'; + el('lockButton').className = locked ? 'success' : 'warning'; + el('exportButton').disabled = false; + el('classTabs').querySelectorAll('button[data-class]').forEach(button => { + button.addEventListener('click', () => switchClass(button.dataset.class)); + }); +} + +function renderStatistics() { + const stats = state.snapshot.statistics; + el('statistics').innerHTML = [ + { label: '班级总人数', value: stats.total_students }, + { label: '已分组人数', value: stats.grouped_students }, + { label: '未分组人数', value: stats.total_students - stats.grouped_students } + ].map(item => `
${text(item.value)}${item.label}
`).join(''); +} + +function memberActions(member, group) { + const isLeader = member.id === group.leader_student_id; + return ` + + + + `; +} + +function renderGroups() { + const grid = el('groupsGrid'); + if (!state.snapshot.groups.length) { + grid.innerHTML = '
该班还没有小组，可先手工创建，或等学生自行创建。
'; + return; + } + grid.innerHTML = state.snapshot.groups.map(group => ` +
+
+
+ + +
+ ${group.member_count}/${state.snapshot.group_limit} 人 +
+
    + ${group.members.length ? group.members.map(member => `
  • ${text(member.student_no)} · ${text(member.name)} ${member.is_leader ? '组长' : ''}${memberActions(member, group)}
    • `).join('') : '
  • 暂无成员
    • '} +
+
+ +
+
+ `).join(''); + + grid.querySelectorAll('form[data-rename]').forEach(form => form.addEventListener('submit', async event => { + event.preventDefault(); + const result = await api('teacher/rename-group', { + method: 'POST', json: { group_id: Number(form.dataset.rename), name: form.querySelector('input').value } + }); + toast(result.message, !result.ok); + if (result.ok) await loadState(); + })); + bindGridActions(grid); +} + +function bindGridActions(root) { + root.querySelectorAll('button[data-action]').forEach(button => button.addEventListener('click', async () => { + const action = button.dataset.action; + if (action === 'disband' && !requireConfirm('解散后所有成员都会变为未分组，确定继续吗？')) return; + if (action === 'remove' && !requireConfirm('确定将该学生移出当前小组吗？')) return; + let result; + if (action === 'leader') { + const groupId = Number(button.dataset.group); + const studentId = button.textContent.trim() === '取消组长' ? null : Number(button.dataset.student); + result = await api('teacher/set-leader', { method: 'POST', json: { group_id: groupId, student_id: studentId } }); + } else if (action === 'remove') { + result = await api('teacher/remove-member', { method: 'POST', json: { student_id: Number(button.dataset.student) } }); + } else if (action === 'disband') { + result = await api('teacher/disband', { method: 'POST', json: { group_id: Number(button.dataset.group) } }); + } + toast(result.message, !result.ok); + if (result.ok) await loadState(); + })); +} + +function renderUngrouped() { + const container = el('ungroupedList'); + const students = state.snapshot.ungrouped; + if (!students.length) { + container.innerHTML = '
没有未分组学生。
'; + return; + } + const options = state.snapshot.groups.map(group => ``).join(''); + container.innerHTML = ` +
+ ${students.map(student => ` +
+
${text(student.student_no)} · ${text(student.name)}${text(student.class_name)}班
+
+ + +
+
`).join('')} +
`; + container.querySelectorAll('form[data-add]').forEach(form => form.addEventListener('submit', async event => { + event.preventDefault(); + const group = Number(form.querySelector('select').value); + if (!group) return; + const result = await api('teacher/add-member', { + method: 'POST', json: { group_id: group, student_id: Number(form.dataset.add) } + }); + toast(result.message, !result.ok); + if (result.ok) await loadState(); + })); +} + +async function loadState() { + const result = await api('state'); + if (!result.ok || !result.authenticated || result.role !== 'teacher') return renderUnauthenticated(); + state.classes = result.classes; + state.snapshot = result.snapshot; + state.activeClass = result.active_class; + state.username = result.teacher.username; + renderState(); +} + +async function submitLogin(event) { + event.preventDefault(); + setButtonsDisabled(el('loginForm'), true); + const result = await api('teacher-login', { + method: 'POST', json: { username: el('username').value.trim(), password: el('password').value } + }); + setButtonsDisabled(el('loginForm'), false); + el('loginNotice').textContent = result.ok ? '' : result.message; + if (result.ok) { + toast(result.message); + await loadState(); + } +} + +async function switchClass(className) { + const result = await api('teacher/class', { method: 'POST', json: { class_name: className } }); + toast(result.message, !result.ok); + if (result.ok) await loadState(); +} + +async function toggleLock() { + const current = state.classes.find(item => item.class_name === state.activeClass); + const locked = !current || !current.locked; + const message = locked ? '停止后学生只能查看分组结果，不能再创建/加入/退出。确定继续吗？' : '确定恢复学生分组操作吗？'; + if (!requireConfirm(message)) return; + const result = await api('teacher/lock', { method: 'POST', json: { class_name: state.activeClass, locked } }); + toast(result.message, !result.ok); + if (result.ok) await loadState(); +} + +async function submitCreateGroup(event) { + event.preventDefault(); + const result = await api('teacher/create-group', { + method: 'POST', json: { class_name: state.activeClass, name: el('newGroupName').value } + }); + toast(result.message, !result.ok); + if (result.ok) { + el('newGroupName').value = ''; + await loadState(); + } +} + +async function importRoster(file) { + if (!file) return; + if (!requireConfirm('导入名单会重建全部学生账号和登录密码，并解散所有班级的已有分组。确定继续吗？')) return; + toast('正在导入 Excel，请稍候...'); + const result = await api(`teacher/import?filename=${encodeURIComponent(file.name)}`, { + method: 'POST', headers: { 'Content-Type': 'application/octet-stream' }, body: file + }); + toast(result.message, !result.ok); + if (result.ok) await loadState(); +} + +async function logout() { + await api('logout', { method: 'POST' }); + renderUnauthenticated(); +} + +el('loginForm').addEventListener('submit', submitLogin); +el('logoutButton').addEventListener('click', logout); +el('lockButton').addEventListener('click', toggleLock); +el('createGroupForm').addEventListener('submit', submitCreateGroup); +el('exportButton').addEventListener('click', () => { + window.location.href = `/api/teacher/export?class=${encodeURIComponent(state.activeClass)}`; +}); +el('importFile').addEventListener('change', event => { + importRoster(event.target.files[0]); + event.target.value = ''; +}); + +loadState(); diff --git a/xlsxio.py b/xlsxio.py new file mode 100644 index 0000000..44dbd90 --- /dev/null +++ b/xlsxio.py @@ -0,0 +1,242 @@ +#!/usr/bin/env python3 +"""Minimal XLSX reader/writer using only the Python standard library.""" + +from __future__ import annotations + +import re +from datetime import datetime +from pathlib import Path +from xml.etree import ElementTree as ET +from zipfile import ZIP_DEFLATED, ZipFile + + +MAIN_NS = "http://schemas.openxmlformats.org/spreadsheetml/2006/main" +REL_NS = "http://schemas.openxmlformats.org/officeDocument/2006/relationships" +PKG_REL_NS = "http://schemas.openxmlformats.org/package/2006/relationships" +NS = {"m": MAIN_NS, "r": REL_NS, "p": PKG_REL_NS} + + +def _column_number(cell_ref: str) -> int: + letters = re.match(r"([A-Z]+)", cell_ref.upper()) + if not letters: + return 0 + value = 0 + for letter in letters.group(1): + value = value * 26 + ord(letter) - 64 + return value + + +def _column_letter(number: int) -> str: + result = "" + while number: + number, remainder = divmod(number - 1, 26) + result = chr(65 + remainder) + result + return result + + +def _shared_strings(archive: ZipFile) -> list[str]: + if "xl/sharedStrings.xml" not in archive.namelist(): + return [] + root = ET.fromstring(archive.read("xl/sharedStrings.xml")) + values = [] + for item in root.findall("m:si", NS): + values.append("".join(node.text or "" for node in item.iter(f"{{{MAIN_NS}}}t"))) + return values + + +def _relationships(archive: ZipFile) -> dict[str, str]: + root = ET.fromstring(archive.read("xl/_rels/workbook.xml.rels")) + return {item.attrib["Id"]: item.attrib["Target"] for item in root} + + +def _cell_value(cell, shared: list[str]) -> str: + cell_type = cell.attrib.get("t") + inline = cell.find("m:is", NS) + if cell_type == "inlineStr" and inline is not None: + return "".join(node.text or "" for node in inline.iter(f"{{{MAIN_NS}}}t")).strip() + value_node = cell.find("m:v", NS) + raw = "" if value_node is None else (value_node.text or "") + if cell_type == "s" and raw: + return shared[int(raw)].strip() + return raw.strip() + + +def read_xlsx_rows(path: str | Path) -> list[dict[str, str]]: + """Read the first worksheet as dictionaries keyed by header text.""" + path = Path(path) + with ZipFile(path, "r") as archive: + shared = _shared_strings(archive) + relationships = _relationships(archive) + workbook = ET.fromstring(archive.read("xl/workbook.xml")) + sheets = workbook.findall("m:sheets/m:sheet", NS) + if not sheets: + raise ValueError("Excel 文件中没有工作表") + relationship_id = sheets[0].attrib[f"{{{REL_NS}}}id"] + target = relationships[relationship_id].lstrip("/") + if not target.startswith("xl/"): + target = "xl/" + target + worksheet = ET.fromstring(archive.read(target)) + + rows: list[dict[int, str]] = [] + for row in worksheet.findall("m:sheetData/m:row", NS): + values: dict[int, str] = {} + for cell in row.findall("m:c", NS): + number = _column_number(cell.attrib.get("r", "A1")) + value = _cell_value(cell, shared) + if value != "": + values[number] = value + rows.append(values) + + header_row_index = -1 + for index, values in enumerate(rows[:20]): + texts = set(values.values()) + if "学号" in texts and "姓名" in texts: + header_row_index = index + break + if header_row_index < 0: + raise ValueError("Excel 中必须包含“学号”和“姓名”表头") + + headers = rows[header_row_index] + result: list[dict[str, str]] = [] + for values in rows[header_row_index + 1:]: + row = {headers.get(number, f"列{number}"): value for number, value in values.items()} + if any(str(value).strip() for value in row.values()): + result.append(row) + return result + + +def _xml_escape(value: object) -> str: + return ( + str(value) + .replace("&", "&") + .replace("<", "<") + .replace(">", ">") + .replace('"', """) + ) + + +def _sheet_xml(headers: list[str], records: list[dict[str, object]]) -> str: + all_headers = list(headers) + all_records = list(records) + row_xml: list[str] = [] + header_cells = [] + for column_number, header in enumerate(all_headers, 1): + reference = f"{_column_letter(column_number)}1" + header_cells.append( + f'{_xml_escape(header)}' + ) + row_xml.append(f'{"".join(header_cells)}') + + for row_number, record in enumerate(all_records, 2): + cells = [] + for column_number, header in enumerate(all_headers, 1): + value = record.get(header, "") + reference = f"{_column_letter(column_number)}{row_number}" + if isinstance(value, bool): + cells.append(f'{_xml_escape("是" if value else "否")}') + elif isinstance(value, (int, float)) and not isinstance(value, bool): + cells.append(f'{value}') + else: + text = "" if value is None else str(value) + preserve = ' xml:space="preserve"' if text != text.strip() else "" + cells.append( + f'{_xml_escape(text)}' + ) + row_xml.append(f'{"".join(cells)}') + + column_xml = [] + for number in range(1, len(all_headers) + 1): + column_xml.append(f'') + + return f''' + + +{"".join(column_xml)} +{"".join(row_xml)} +''' + + +def write_xlsx(sheets: list[tuple[str, list[str], list[dict[str, object]]]]) -> bytes: + """Create a valid XLSX in memory. Each item is (sheet name, headers, records).""" + import posixpath + + used_names = set() + safe_names = [] + for index, (name, _, _) in enumerate(sheets, 1): + clean = re.sub(r"[\\/?*\\[\\]:]", "-", name).strip() or f"Sheet{index}" + clean = clean[:31] + candidate = clean + suffix = 2 + while candidate.lower() in used_names: + tail = f"-{suffix}" + candidate = clean[: 31 - len(tail)] + tail + suffix += 1 + used_names.add(candidate.lower()) + safe_names.append(candidate) + + sheet_entries = [] + workbook_entries = [] + worksheet_files = [] + for index, name in enumerate(safe_names, 1): + filename = f"xl/worksheets/sheet{index}.xml" + worksheet_files.append((filename, _sheet_xml(sheets[index - 1][1], sheets[index - 1][2]))) + sheet_entries.append( + f'' + ) + workbook_entries.append( + f'' + ) + style_relationship_id = len(sheets) + 1 + + content_overrides = "".join( + f'' + for index in range(1, len(sheets) + 1) + ) + + files = { + "[Content_Types].xml": f''' + + + + +{content_overrides} +''', + "_rels/.rels": ''' + + +''', + "xl/workbook.xml": f''' + +{"".join(sheet_entries)} +''', + "xl/_rels/workbook.xml.rels": f''' +{"".join(workbook_entries)} + +''', + "xl/styles.xml": ''' + + + + + + +''', + } + files.update(worksheet_files) + + import io + output = io.BytesIO() + with ZipFile(output, "w", ZIP_DEFLATED) as archive: + for filename, content in files.items(): + archive.writestr(filename, content.encode("utf-8")) + return output.getvalue() + + +def export_filename(prefix: str, class_name: str | None = None) -> str: + timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") + suffix = f"-{class_name}班" if class_name else "" + return f"{prefix}{suffix}-{timestamp}.xlsx"