1179 lines
52 KiB
Python
1179 lines
52 KiB
Python
#!/usr/bin/env python3
|
||
"""学生自助分组系统 - B/S 架构，仅使用 Python 标准库。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import http.server
|
||
import hmac
|
||
import json
|
||
import logging
|
||
import os
|
||
import secrets
|
||
import sqlite3
|
||
import threading
|
||
import time
|
||
import urllib.parse
|
||
from http import HTTPStatus
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from xlsxio import export_filename, read_xlsx_rows, write_xlsx
|
||
|
||
|
||
BASE_DIR = Path(__file__).resolve().parent
|
||
DB_PATH = Path(os.environ.get("GROUP_DB", BASE_DIR / "selfgroup.db"))
|
||
XLSX_PATH = Path(os.environ.get("GROUP_XLSX", BASE_DIR / "2025mt.xlsx"))
|
||
STATIC_DIR = BASE_DIR / "static"
|
||
TEACHER_USER = os.environ.get("TEACHER_USER", "teacher")
|
||
TEACHER_PASSWORD = os.environ.get("TEACHER_PASSWORD", "teacher123")
|
||
GROUP_LIMIT = 6
|
||
SESSION_DAYS = 7
|
||
|
||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
||
LOG = logging.getLogger("group-system")
|
||
|
||
|
||
def hash_password(password: str, salt: bytes | None = None) -> str:
|
||
salt = salt or secrets.token_bytes(16)
|
||
digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, 180_000)
|
||
return salt.hex() + "$" + digest.hex()
|
||
|
||
|
||
def verify_password(password: str, stored: str | None) -> bool:
|
||
if not stored or "$" not in stored:
|
||
return False
|
||
try:
|
||
salt_hex, digest_hex = stored.split("$", 1)
|
||
salt = bytes.fromhex(salt_hex)
|
||
expected = bytes.fromhex(digest_hex)
|
||
except ValueError:
|
||
return False
|
||
actual = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, 180_000)
|
||
return hmac.compare_digest(actual, expected)
|
||
|
||
|
||
class GroupStore:
|
||
def __init__(self, path: Path):
|
||
self.path = path
|
||
self.lock = threading.RLock()
|
||
self.connection = sqlite3.connect(str(path), check_same_thread=False)
|
||
self.connection.row_factory = sqlite3.Row
|
||
self.connection.execute("PRAGMA foreign_keys = ON")
|
||
self.init_database()
|
||
|
||
def init_database(self) -> None:
|
||
with self.lock, self.connection:
|
||
self.connection.executescript(
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS students (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
class_name TEXT NOT NULL,
|
||
student_no TEXT NOT NULL UNIQUE,
|
||
name TEXT NOT NULL,
|
||
gender TEXT DEFAULT '',
|
||
admission_no TEXT DEFAULT '',
|
||
remark TEXT DEFAULT '',
|
||
password_hash TEXT NOT NULL,
|
||
must_change_password INTEGER NOT NULL DEFAULT 1
|
||
);
|
||
CREATE TABLE IF NOT EXISTS groups (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
class_name TEXT NOT NULL,
|
||
name TEXT NOT NULL,
|
||
leader_student_id INTEGER REFERENCES students(id) ON DELETE SET NULL,
|
||
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||
);
|
||
CREATE TABLE IF NOT EXISTS group_members (
|
||
group_id INTEGER NOT NULL REFERENCES groups(id) ON DELETE CASCADE,
|
||
student_id INTEGER NOT NULL UNIQUE REFERENCES students(id) ON DELETE CASCADE,
|
||
joined_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||
);
|
||
CREATE TABLE IF NOT EXISTS class_controls (
|
||
class_name TEXT PRIMARY KEY,
|
||
locked INTEGER NOT NULL DEFAULT 0
|
||
);
|
||
CREATE TABLE IF NOT EXISTS settings (
|
||
key TEXT PRIMARY KEY,
|
||
value TEXT NOT NULL
|
||
);
|
||
CREATE TABLE IF NOT EXISTS sessions (
|
||
token_hash TEXT PRIMARY KEY,
|
||
role TEXT NOT NULL,
|
||
username TEXT NOT NULL,
|
||
student_id INTEGER REFERENCES students(id) ON DELETE CASCADE,
|
||
created_at INTEGER NOT NULL,
|
||
expires_at INTEGER NOT NULL
|
||
);
|
||
"""
|
||
)
|
||
self.connection.execute(
|
||
"INSERT OR IGNORE INTO settings(key,value) VALUES('active_class','1')"
|
||
)
|
||
self.connection.execute(
|
||
"INSERT OR IGNORE INTO settings(key,value) VALUES('teacher_username',?)",
|
||
(TEACHER_USER,),
|
||
)
|
||
self.connection.execute(
|
||
"INSERT OR IGNORE INTO settings(key,value) VALUES('teacher_password',?)",
|
||
(hash_password(TEACHER_PASSWORD),),
|
||
)
|
||
|
||
def import_roster(self, path: Path) -> tuple[int, int]:
|
||
rows = read_xlsx_rows(path)
|
||
students = []
|
||
errors = []
|
||
seen = set()
|
||
for index, row in enumerate(rows, 2):
|
||
student_no = str(row.get("学号", "")).strip()
|
||
name = str(row.get("姓名", "")).strip()
|
||
class_name = str(
|
||
row.get("班级", row.get("班别", row.get("行政班", "")))
|
||
).strip()
|
||
gender = str(row.get("性别", "")).strip()
|
||
admission_no = str(row.get("录取编号", "")).strip()
|
||
remark = str(row.get("备注", "")).strip()
|
||
if not student_no or not name or not class_name:
|
||
errors.append(f"第 {index} 行：学号、姓名、班级均不能为空")
|
||
continue
|
||
if student_no in seen:
|
||
errors.append(f"第 {index} 行：学号 {student_no} 重复")
|
||
continue
|
||
seen.add(student_no)
|
||
students.append(
|
||
(class_name, student_no, name, gender, admission_no, remark)
|
||
)
|
||
if errors:
|
||
raise ValueError("；".join(errors[:8]))
|
||
if not students:
|
||
raise ValueError("Excel 中没有可导入的学生")
|
||
|
||
with self.lock, self.connection:
|
||
self.connection.executescript(
|
||
"DELETE FROM sessions; DELETE FROM group_members; DELETE FROM groups; DELETE FROM students;"
|
||
)
|
||
self.connection.executemany(
|
||
"""INSERT INTO students
|
||
(class_name,student_no,name,gender,admission_no,remark,password_hash,must_change_password)
|
||
VALUES(?,?,?,?,?,?,?,1)""",
|
||
[
|
||
(
|
||
item[0],
|
||
item[1],
|
||
item[2],
|
||
item[3],
|
||
item[4],
|
||
item[5],
|
||
hash_password(item[1]),
|
||
)
|
||
for item in students
|
||
],
|
||
)
|
||
active = self.get_setting("active_class", conn=self.connection) or students[0][0]
|
||
if not any(item[0] == active for item in students):
|
||
active = students[0][0]
|
||
self.connection.execute(
|
||
"UPDATE settings SET value=? WHERE key='active_class'", (active,)
|
||
)
|
||
return len(students), len({item[0] for item in students})
|
||
|
||
def get_setting(self, key: str, conn: sqlite3.Connection | None = None) -> str | None:
|
||
conn = conn or self.connection
|
||
row = conn.execute("SELECT value FROM settings WHERE key=?", (key,)).fetchone()
|
||
return None if row is None else row["value"]
|
||
|
||
def set_setting(self, key: str, value: str) -> None:
|
||
with self.lock, self.connection:
|
||
self.connection.execute(
|
||
"INSERT INTO settings(key,value) VALUES(?,?) "
|
||
"ON CONFLICT(key) DO UPDATE SET value=excluded.value",
|
||
(key, value),
|
||
)
|
||
|
||
def class_list(self) -> list[dict[str, Any]]:
|
||
with self.lock:
|
||
rows = self.connection.execute(
|
||
"""SELECT class_name, COUNT(*) AS student_count
|
||
FROM students GROUP BY class_name
|
||
ORDER BY CAST(class_name AS REAL), class_name"""
|
||
).fetchall()
|
||
active = self.get_setting("active_class")
|
||
locked = {
|
||
row["class_name"]: bool(row["locked"])
|
||
for row in self.connection.execute("SELECT class_name,locked FROM class_controls")
|
||
}
|
||
return [
|
||
{
|
||
"class_name": row["class_name"],
|
||
"student_count": row["student_count"],
|
||
"locked": bool(locked.get(row["class_name"], False)),
|
||
"active": row["class_name"] == active,
|
||
}
|
||
for row in rows
|
||
]
|
||
|
||
def active_class(self) -> str:
|
||
return self.get_setting("active_class") or ""
|
||
|
||
def is_locked(self, class_name: str) -> bool:
|
||
with self.lock:
|
||
row = self.connection.execute(
|
||
"SELECT locked FROM class_controls WHERE class_name=?", (class_name,)
|
||
).fetchone()
|
||
return bool(row and row["locked"])
|
||
|
||
def create_session(self, role: str, username: str, student_id: int | None = None) -> str:
|
||
token = secrets.token_urlsafe(32)
|
||
token_hash = hashlib.sha256(token.encode()).hexdigest()
|
||
now = int(time.time())
|
||
with self.lock, self.connection:
|
||
self.connection.execute(
|
||
"INSERT INTO sessions(token_hash,role,username,student_id,created_at,expires_at) VALUES(?,?,?,?,?,?)",
|
||
(token_hash, role, username, student_id, now, now + SESSION_DAYS * 86400),
|
||
)
|
||
self.connection.execute(
|
||
"DELETE FROM sessions WHERE expires_at < ?", (now,)
|
||
)
|
||
return token
|
||
|
||
def get_session(self, token: str | None) -> dict[str, Any] | None:
|
||
if not token:
|
||
return None
|
||
token_hash = hashlib.sha256(token.encode()).hexdigest()
|
||
now = int(time.time())
|
||
with self.lock:
|
||
row = self.connection.execute(
|
||
"""SELECT s.token_hash,s.role,s.username,s.student_id,s.expires_at,
|
||
st.class_name,st.student_no,st.name,st.must_change_password
|
||
FROM sessions s LEFT JOIN students st ON st.id=s.student_id
|
||
WHERE s.token_hash=? AND s.expires_at>?""",
|
||
(token_hash, now),
|
||
).fetchone()
|
||
if row is None:
|
||
return None
|
||
return dict(row)
|
||
|
||
def destroy_session(self, token: str | None) -> None:
|
||
if not token:
|
||
return
|
||
token_hash = hashlib.sha256(token.encode()).hexdigest()
|
||
with self.lock, self.connection:
|
||
self.connection.execute("DELETE FROM sessions WHERE token_hash=?", (token_hash,))
|
||
|
||
def login_student(self, student_no: str, password: str) -> tuple[dict[str, Any] | None, str]:
|
||
student_no = student_no.strip()
|
||
active = self.active_class()
|
||
with self.lock:
|
||
student = self.connection.execute(
|
||
"SELECT * FROM students WHERE student_no=?", (student_no,)
|
||
).fetchone()
|
||
if student is None or not verify_password(password, student["password_hash"]):
|
||
return None, "学号或密码不正确"
|
||
if student["class_name"] != active:
|
||
return None, f"当前仅开放 {active} 班登录，{student['class_name']} 班暂不能登录"
|
||
token = self.create_session("student", student["student_no"], student["id"])
|
||
return {
|
||
"token": token,
|
||
"student": {
|
||
"id": student["id"],
|
||
"name": student["name"],
|
||
"student_no": student["student_no"],
|
||
"class_name": student["class_name"],
|
||
"must_change_password": bool(student["must_change_password"]),
|
||
},
|
||
}, "登录成功"
|
||
|
||
def login_teacher(self, username: str, password: str) -> tuple[dict[str, Any] | None, str]:
|
||
with self.lock:
|
||
expected_user = self.get_setting("teacher_username") or TEACHER_USER
|
||
expected_password = self.get_setting("teacher_password")
|
||
if username.strip() != expected_user or not verify_password(password, expected_password):
|
||
return None, "教师账号或密码不正确"
|
||
token = self.create_session("teacher", expected_user, None)
|
||
return {"token": token, "teacher": {"username": expected_user}}, "登录成功"
|
||
|
||
def change_student_password(
|
||
self, session: dict[str, Any], old_password: str, new_password: str
|
||
) -> tuple[bool, str]:
|
||
student_id = session.get("student_id")
|
||
if not student_id:
|
||
return False, "学生会话无效"
|
||
old_password = old_password or ""
|
||
new_password = new_password or ""
|
||
if len(new_password) < 8:
|
||
return False, "新密码至少 8 位"
|
||
if new_password == old_password:
|
||
return False, "新密码不能与原密码相同"
|
||
with self.lock:
|
||
student = self.connection.execute(
|
||
"SELECT * FROM students WHERE id=?", (student_id,)
|
||
).fetchone()
|
||
if student is None or not verify_password(old_password, student["password_hash"]):
|
||
return False, "原密码不正确"
|
||
if new_password == student["student_no"]:
|
||
return False, "新密码不能设置为学号"
|
||
with self.connection:
|
||
self.connection.execute(
|
||
"UPDATE students SET password_hash=?, must_change_password=0 WHERE id=?",
|
||
(hash_password(new_password), student_id),
|
||
)
|
||
token_hash = hashlib.sha256(session["_token"].encode()).hexdigest()
|
||
self.connection.execute(
|
||
"DELETE FROM sessions WHERE student_id=? AND token_hash<>?",
|
||
(student_id, token_hash),
|
||
)
|
||
return True, "密码修改成功"
|
||
|
||
def get_student(self, student_id: int) -> dict[str, Any] | None:
|
||
with self.lock:
|
||
row = self.connection.execute("SELECT * FROM students WHERE id=?", (student_id,)).fetchone()
|
||
return dict(row) if row else None
|
||
|
||
def student_membership(self, student_id: int) -> dict[str, Any] | None:
|
||
with self.lock:
|
||
row = self.connection.execute(
|
||
"""SELECT m.group_id,g.name,g.class_name,g.leader_student_id
|
||
FROM group_members m JOIN groups g ON g.id=m.group_id
|
||
WHERE m.student_id=?""",
|
||
(student_id,),
|
||
).fetchone()
|
||
return dict(row) if row else None
|
||
|
||
def class_snapshot(self, class_name: str) -> dict[str, Any]:
|
||
with self.lock:
|
||
groups = self.connection.execute(
|
||
"""SELECT g.id,g.name,g.leader_student_id,g.created_at,
|
||
s.student_no AS leader_student_no,s.name AS leader_name
|
||
FROM groups g LEFT JOIN students s ON s.id=g.leader_student_id
|
||
WHERE g.class_name=? ORDER BY g.id""",
|
||
(class_name,),
|
||
).fetchall()
|
||
group_ids = [row["id"] for row in groups]
|
||
members: dict[int, list[dict[str, Any]]] = {row["id"]: [] for row in groups}
|
||
if group_ids:
|
||
placeholders = ",".join("?" for _ in group_ids)
|
||
member_rows = self.connection.execute(
|
||
f"""SELECT m.group_id,st.id,st.student_no,st.name,st.gender,g.leader_student_id
|
||
FROM group_members m JOIN students st ON st.id=m.student_id
|
||
JOIN groups g ON g.id=m.group_id
|
||
WHERE m.group_id IN ({placeholders})
|
||
ORDER BY CASE WHEN st.id=g.leader_student_id THEN 0 ELSE 1 END,m.joined_at,st.student_no""",
|
||
group_ids,
|
||
).fetchall()
|
||
pending = [dict(row) for row in member_rows]
|
||
for member in pending:
|
||
member["is_leader"] = member["id"] == member["leader_student_id"]
|
||
member.pop("leader_student_id", None)
|
||
members[member["group_id"]].append(member)
|
||
|
||
ungrouped_rows = self.connection.execute(
|
||
"""SELECT id,student_no,name,gender FROM students
|
||
WHERE class_name=? AND id NOT IN (SELECT student_id FROM group_members)
|
||
ORDER BY student_no""",
|
||
(class_name,),
|
||
).fetchall()
|
||
total = self.connection.execute(
|
||
"SELECT COUNT(*) AS n FROM students WHERE class_name=?", (class_name,)
|
||
).fetchone()["n"]
|
||
grouped = self.connection.execute(
|
||
"""SELECT COUNT(*) AS n FROM group_members m
|
||
JOIN students s ON s.id=m.student_id WHERE s.class_name=?""",
|
||
(class_name,),
|
||
).fetchone()["n"]
|
||
|
||
group_data = []
|
||
for row in groups:
|
||
group_data.append(
|
||
{
|
||
**dict(row),
|
||
"locked": self.is_locked(class_name),
|
||
"member_count": len(members[row["id"]]),
|
||
"full": len(members[row["id"]]) >= GROUP_LIMIT,
|
||
"members": members[row["id"]],
|
||
}
|
||
)
|
||
return {
|
||
"class_name": class_name,
|
||
"locked": self.is_locked(class_name),
|
||
"group_limit": GROUP_LIMIT,
|
||
"groups": group_data,
|
||
"ungrouped": [dict(row) for row in ungrouped_rows],
|
||
"statistics": {"total_students": total, "grouped_students": grouped, "group_count": len(group_data)},
|
||
}
|
||
|
||
def create_group(self, student_id: int, group_name: str) -> tuple[bool, str, int | None]:
|
||
student = self.get_student(student_id)
|
||
if student is None:
|
||
return False, "学生不存在", None
|
||
if self.is_locked(student["class_name"]):
|
||
return False, "老师已停止分组操作，当前只能查看", None
|
||
current = self.student_membership(student_id)
|
||
if current is not None:
|
||
return False, "你已加入其他组，不能创建组；请先退出当前组", None
|
||
if student["must_change_password"]:
|
||
return False, "首次登录必须先修改密码", None
|
||
group_name = group_name.strip() or 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"]
|
||
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"]
|
||
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 == "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)
|
||
|
||
|
||
class Handler(http.server.BaseHTTPRequestHandler):
|
||
def do_GET(self):
|
||
route(self)
|
||
|
||
def do_POST(self):
|
||
route(self)
|
||
|
||
def do_HEAD(self):
|
||
route(self)
|
||
|
||
def log_message(self, format_string, *args):
|
||
LOG.info("%s - %s", self.address_string(), format_string % args)
|
||
|
||
|
||
def main() -> None:
|
||
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||
if not any(STORE.class_list()):
|
||
if XLSX_PATH.exists():
|
||
student_count, class_count = STORE.import_roster(XLSX_PATH)
|
||
LOG.info("已导入 %s： %d 名学生 / %d 个班", XLSX_PATH, student_count, class_count)
|
||
else:
|
||
LOG.warning("未找到 %s，请先通过教师后台上传学生名单", XLSX_PATH)
|
||
|
||
host = os.environ.get("HOST", "0.0.0.0")
|
||
port = int(os.environ.get("PORT", "8000"))
|
||
server = http.server.ThreadingHTTPServer((host, port), Handler)
|
||
LOG.info("学生分组系统已启动：http://0.0.0.0:%d", port)
|
||
LOG.info("学生入口：http://<老师电脑IP>:%d/ 教师入口：http://<老师电脑IP>:%d/teacher", port, port)
|
||
try:
|
||
server.serve_forever()
|
||
except KeyboardInterrupt:
|
||
LOG.info("服务停止")
|
||
finally:
|
||
server.server_close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|