diff --git a/README.md b/README.md index 5167958..a1c2b14 100644 --- a/README.md +++ b/README.md @@ -74,6 +74,21 @@ TEACHER_USER=yourname TEACHER_PASSWORD=your-strong-password python3 app.py - `selfgroup.db`：学生、密码、分组、班级状态和登录会话。 - `2025mt.xlsx`：首次启动导入的原始名单。 +## 代码结构 + +服务端按职责拆分在 `group_system/` 包中： + +```text +app.py 启动入口，保留 python3 app.py 的用法 +group_system/config.py 路径、端口、账号和业务默认值 +group_system/security.py PBKDF2 密码散列与校验 +group_system/storage/ SQLite 连接、名单导入、账号、分组、管理、导出 +group_system/http/ API 适配、静态文件路由和 HTTP 服务器 +group_system/xlsx_io.py 无第三方依赖的 XLSX 读取/生成 +``` + +前端仍在 `static/` 目录中，接口协议保持不变。 + ## 建议部署方式 1. 在教师电脑上运行 `python3 app.py`。 diff --git a/app.py b/app.py index 00b2651..51a3da1 100644 --- a/app.py +++ b/app.py @@ -1,1271 +1,37 @@ #!/usr/bin/env python3 -"""学生自助分组系统 - B/S 架构，仅使用 Python 标准库。""" +"""学生自助分组系统入口。 + +具体实现按配置、存储、服务、HTTP 层拆分在 group_system 包中； +这里保留传统启动命令 `python3 app.py` 和旧的 STORE 全局变量。 +""" from __future__ import annotations -import hashlib -import http.server -import hmac -import json import logging -import os -import secrets -import sqlite3 -import socket -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", "teacher!@#") -GROUP_LIMIT = 6 -SESSION_DAYS = 7 +from group_system.config import ( + BASE_DIR, + DB_PATH, + GROUP_LIMIT, + SESSION_DAYS, + STATIC_DIR, + TEACHER_PASSWORD, + TEACHER_USER, + XLSX_PATH, +) +from group_system.http.server import run +from group_system.security import hash_password, verify_password +from group_system.storage.store import GroupStore logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") + +# 保留模块级实例，兼容项目里原有 `from app import STORE` 的引用方式。 +STORE = GroupStore(DB_PATH) 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),), - ) - self.connection.execute( - "INSERT OR IGNORE INTO settings(key,value) VALUES('group_limit',?)", - (str(GROUP_LIMIT),), - ) - - 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 group_limit(self) -> int: - value = self.get_setting("group_limit") - try: - limit = int(value or GROUP_LIMIT) - except ValueError: - limit = GROUP_LIMIT - return max(2, limit) - - def set_group_limit(self, limit: int) -> tuple[bool, str]: - if limit < 2 or limit > 100: - return False, "分组上限必须在 2 到 100 人之间" - self.set_setting("group_limit", str(limit)) - return True, f"分组上限已调整为 {limit} 人" - - 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_teacher_password( - self, session: dict[str, Any], old_password: str, new_password: str - ) -> tuple[bool, str]: - if session.get("role") != "teacher": - 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: - expected_user = self.get_setting("teacher_username") or TEACHER_USER - if session.get("username") != expected_user: - return False, "教师账号已变更，请重新登录" - stored_password = self.get_setting("teacher_password") - if not verify_password(old_password, stored_password): - return False, "原密码不正确" - self.set_setting("teacher_password", hash_password(new_password)) - return True, "教师密码修改成功" - - 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"]]) >= 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']} 移出小组" - - 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, "小组已解散" - - 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 = { - "ok": True, - "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 == "teacher/change-password" and method == "POST": - session, error = self.require_teacher() - if error: - return error - data = self.read_json() - ok, message = STORE.change_teacher_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 == "teacher/group-limit" and method == "POST": - _, error = self.require_teacher() - if error: - return error - data = self.read_json() - ok, message = STORE.set_group_limit(int(data.get("limit", 0))) - return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, { - "ok": ok, "message": message, "group_limit": STORE.group_limit() - }) - - 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 == "groups/remove-member" and method == "POST": - session, error = self.require_student() - if error: - return error - data = self.read_json() - ok, message = STORE.leader_remove_member( - session["student_id"], int(data.get("student_id", 0)) - ) - return json_response( - HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, - {"ok": ok, "message": 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) - - -def local_ipv4_addresses() -> list[str]: - addresses = set() - try: - host_name = socket.gethostname() - for item in socket.getaddrinfo(host_name, None, socket.AF_INET): - address = item[4][0] - if address != "127.0.0.1": - addresses.add(address) - except socket.gaierror: - pass - - # Connecting a UDP socket reveals the address used for the default route. - try: - probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - try: - probe.connect(("192.0.2.1", 9)) - address = probe.getsockname()[0] - if address != "127.0.0.1": - addresses.add(address) - finally: - probe.close() - except OSError: - pass - return sorted(addresses) - - -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) - addresses = local_ipv4_addresses() - if addresses: - for address in addresses: - LOG.info("学生入口：http://%s:%d/ 教师入口：http://%s:%d/teacher", address, port, address, port) - else: - LOG.warning("未能自动获取局域网 IP，请在教师机上查看网络设置；端口：%d", port) - try: - server.serve_forever() - except KeyboardInterrupt: - LOG.info("服务停止") - finally: - server.server_close() + run(STORE, XLSX_PATH) if __name__ == "__main__": diff --git a/group_system/__init__.py b/group_system/__init__.py new file mode 100644 index 0000000..3f10321 --- /dev/null +++ b/group_system/__init__.py @@ -0,0 +1 @@ +"""学生自助分组系统应用包。""" diff --git a/group_system/config.py b/group_system/config.py new file mode 100644 index 0000000..812f7f1 --- /dev/null +++ b/group_system/config.py @@ -0,0 +1,17 @@ +"""运行配置和环境变量。""" + +from __future__ import annotations + +import os +from pathlib import Path + +BASE_DIR = Path(__file__).resolve().parent.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", "teacher!@#") +GROUP_LIMIT = 6 +SESSION_DAYS = 7 +MAX_JSON_BYTES = 1024 * 1024 +MAX_UPLOAD_BYTES = 10 * 1024 * 1024 diff --git a/group_system/http/__init__.py b/group_system/http/__init__.py new file mode 100644 index 0000000..dd2076c --- /dev/null +++ b/group_system/http/__init__.py @@ -0,0 +1 @@ +"""HTTP 路由和响应工具。""" diff --git a/group_system/http/api.py b/group_system/http/api.py new file mode 100644 index 0000000..5ff8a6d --- /dev/null +++ b/group_system/http/api.py @@ -0,0 +1,358 @@ +"""学生和教师 API 适配层。""" + +from __future__ import annotations + +import json +import logging +import sqlite3 +import urllib.parse +from http import HTTPStatus +from typing import Any + +from group_system.config import BASE_DIR, MAX_JSON_BYTES, MAX_UPLOAD_BYTES +from group_system.http.responses import ( + clear_session_cookie, + json_response, + make_session_cookie, +) +from group_system.storage.store import GroupStore + +LOG = logging.getLogger("group-system") + + +class GroupRequestHandler: + """把 HTTP 请求转换成 GroupStore 调用，不直接访问数据库。""" + + server_version = "StudentGroupSystem/1.0" + + def __init__(self, handler, store: GroupStore): + self.handler = handler + self.store = store + + 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 = self.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 > MAX_JSON_BYTES: + 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 = self.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 = self.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": + self.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 = self.store.active_class() + common = { + "ok": True, + "authenticated": True, + "role": session["role"], + "active_class": active, + "classes": self.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": self.store.class_snapshot(selected), + }, + ) + return json_response( + HTTPStatus.OK, + { + **common, + "view_class": active, + "teacher": {"username": session["username"]}, + "snapshot": self.store.class_snapshot(active), + }, + ) + + student = self.store.get_student(session["student_id"]) + if student is None: + return json_response(HTTPStatus.OK, {"ok": True, "authenticated": False}) + membership = self.store.student_membership(student["id"]) + snapshot = self.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 = self.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 == "teacher/change-password" and method == "POST": + session, error = self.require_teacher() + if error: + return error + data = self.read_json() + ok, message = self.store.change_teacher_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 == "teacher/group-limit" and method == "POST": + _, error = self.require_teacher() + if error: + return error + data = self.read_json() + ok, message = self.store.set_group_limit(int(data.get("limit", 0))) + return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, { + "ok": ok, "message": message, "group_limit": self.store.group_limit() + }) + + if action == "groups" and method == "POST": + session, error = self.require_student() + if error: + return error + student = self.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, _ = self.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 = self.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 = self.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 == "groups/remove-member" and method == "POST": + session, error = self.require_student() + if error: + return error + data = self.read_json() + ok, message = self.store.leader_remove_member( + session["student_id"], int(data.get("student_id", 0)) + ) + return json_response( + HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, + {"ok": ok, "message": message}, + ) + + if action == "teacher/class" and method == "POST": + _, error = self.require_teacher() + if error: + return error + data = self.read_json() + ok, message = self.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", self.store.active_class())) + ok, message = self.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 = self.store.teacher_create_group( + str(data.get("class_name", self.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 = self.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 = self.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 = self.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 = self.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 = self.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 > MAX_UPLOAD_BYTES: + 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 = self.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 = self.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": "接口不存在"}) diff --git a/group_system/http/responses.py b/group_system/http/responses.py new file mode 100644 index 0000000..386f3ff --- /dev/null +++ b/group_system/http/responses.py @@ -0,0 +1,41 @@ +"""HTTP 响应和 Cookie 工具。""" + +from __future__ import annotations + +import json +from http import HTTPStatus +from http.server import BaseHTTPRequestHandler +from typing import Any + +from group_system.config import SESSION_DAYS + + +def json_response( + status: int | HTTPStatus, + payload: Any, + cookies: list[dict[str, Any]] | None = None, +): + """Return a callable that serializes and sends a JSON response.""" + def send(handler: BaseHTTPRequestHandler): + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + handler.send_response(int(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; " + f"Max-Age={SESSION_DAYS * 86400}" + ) + + +def clear_session_cookie() -> str: + return "GROUP_SESSION=; Path=/; HttpOnly; SameSite=Lax; Max-Age=0" diff --git a/group_system/http/server.py b/group_system/http/server.py new file mode 100644 index 0000000..700be72 --- /dev/null +++ b/group_system/http/server.py @@ -0,0 +1,138 @@ +"""HTTP 服务器、静态路由和应用启动流程。""" + +from __future__ import annotations + +import http.server +import logging +import os +import socket +import urllib.parse +from http import HTTPStatus +from pathlib import Path + +from group_system.config import STATIC_DIR, XLSX_PATH +from group_system.http.api import GroupRequestHandler +from group_system.http.static_files import send_text, serve_file +from group_system.storage.store import GroupStore + +LOG = logging.getLogger("group-system") + + +class GroupHTTPServer(http.server.ThreadingHTTPServer): + """Carry the application store on the standard threading HTTP server.""" + + def __init__(self, address: tuple[str, int], store: GroupStore): + super().__init__(address, RequestHandler) + self.store = store + + +class RequestHandler(http.server.BaseHTTPRequestHandler): + @property + def store(self) -> GroupStore: + return self.server.store # type: ignore[attr-defined] + + def do_GET(self) -> None: + route(self) + + def do_POST(self) -> None: + route(self) + + def do_HEAD(self) -> None: + route(self) + + def log_message(self, format_string: str, *args) -> None: + LOG.info("%s - %s", self.address_string(), format_string % args) + + +def route(handler: RequestHandler) -> None: + path = urllib.parse.urlparse(handler.path).path + if path.startswith("/api/"): + request = GroupRequestHandler(handler, handler.store) + result = request.handle_api(path[len("/api/"):], handler.command) + if callable(result): + result(handler) + return + + if path in ("/", "/index.html", "/student"): + serve_file(handler, STATIC_DIR / "index.html") + return + if path in ("/teacher", "/admin"): + serve_file(handler, STATIC_DIR / "teacher.html") + return + + static_root = STATIC_DIR.resolve() + safe_path = (static_root / urllib.parse.unquote(path).lstrip("/")).resolve() + try: + safe_path.relative_to(static_root) + except ValueError: + send_text(handler, HTTPStatus.NOT_FOUND, "Not Found") + return + serve_file(handler, safe_path) + + +def local_ipv4_addresses() -> list[str]: + addresses: set[str] = set() + try: + host_name = socket.gethostname() + for item in socket.getaddrinfo(host_name, None, socket.AF_INET): + address = item[4][0] + if address != "127.0.0.1": + addresses.add(address) + except socket.gaierror: + pass + + # A UDP connection does not send packets here, but reveals the default-route address. + try: + probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + probe.connect(("192.0.2.1", 9)) + address = probe.getsockname()[0] + if address != "127.0.0.1": + addresses.add(address) + finally: + probe.close() + except OSError: + pass + return sorted(addresses) + + +def bootstrap_store(store: GroupStore, xlsx_path: Path) -> None: + """Import the initial roster only when the database is still empty.""" + if any(store.class_list()): + return + if not xlsx_path.exists(): + LOG.warning("未找到 %s，请先通过教师后台上传学生名单", xlsx_path) + return + student_count, class_count = store.import_roster(xlsx_path) + LOG.info("已导入 %s： %d 名学生 / %d 个班", xlsx_path, student_count, class_count) + + +def run( + store: GroupStore, + xlsx_path: Path = XLSX_PATH, + host: str | None = None, + port: int | None = None, +) -> None: + bootstrap_store(store, xlsx_path) + host = host or os.environ.get("HOST", "0.0.0.0") + port = port if port is not None else int(os.environ.get("PORT", "8000")) + server = GroupHTTPServer((host, port), store) + LOG.info("学生分组系统已启动：http://0.0.0.0:%d", port) + addresses = local_ipv4_addresses() + if addresses: + for address in addresses: + LOG.info( + "学生入口：http://%s:%d/ 教师入口：http://%s:%d/teacher", + address, + port, + address, + port, + ) + else: + LOG.warning("未能自动获取局域网 IP，请在教师机上查看网络设置；端口：%d", port) + try: + server.serve_forever() + except KeyboardInterrupt: + LOG.info("服务停止") + finally: + server.server_close() diff --git a/group_system/http/static_files.py b/group_system/http/static_files.py new file mode 100644 index 0000000..8be97eb --- /dev/null +++ b/group_system/http/static_files.py @@ -0,0 +1,43 @@ +"""静态页面与资源文件服务。""" + +from __future__ import annotations + +from http import HTTPStatus +from http.server import BaseHTTPRequestHandler +from pathlib import Path + +from group_system.config import STATIC_DIR + +CONTENT_TYPES = { + ".html": "text/html; charset=utf-8", + ".htm": "text/html; charset=utf-8", + ".css": "text/css; charset=utf-8", + ".js": "text/javascript; charset=utf-8", + ".mjs": "text/javascript; charset=utf-8", + ".svg": "image/svg+xml", + ".png": "image/png", +} + + +def send_text(handler: BaseHTTPRequestHandler, status: int, text: str) -> None: + 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) + + +def serve_file(handler: BaseHTTPRequestHandler, path: Path) -> None: + path = path if path.is_absolute() else STATIC_DIR / path + if not path.is_file(): + send_text(handler, HTTPStatus.NOT_FOUND, "Not Found") + return + content_type = CONTENT_TYPES.get(path.suffix.lower(), "application/octet-stream") + content = path.read_bytes() + handler.send_response(HTTPStatus.OK) + handler.send_header("Content-Type", content_type) + handler.send_header("Content-Length", str(len(content))) + handler.send_header("Cache-Control", "no-cache") + handler.end_headers() + handler.wfile.write(content) diff --git a/group_system/security.py b/group_system/security.py new file mode 100644 index 0000000..0a02626 --- /dev/null +++ b/group_system/security.py @@ -0,0 +1,26 @@ +"""密码散列与校验。""" + +from __future__ import annotations + +import hashlib +import hmac +import secrets + + +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) diff --git a/group_system/storage/__init__.py b/group_system/storage/__init__.py new file mode 100644 index 0000000..c140893 --- /dev/null +++ b/group_system/storage/__init__.py @@ -0,0 +1 @@ +"""SQLite 存储层。""" diff --git a/group_system/storage/accounts.py b/group_system/storage/accounts.py new file mode 100644 index 0000000..f250c91 --- /dev/null +++ b/group_system/storage/accounts.py @@ -0,0 +1,152 @@ +"""账号登录、会话和密码管理。""" + +from __future__ import annotations + +import hashlib +import secrets +import time +from typing import Any + +from group_system.config import SESSION_DAYS, TEACHER_USER +from group_system.security import hash_password, verify_password + + +class AccountStoreMixin: + + 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_teacher_password( + self, session: dict[str, Any], old_password: str, new_password: str + ) -> tuple[bool, str]: + if session.get("role") != "teacher": + 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: + expected_user = self.get_setting("teacher_username") or TEACHER_USER + if session.get("username") != expected_user: + return False, "教师账号已变更，请重新登录" + stored_password = self.get_setting("teacher_password") + if not verify_password(old_password, stored_password): + return False, "原密码不正确" + self.set_setting("teacher_password", hash_password(new_password)) + return True, "教师密码修改成功" + + 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 diff --git a/group_system/storage/admin.py b/group_system/storage/admin.py new file mode 100644 index 0000000..f993d69 --- /dev/null +++ b/group_system/storage/admin.py @@ -0,0 +1,136 @@ +"""教师端班级和分组管理。""" + +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, "小组已解散" diff --git a/group_system/storage/database.py b/group_system/storage/database.py new file mode 100644 index 0000000..59fe544 --- /dev/null +++ b/group_system/storage/database.py @@ -0,0 +1,142 @@ +"""数据库连接、Schema 和系统级设置。""" + +from __future__ import annotations + +import sqlite3 +import threading +from pathlib import Path +from typing import Any + +from group_system.config import GROUP_LIMIT, TEACHER_PASSWORD, TEACHER_USER +from group_system.security import hash_password + + +class DatabaseCore: + + 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),), + ) + self.connection.execute( + "INSERT OR IGNORE INTO settings(key,value) VALUES('group_limit',?)", + (str(GROUP_LIMIT),), + ) + + 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 group_limit(self) -> int: + value = self.get_setting("group_limit") + try: + limit = int(value or GROUP_LIMIT) + except ValueError: + limit = GROUP_LIMIT + return max(2, limit) + + def set_group_limit(self, limit: int) -> tuple[bool, str]: + if limit < 2 or limit > 100: + return False, "分组上限必须在 2 到 100 人之间" + self.set_setting("group_limit", str(limit)) + return True, f"分组上限已调整为 {limit} 人" + + 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"]) diff --git a/group_system/storage/export.py b/group_system/storage/export.py new file mode 100644 index 0000000..3c0f451 --- /dev/null +++ b/group_system/storage/export.py @@ -0,0 +1,92 @@ +"""分组数据导出。""" + +from __future__ import annotations + +from group_system.xlsx_io import export_filename, write_xlsx + + +class ExportStoreMixin: + + 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) diff --git a/group_system/storage/groups.py b/group_system/storage/groups.py new file mode 100644 index 0000000..83b9ad7 --- /dev/null +++ b/group_system/storage/groups.py @@ -0,0 +1,195 @@ +"""学生端分组用例和班级快照查询。""" + +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']} 移出小组" diff --git a/group_system/storage/roster.py b/group_system/storage/roster.py new file mode 100644 index 0000000..2cb7446 --- /dev/null +++ b/group_system/storage/roster.py @@ -0,0 +1,69 @@ +"""学生名单导入。""" + +from __future__ import annotations + +from pathlib import Path + +from group_system.security import hash_password +from group_system.xlsx_io import read_xlsx_rows + + +class RosterStoreMixin: + + 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}) diff --git a/group_system/storage/store.py b/group_system/storage/store.py new file mode 100644 index 0000000..f1f3acf --- /dev/null +++ b/group_system/storage/store.py @@ -0,0 +1,23 @@ +"""GroupStore 组合各个存储模块，作为兼容旧代码的门面。""" + +from __future__ import annotations + +from pathlib import Path + +from group_system.storage.accounts import AccountStoreMixin +from group_system.storage.admin import TeacherGroupAdminMixin +from group_system.storage.database import DatabaseCore +from group_system.storage.export import ExportStoreMixin +from group_system.storage.groups import GroupOperationsMixin +from group_system.storage.roster import RosterStoreMixin + + +class GroupStore( + DatabaseCore, + RosterStoreMixin, + AccountStoreMixin, + GroupOperationsMixin, + TeacherGroupAdminMixin, + ExportStoreMixin, +): + """封装所有 SQLite 持久化操作。HTTP 层只依赖这个门面。""" diff --git a/group_system/xlsx_io.py b/group_system/xlsx_io.py new file mode 100644 index 0000000..44dbd90 --- /dev/null +++ b/group_system/xlsx_io.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" diff --git a/xlsxio.py b/xlsxio.py index 44dbd90..507150d 100644 --- a/xlsxio.py +++ b/xlsxio.py @@ -1,242 +1,5 @@ -#!/usr/bin/env python3 -"""Minimal XLSX reader/writer using only the Python standard library.""" +"""Compatibility import facade for the original xlsxio module.""" -from __future__ import annotations +from group_system.xlsx_io import export_filename, read_xlsx_rows, write_xlsx -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" +__all__ = ["export_filename", "read_xlsx_rows", "write_xlsx"]