add shortcuts traner
This commit is contained in:
795
shortcut_trainer/app.py
Normal file
795
shortcut_trainer/app.py
Normal file
@@ -0,0 +1,795 @@
|
||||
#!/usr/bin/env python3
|
||||
"""快捷键公开课训练系统。
|
||||
|
||||
零第三方依赖。学生账号直接使用原分组系统的 selfgroup.db；
|
||||
训练相关数据表也创建在同一个数据库中，并使用 trainer_ 前缀。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import errno
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import secrets
|
||||
import shutil
|
||||
import socket
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
from http import HTTPStatus
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from urllib.parse import unquote, urlsplit
|
||||
|
||||
BASE_DIR = Path(__file__).resolve().parent
|
||||
STATIC_DIR = BASE_DIR / "static"
|
||||
DB_PATH = Path(os.environ.get("SHARED_DB", BASE_DIR.parent / "selfgroup.db"))
|
||||
PORT = int(os.environ.get("PORT", "8010"))
|
||||
HOST = os.environ.get("HOST", "0.0.0.0")
|
||||
TEACHER_USER = os.environ.get("TEACHER_USER", "teacher")
|
||||
TEACHER_PASSWORD = os.environ.get("TEACHER_PASSWORD", "teacher123")
|
||||
SESSION_DAYS = 2
|
||||
PAGES = [
|
||||
{"page": 1, "key": "P", "title": "红色火星站", "color": "#ff3b52"},
|
||||
{"page": 2, "key": "Q", "title": "蓝色深海舱", "color": "#357bff"},
|
||||
{"page": 3, "key": "R", "title": "绿色能量站", "color": "#00d164"},
|
||||
{"page": 4, "key": "T", "title": "黄色指挥所", "color": "#ffdb00"},
|
||||
{"page": 5, "key": "Y", "title": "紫色观测台", "color": "#c159ff"},
|
||||
{"page": 6, "key": "U", "title": "橙色发射场", "color": "#ff7a18"},
|
||||
]
|
||||
BUTTON_NAMES = ["火箭准备", "燃料检查", "通讯开启", "系统诊断", "备用电源", "外部链接"]
|
||||
TEST_PASSWORD = "test1234"
|
||||
TEST_STUDENTS = [
|
||||
("测试", "test01", "测试甲"),
|
||||
("测试", "test02", "测试乙"),
|
||||
("测试", "test03", "测试丙"),
|
||||
("测试", "test04", "测试丁"),
|
||||
("测试", "test05", "测试戊"),
|
||||
]
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
||||
LOG = logging.getLogger("shortcut-trainer")
|
||||
|
||||
|
||||
def hash_password(password: str, salt: bytes | None = None) -> str:
|
||||
salt = salt or secrets.token_bytes(16)
|
||||
digest = hashlib.pbkdf2_hmac("sha256", password.encode(), 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, expected = bytes.fromhex(salt_hex), bytes.fromhex(digest_hex)
|
||||
# 原分组系统使用 180000 次；早期训练库使用过 120000 次。
|
||||
return any(
|
||||
hmac.compare_digest(
|
||||
hashlib.pbkdf2_hmac("sha256", password.encode(), salt, iterations),
|
||||
expected,
|
||||
)
|
||||
for iterations in (180_000, 120_000)
|
||||
)
|
||||
except (ValueError, TypeError):
|
||||
return False
|
||||
|
||||
|
||||
class Store:
|
||||
def __init__(self, path: Path):
|
||||
self.path = path
|
||||
self.lock = threading.RLock()
|
||||
self.conn = sqlite3.connect(str(path), check_same_thread=False)
|
||||
self.conn.row_factory = sqlite3.Row
|
||||
self.conn.execute("PRAGMA foreign_keys = ON")
|
||||
self.conn.execute("PRAGMA busy_timeout = 5000")
|
||||
self.init_db()
|
||||
|
||||
def init_db(self) -> None:
|
||||
with self.lock, self.conn:
|
||||
self.conn.executescript(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS trainer_presence (
|
||||
student_id INTEGER PRIMARY KEY REFERENCES students(id) ON DELETE CASCADE,
|
||||
last_seen INTEGER NOT NULL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS trainer_rounds (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
label TEXT NOT NULL,
|
||||
expected_json TEXT NOT NULL,
|
||||
active INTEGER NOT NULL DEFAULT 1,
|
||||
created_at INTEGER NOT NULL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS trainer_scores (
|
||||
round_id INTEGER NOT NULL REFERENCES trainer_rounds(id) ON DELETE CASCADE,
|
||||
student_id INTEGER NOT NULL REFERENCES students(id) ON DELETE CASCADE,
|
||||
correct INTEGER NOT NULL,
|
||||
finished_at INTEGER NOT NULL,
|
||||
payload TEXT NOT NULL,
|
||||
UNIQUE(round_id, student_id)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS trainer_events (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
round_id INTEGER REFERENCES trainer_rounds(id) ON DELETE SET NULL,
|
||||
student_id INTEGER NOT NULL REFERENCES students(id) ON DELETE CASCADE,
|
||||
kind TEXT NOT NULL,
|
||||
page_no INTEGER,
|
||||
item_index INTEGER,
|
||||
value TEXT,
|
||||
created_at INTEGER NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_trainer_events_round
|
||||
ON trainer_events(round_id, created_at);
|
||||
"""
|
||||
)
|
||||
self.conn.execute(
|
||||
"INSERT OR IGNORE INTO settings(key,value) VALUES('teacher_username',?)", (TEACHER_USER,)
|
||||
)
|
||||
self.conn.execute(
|
||||
"INSERT OR IGNORE INTO settings(key,value) VALUES('teacher_password',?)",
|
||||
(hash_password(TEACHER_PASSWORD),),
|
||||
)
|
||||
self.ensure_test_students()
|
||||
|
||||
def setting(self, key: str) -> str | None:
|
||||
with self.lock:
|
||||
row = self.conn.execute("SELECT value FROM settings WHERE key=?", (key,)).fetchone()
|
||||
return row["value"] if row else None
|
||||
|
||||
def login_student(self, username: str, password: str) -> tuple[dict | None, str]:
|
||||
username = username.strip()
|
||||
with self.lock:
|
||||
row = self.conn.execute(
|
||||
"SELECT * FROM students WHERE student_no=?", (username,)
|
||||
).fetchone()
|
||||
if not row or not verify_password(password, row["password_hash"]):
|
||||
return None, "学号或密码不正确"
|
||||
token = self.create_session("student", username, row["id"])
|
||||
return {"token": token, "student": dict(row)}, "登录成功"
|
||||
|
||||
def login_teacher(self, username: str, password: str) -> tuple[dict | None, str]:
|
||||
if username.strip() != (self.setting("teacher_username") or TEACHER_USER):
|
||||
return None, "教师账号或密码不正确"
|
||||
if not verify_password(password, self.setting("teacher_password")):
|
||||
return None, "教师账号或密码不正确"
|
||||
token = self.create_session("teacher", username)
|
||||
return {"token": token}, "登录成功"
|
||||
|
||||
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()
|
||||
expires = int(time.time() + SESSION_DAYS * 86400)
|
||||
with self.lock, self.conn:
|
||||
self.conn.execute(
|
||||
"""INSERT INTO sessions(token_hash,role,username,student_id,created_at,expires_at)
|
||||
VALUES(?,?,?,?,?,?)""",
|
||||
(token_hash, role, username, student_id, int(time.time()), expires),
|
||||
)
|
||||
return token
|
||||
|
||||
def session(self, token: str | None) -> dict | None:
|
||||
if not token:
|
||||
return None
|
||||
token_hash = hashlib.sha256(token.encode()).hexdigest()
|
||||
now = int(time.time())
|
||||
with self.lock:
|
||||
row = self.conn.execute(
|
||||
"SELECT * FROM sessions WHERE token_hash=? AND expires_at>?", (token_hash, now)
|
||||
).fetchone()
|
||||
return dict(row) if row else None
|
||||
|
||||
def logout(self, token: str | None) -> None:
|
||||
if not token:
|
||||
return
|
||||
with self.lock, self.conn:
|
||||
self.conn.execute("DELETE FROM sessions WHERE token_hash=?", (hashlib.sha256(token.encode()).hexdigest(),))
|
||||
|
||||
def student(self, student_id: int) -> dict | None:
|
||||
with self.lock:
|
||||
row = self.conn.execute("SELECT * FROM students WHERE id=?", (student_id,)).fetchone()
|
||||
return dict(row) if row else None
|
||||
|
||||
def ensure_test_students(self) -> None:
|
||||
with self.lock, self.conn:
|
||||
for class_name, student_no, name in TEST_STUDENTS:
|
||||
exists = self.conn.execute(
|
||||
"SELECT id FROM students WHERE student_no=?", (student_no,)
|
||||
).fetchone()
|
||||
if exists:
|
||||
continue
|
||||
self.conn.execute(
|
||||
"""INSERT INTO students
|
||||
(class_name,student_no,name,gender,admission_no,remark,password_hash,must_change_password)
|
||||
VALUES(?,?,?,'','','测试账号',?,0)""",
|
||||
(class_name, student_no, name, hash_password(TEST_PASSWORD)),
|
||||
)
|
||||
LOG.info("已创建测试账号 %s / %s", student_no, TEST_PASSWORD)
|
||||
|
||||
def teacher_set_student_password(self, student_id: int, new_password: str) -> tuple[bool, str]:
|
||||
new_password = new_password or ""
|
||||
if len(new_password) < 8:
|
||||
return False, "新密码至少 8 位"
|
||||
with self.lock, self.conn:
|
||||
student = self.conn.execute(
|
||||
"SELECT student_no,name FROM students WHERE id=?", (student_id,)
|
||||
).fetchone()
|
||||
if student is None:
|
||||
return False, "学生不存在"
|
||||
if new_password == student["student_no"]:
|
||||
return False, "新密码不能与学号相同"
|
||||
self.conn.execute(
|
||||
"UPDATE students SET password_hash=?, must_change_password=0 WHERE id=?",
|
||||
(hash_password(new_password), student_id),
|
||||
)
|
||||
self.conn.execute("DELETE FROM sessions WHERE student_id=?", (student_id,))
|
||||
return True, f"已修改 {student['student_no']} {student['name']} 的密码"
|
||||
|
||||
def create_student(self, class_name: str, student_no: str, name: str, password: str) -> tuple[bool, str]:
|
||||
class_name = (class_name or "").strip()
|
||||
student_no = (student_no or "").strip()
|
||||
name = (name or "").strip()
|
||||
password = password or ""
|
||||
if not class_name or not student_no or not name:
|
||||
return False, "班级、学号和姓名不能为空"
|
||||
if len(password) < 8:
|
||||
return False, "密码至少 8 位"
|
||||
if password == student_no:
|
||||
return False, "密码不能与学号相同"
|
||||
with self.lock, self.conn:
|
||||
exists = self.conn.execute(
|
||||
"SELECT 1 FROM students WHERE student_no=?", (student_no,)
|
||||
).fetchone()
|
||||
if exists:
|
||||
return False, "学号已存在"
|
||||
self.conn.execute(
|
||||
"""INSERT INTO students
|
||||
(class_name,student_no,name,gender,admission_no,remark,password_hash,must_change_password)
|
||||
VALUES(?,?,?,'','','教师添加',?,0)""",
|
||||
(class_name, student_no, name, hash_password(password)),
|
||||
)
|
||||
return True, f"已添加 {student_no} {name}"
|
||||
|
||||
def change_password(self, student_id: int, old: str, new: str) -> tuple[bool, str]:
|
||||
if len(new) < 6:
|
||||
return False, "新密码至少 6 位"
|
||||
row = self.student(student_id)
|
||||
if not row or not verify_password(old, row["password_hash"]):
|
||||
return False, "原密码不正确"
|
||||
with self.lock, self.conn:
|
||||
self.conn.execute(
|
||||
"UPDATE students SET password_hash=?, must_change_password=0 WHERE id=?",
|
||||
(hash_password(new), student_id),
|
||||
)
|
||||
return True, "密码已修改"
|
||||
|
||||
def touch(self, student_id: int) -> None:
|
||||
with self.lock, self.conn:
|
||||
self.conn.execute(
|
||||
"INSERT INTO trainer_presence(student_id,last_seen) VALUES(?,?) "
|
||||
"ON CONFLICT(student_id) DO UPDATE SET last_seen=excluded.last_seen",
|
||||
(student_id, int(time.time())),
|
||||
)
|
||||
|
||||
def active_round(self) -> dict | None:
|
||||
with self.lock:
|
||||
row = self.conn.execute(
|
||||
"SELECT * FROM trainer_rounds WHERE active=1 ORDER BY id DESC LIMIT 1"
|
||||
).fetchone()
|
||||
if not row:
|
||||
return None
|
||||
return {**dict(row), "expected": json.loads(row["expected_json"])}
|
||||
|
||||
def start_round(self, label: str, expected: dict) -> dict:
|
||||
now = int(time.time())
|
||||
with self.lock, self.conn:
|
||||
self.conn.execute("UPDATE trainer_rounds SET active=0 WHERE active=1")
|
||||
self._progress = {}
|
||||
cur = self.conn.execute(
|
||||
"INSERT INTO trainer_rounds(label,expected_json,active,created_at) VALUES(?,?,1,?)",
|
||||
(label, json.dumps(expected, ensure_ascii=False), now),
|
||||
)
|
||||
round_id = cur.lastrowid
|
||||
return {"id": round_id, "label": label, "expected": expected, "active": 1, "created_at": now}
|
||||
|
||||
def stop_round(self) -> None:
|
||||
with self.lock, self.conn:
|
||||
self.conn.execute("UPDATE trainer_rounds SET active=0 WHERE active=1")
|
||||
|
||||
def add_event(self, student_id: int, data: dict) -> dict:
|
||||
kind = str(data.get("type", "")).lower()
|
||||
aliases = {"page": "page", "button": "button", "shortcut": "shortcut", "key": "shortcut",
|
||||
"menu": "menu", "window": "window", "url": "url", "link": "url"}
|
||||
kind = aliases.get(kind, "")
|
||||
if not kind:
|
||||
return {"ok": False, "message": "未知操作类型"}
|
||||
round_id = self.active_round()["id"] if self.active_round() else None
|
||||
page_no = int(data.get("page") or data.get("pageNo") or 0)
|
||||
try:
|
||||
item_index = int(data.get("index") or data.get("button") or 0)
|
||||
except (TypeError, ValueError):
|
||||
item_index = 0
|
||||
value = str(data.get("key") or data.get("action") or data.get("url") or data.get("name") or data.get("item") or "")
|
||||
now = int(time.time())
|
||||
with self.lock, self.conn:
|
||||
self.conn.execute(
|
||||
"INSERT INTO trainer_events(round_id,student_id,kind,page_no,item_index,value,created_at) VALUES(?,?,?,?,?,?,?)",
|
||||
(round_id, student_id, kind, page_no, item_index, value, now),
|
||||
)
|
||||
result = self.score(student_id, round_id, kind, page_no, item_index, value) if round_id else None
|
||||
return {
|
||||
"ok": True,
|
||||
"accepted": True,
|
||||
"correct": bool(result),
|
||||
"roundId": round_id,
|
||||
"result": result,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def normalize_key(value: str) -> str:
|
||||
parts = [p.strip().upper() for p in str(value).replace(" ", "").split("+") if p.strip()]
|
||||
mods = [m for m in ("CTRL", "CONTROL", "ALT", "SHIFT", "META", "CMD") if m in parts or (m == "CTRL" and "CONTROL" in parts)]
|
||||
mods = ["CTRL" if m in ("CTRL", "CONTROL") else m for m in mods]
|
||||
rest = [p for p in parts if p not in ("CTRL", "CONTROL", "ALT", "SHIFT", "META", "CMD")]
|
||||
return "+".join(mods + rest)
|
||||
|
||||
def _step_match(self, step: dict, kind: str, page_no: int, item_index: int, value: str) -> bool:
|
||||
step_type = str(step.get("type", "")).lower()
|
||||
if kind != step_type:
|
||||
return False
|
||||
if kind == "page":
|
||||
return page_no == int(step.get("page", 0) or 0)
|
||||
if kind == "button":
|
||||
return (
|
||||
(not step.get("index") or item_index == int(step["index"]))
|
||||
and (not step.get("name") or value == str(step["name"]))
|
||||
and (not step.get("page") or page_no == int(step["page"]))
|
||||
)
|
||||
if kind == "shortcut":
|
||||
return self.normalize_key(value) == self.normalize_key(step.get("key", ""))
|
||||
if kind == "menu":
|
||||
return str(step.get("item", "")).lower() == value.lower()
|
||||
if kind == "window":
|
||||
return str(step.get("action", "")).lower() == value.lower()
|
||||
if kind == "url":
|
||||
expected = str(step.get("url", "")).lower().replace("https://", "").replace("http://", "").lstrip("www.")
|
||||
actual = value.lower().replace("https://", "").replace("http://", "").lstrip("www.")
|
||||
return expected in actual
|
||||
return False
|
||||
|
||||
def score(self, student_id: int, round_id: int, kind: str, page_no: int, item_index: int, value: str) -> dict | None:
|
||||
with self.lock:
|
||||
row = self.conn.execute(
|
||||
"SELECT * FROM trainer_rounds WHERE id=? AND active=1", (round_id,)
|
||||
).fetchone()
|
||||
if not row:
|
||||
return None
|
||||
steps = json.loads(row["expected_json"]).get("steps", [])
|
||||
key = f"{round_id}:{student_id}"
|
||||
if not hasattr(self, "_progress"):
|
||||
self._progress = {}
|
||||
progress = self._progress.get(key, 0)
|
||||
if progress >= len(steps):
|
||||
return None
|
||||
while progress < len(steps):
|
||||
step = steps[progress]
|
||||
if (
|
||||
kind != "page"
|
||||
and str(step.get("type", "")).lower() == "page"
|
||||
and page_no == int(step.get("page") or 0)
|
||||
):
|
||||
progress += 1
|
||||
continue
|
||||
break
|
||||
if progress < len(steps):
|
||||
if not self._step_match(steps[progress], kind, page_no, item_index, value):
|
||||
self._progress[key] = progress
|
||||
return None
|
||||
progress += 1
|
||||
if progress < len(steps):
|
||||
self._progress[key] = progress
|
||||
return None
|
||||
self._progress.pop(key, None)
|
||||
|
||||
now = int(time.time())
|
||||
with self.lock, self.conn:
|
||||
self.conn.execute(
|
||||
"""INSERT OR IGNORE INTO trainer_scores(round_id,student_id,correct,finished_at,payload)
|
||||
VALUES(?,?,?,?,?)""",
|
||||
(round_id, student_id, 1, now, json.dumps({"page": page_no, "index": item_index, "value": value}, ensure_ascii=False)),
|
||||
)
|
||||
row = self.conn.execute(
|
||||
"""SELECT rank_no AS rank, total, finished_at FROM (
|
||||
SELECT student_id, finished_at,
|
||||
ROW_NUMBER() OVER(ORDER BY finished_at, student_id) AS rank_no,
|
||||
COUNT(*) OVER() AS total
|
||||
FROM trainer_scores WHERE round_id=? AND correct=1
|
||||
) ranked WHERE student_id=?""",
|
||||
(round_id, student_id),
|
||||
).fetchone()
|
||||
if not row:
|
||||
return {"rank": 1, "total": 1, "finishedAt": now}
|
||||
return {"rank": int(row["rank"]), "total": int(row["total"]), "finishedAt": int(row["finished_at"])}
|
||||
|
||||
def student_result(self, student_id: int, round_id: int | None = None) -> dict | None:
|
||||
if round_id is None:
|
||||
active = self.active_round()
|
||||
round_id = active["id"] if active else None
|
||||
if not round_id:
|
||||
return None
|
||||
with self.lock:
|
||||
row = self.conn.execute(
|
||||
"SELECT finished_at FROM trainer_scores WHERE round_id=? AND student_id=? AND correct=1",
|
||||
(round_id, student_id),
|
||||
).fetchone()
|
||||
if not row:
|
||||
return None
|
||||
total = self.conn.execute(
|
||||
"SELECT COUNT(*) AS n FROM trainer_scores WHERE round_id=? AND correct=1",
|
||||
(round_id,),
|
||||
).fetchone()["n"]
|
||||
rank = self.conn.execute(
|
||||
"""SELECT COUNT(*) AS n FROM trainer_scores
|
||||
WHERE round_id=? AND correct=1 AND (
|
||||
finished_at<? OR (finished_at=? AND student_id<=?)
|
||||
)""",
|
||||
(round_id, row["finished_at"], row["finished_at"], student_id),
|
||||
).fetchone()["n"]
|
||||
return {"rank": int(rank or 1), "total": int(total or 1), "finishedAt": int(row["finished_at"])}
|
||||
|
||||
def teacher_state(self) -> dict:
|
||||
cutoff = int(time.time() - 25)
|
||||
active = self.active_round()
|
||||
round_id = active["id"] if active else None
|
||||
with self.lock:
|
||||
students = self.conn.execute(
|
||||
"""SELECT id,class_name,student_no,name,must_change_password
|
||||
FROM students ORDER BY class_name,student_no"""
|
||||
).fetchall()
|
||||
presence = {
|
||||
r["student_id"]: r["last_seen"]
|
||||
for r in self.conn.execute(
|
||||
"SELECT student_id,last_seen FROM trainer_presence WHERE last_seen>?", (cutoff,)
|
||||
)
|
||||
}
|
||||
scores = []
|
||||
events = []
|
||||
if round_id:
|
||||
scores = self.conn.execute(
|
||||
"""SELECT s.rank_no AS rank, st.name, st.student_no, st.class_name, sc.finished_at
|
||||
FROM (SELECT ROW_NUMBER() OVER(ORDER BY finished_at) AS rank_no, student_id, finished_at
|
||||
FROM trainer_scores WHERE round_id=? AND correct=1) s
|
||||
JOIN students st ON st.id=s.student_id
|
||||
JOIN trainer_scores sc ON sc.round_id=? AND sc.student_id=s.student_id
|
||||
ORDER BY s.rank_no LIMIT 30""",
|
||||
(round_id, round_id),
|
||||
).fetchall()
|
||||
events = self.conn.execute(
|
||||
"""SELECT e.id, e.kind, e.page_no, e.item_index, e.value, e.created_at,
|
||||
st.name, st.student_no
|
||||
FROM trainer_events e JOIN students st ON st.id=e.student_id
|
||||
WHERE e.round_id=? ORDER BY e.id DESC LIMIT 100""",
|
||||
(round_id,),
|
||||
).fetchall()
|
||||
return {
|
||||
"round": active,
|
||||
"students": [
|
||||
{**dict(s), "online": s["id"] in presence, "finished": False}
|
||||
for s in students
|
||||
],
|
||||
"scores": [dict(r) for r in scores],
|
||||
"events": [dict(r) for r in events],
|
||||
"pages": PAGES,
|
||||
}
|
||||
|
||||
|
||||
STORE = Store(DB_PATH)
|
||||
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
server_version = "ShortcutTrainer/1.0"
|
||||
|
||||
def log_message(self, fmt, *args):
|
||||
return
|
||||
|
||||
def do_GET(self):
|
||||
path = urlsplit(self.path).path
|
||||
if path.startswith("/api/"):
|
||||
self.api_get(path)
|
||||
return
|
||||
if path == "/":
|
||||
self.serve("index.html", "text/html; charset=utf-8")
|
||||
elif path.startswith("/page/"):
|
||||
self.serve("page.html", "text/html; charset=utf-8")
|
||||
elif path == "/teacher":
|
||||
self.serve("teacher.html", "text/html; charset=utf-8")
|
||||
else:
|
||||
self.static(path)
|
||||
|
||||
def do_POST(self):
|
||||
path = urlsplit(self.path).path
|
||||
if not path.startswith("/api/"):
|
||||
self.json(HTTPStatus.NOT_FOUND, {"ok": False, "message": "Not Found"})
|
||||
return
|
||||
try:
|
||||
length = min(int(self.headers.get("Content-Length", "0")), 1024 * 256)
|
||||
data = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
|
||||
except Exception:
|
||||
self.json(HTTPStatus.BAD_REQUEST, {"ok": False, "message": "请求格式错误"})
|
||||
return
|
||||
action = path[len("/api/"):]
|
||||
if action == "login":
|
||||
result, msg = STORE.login_student(str(data.get("username", "")), str(data.get("password", "")))
|
||||
if result:
|
||||
self.set_cookie(result["token"])
|
||||
self.json(HTTPStatus.OK, {"ok": True, "message": msg, "student": {
|
||||
"name": result["student"]["name"], "studentNo": result["student"]["student_no"],
|
||||
"className": result["student"]["class_name"],
|
||||
"mustChangePassword": bool(result["student"]["must_change_password"])
|
||||
}})
|
||||
else:
|
||||
self.json(HTTPStatus.OK, {"ok": False, "message": msg})
|
||||
elif action == "teacher/login":
|
||||
result, msg = STORE.login_teacher(str(data.get("username", "")), str(data.get("password", "")))
|
||||
if result:
|
||||
self.set_cookie(result["token"])
|
||||
self.json(HTTPStatus.OK, {"ok": bool(result), "message": msg})
|
||||
elif action == "logout":
|
||||
STORE.logout(self.cookie_token())
|
||||
self.set_cookie("", expire=True)
|
||||
self.json(HTTPStatus.OK, {"ok": True})
|
||||
elif action == "password":
|
||||
session = STORE.session(self.cookie_token())
|
||||
if not session or session["role"] != "student":
|
||||
self.json(HTTPStatus.UNAUTHORIZED, {"ok": False, "message": "请先登录"})
|
||||
return
|
||||
ok, msg = STORE.change_password(session["student_id"], str(data.get("oldPassword", "")), str(data.get("newPassword", "")))
|
||||
self.json(HTTPStatus.OK, {"ok": ok, "message": msg})
|
||||
elif action == "event":
|
||||
session = STORE.session(self.cookie_token())
|
||||
if not session or session["role"] != "student":
|
||||
self.json(HTTPStatus.UNAUTHORIZED, {"ok": False, "message": "未登录"})
|
||||
return
|
||||
STORE.touch(session["student_id"])
|
||||
self.json(HTTPStatus.OK, STORE.add_event(session["student_id"], data))
|
||||
elif action == "teacher/round":
|
||||
if not self.require_teacher():
|
||||
return
|
||||
expected = data.get("expected")
|
||||
if not isinstance(expected, dict) or not isinstance(expected.get("steps"), list) or not expected["steps"]:
|
||||
self.json(HTTPStatus.OK, {"ok": False, "message": "请提供 expected.steps"})
|
||||
return
|
||||
round_data = STORE.start_round(str(data.get("label", "新指令")), expected)
|
||||
self.json(HTTPStatus.OK, {"ok": True, "round": round_data})
|
||||
elif action == "teacher/stop":
|
||||
if not self.require_teacher():
|
||||
return
|
||||
STORE.stop_round()
|
||||
self.json(HTTPStatus.OK, {"ok": True})
|
||||
elif action == "teacher/student-password":
|
||||
if not self.require_teacher():
|
||||
return
|
||||
try:
|
||||
student_id = int(data.get("studentId") or data.get("student_id") or 0)
|
||||
except (TypeError, ValueError):
|
||||
student_id = 0
|
||||
ok, msg = STORE.teacher_set_student_password(
|
||||
student_id, str(data.get("newPassword") or data.get("new_password") or "")
|
||||
)
|
||||
self.json(HTTPStatus.OK, {"ok": ok, "message": msg})
|
||||
elif action == "teacher/student":
|
||||
if not self.require_teacher():
|
||||
return
|
||||
ok, msg = STORE.create_student(
|
||||
str(data.get("className") or data.get("class_name") or ""),
|
||||
str(data.get("studentNo") or data.get("student_no") or ""),
|
||||
str(data.get("name") or ""),
|
||||
str(data.get("password") or ""),
|
||||
)
|
||||
self.json(HTTPStatus.OK, {"ok": ok, "message": msg})
|
||||
else:
|
||||
self.json(HTTPStatus.NOT_FOUND, {"ok": False, "message": "接口不存在"})
|
||||
|
||||
def api_get(self, path: None):
|
||||
return
|
||||
|
||||
def require_teacher(self) -> bool:
|
||||
session = STORE.session(self.cookie_token())
|
||||
if not session or session["role"] != "teacher":
|
||||
self.json(HTTPStatus.UNAUTHORIZED, {"ok": False, "message": "请先登录教师端"})
|
||||
return False
|
||||
return True
|
||||
|
||||
def cookie_token(self) -> str | None:
|
||||
raw = self.headers.get("Cookie", "")
|
||||
for item in raw.split(";"):
|
||||
if "=" in item:
|
||||
k, v = item.strip().split("=", 1)
|
||||
if k == "trainer_session":
|
||||
return unquote(v)
|
||||
return None
|
||||
|
||||
def set_cookie(self, token: str, expire: bool = False):
|
||||
self.cookie_to_set = (token or "", expire)
|
||||
|
||||
def json(self, status: int, payload: dict):
|
||||
body = json.dumps(payload, ensure_ascii=False).encode()
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||||
self.send_header("Cache-Control", "no-store")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
if getattr(self, "cookie_to_set", None) is not None:
|
||||
value, expire = self.cookie_to_set
|
||||
max_age = 0 if expire else SESSION_DAYS * 86400
|
||||
self.send_header("Set-Cookie", f"trainer_session={value}; Path=/; Max-Age={max_age}; SameSite=Lax")
|
||||
self.cookie_to_set = None
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def static(self, path: str):
|
||||
name = path.lstrip("/")
|
||||
types = {".css": "text/css; charset=utf-8", ".js": "text/javascript; charset=utf-8",
|
||||
".svg": "image/svg+xml", ".png": "image/png", ".ico": "image/x-icon"}
|
||||
suffix = Path(name).suffix.lower()
|
||||
safe = Path(name).name
|
||||
if suffix not in types or not (STATIC_DIR / safe).is_file():
|
||||
self.send_response(HTTPStatus.NOT_FOUND)
|
||||
self.end_headers()
|
||||
return
|
||||
self.serve(safe, types[suffix])
|
||||
|
||||
def serve(self, name: str, content_type: str):
|
||||
body = (STATIC_DIR / name).read_bytes()
|
||||
self.send_response(HTTPStatus.OK)
|
||||
self.send_header("Content-Type", content_type)
|
||||
self.send_header("Cache-Control", "no-cache")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
|
||||
def student_state(session: dict) -> dict:
|
||||
student = STORE.student(session["student_id"])
|
||||
STORE.touch(session["student_id"])
|
||||
active = STORE.active_round()
|
||||
safe_student = None
|
||||
if student:
|
||||
safe_student = {
|
||||
"id": student["id"],
|
||||
"className": student["class_name"],
|
||||
"studentNo": student["student_no"],
|
||||
"name": student["name"],
|
||||
"mustChangePassword": bool(student["must_change_password"]),
|
||||
}
|
||||
if active:
|
||||
active = {k: active[k] for k in ("id", "label", "active", "created_at")}
|
||||
result = STORE.student_result(session["student_id"]) if student else None
|
||||
return {
|
||||
"ok": True,
|
||||
"student": safe_student,
|
||||
"round": active,
|
||||
"result": result,
|
||||
"pages": PAGES,
|
||||
}
|
||||
|
||||
|
||||
# 把学生端接口挂到 Handler；为保持单文件清晰，这里用方法注入。
|
||||
def api_get_handler(handler, path: str):
|
||||
action = path[len("/api/"):]
|
||||
if action in ("me", "student/state"):
|
||||
session = STORE.session(handler.cookie_token())
|
||||
if not session or session["role"] != "student":
|
||||
handler.json(HTTPStatus.OK, {"ok": True, "authenticated": False})
|
||||
return
|
||||
handler.json(HTTPStatus.OK, {**student_state(session), "authenticated": True})
|
||||
elif action == "teacher/state":
|
||||
if not handler.require_teacher():
|
||||
return
|
||||
handler.json(HTTPStatus.OK, {"ok": True, **STORE.teacher_state()})
|
||||
else:
|
||||
handler.json(HTTPStatus.NOT_FOUND, {"ok": False, "message": "接口不存在"})
|
||||
|
||||
|
||||
Handler.api_get = api_get_handler
|
||||
|
||||
|
||||
class DualStackHTTPServer(ThreadingHTTPServer):
|
||||
"""监听通配 IPv6 地址，同时接受 IPv4 连接。"""
|
||||
|
||||
address_family = socket.AF_INET6
|
||||
|
||||
def server_bind(self) -> None:
|
||||
if hasattr(socket, "IPPROTO_IPV6") and hasattr(socket, "IPV6_V6ONLY"):
|
||||
self.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
|
||||
super().server_bind()
|
||||
|
||||
|
||||
def discover_local_addresses() -> list[str]:
|
||||
"""尽量列出版本库常见部署环境里的本机地址。"""
|
||||
addresses: set[str] = {"127.0.0.1", "::1"}
|
||||
if os.name == "nt":
|
||||
try:
|
||||
text = subprocess.run(
|
||||
["ipconfig"], capture_output=True, text=True, check=False, timeout=2
|
||||
).stdout
|
||||
for line in text.splitlines():
|
||||
if re.search(r"IPv4|IP Address|IP 地址", line, re.IGNORECASE):
|
||||
match = re.search(r"(\d{1,3}(?:\.\d{1,3}){3})", line)
|
||||
if match:
|
||||
addresses.add(match.group(1))
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
pass
|
||||
else:
|
||||
try:
|
||||
command = ["ip", "-o", "address"] if shutil.which("ip") else ["ifconfig"]
|
||||
text = subprocess.run(
|
||||
command, capture_output=True, text=True, check=False, timeout=2
|
||||
).stdout
|
||||
for match in re.finditer(r"inet (?:addr:)?(\d{1,3}(?:\.\d{1,3}){3})", text):
|
||||
addresses.add(match.group(1))
|
||||
for match in re.finditer(r"inet6 ([0-9a-f:]+)", text, re.IGNORECASE):
|
||||
if not match.group(1).lower().startswith("fe80"):
|
||||
addresses.add(match.group(1))
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
pass
|
||||
|
||||
names = {socket.gethostname(), "localhost"}
|
||||
try:
|
||||
names.add(socket.getfqdn())
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
for name in tuple(names):
|
||||
try:
|
||||
addresses.update(socket.gethostbyname_ex(name)[2])
|
||||
except OSError:
|
||||
pass
|
||||
try:
|
||||
for info in socket.getaddrinfo(name, PORT, type=socket.SOCK_STREAM):
|
||||
addresses.add(info[4][0])
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
skip = {"0.0.0.0", "::", "0:0:0:0:0:0:0:0"}
|
||||
return sorted((item for item in addresses if item not in skip), key=lambda x: (":" in x, x))
|
||||
|
||||
|
||||
def server_url(address: str) -> str:
|
||||
display = f"[{address}]" if ":" in address else address
|
||||
return f"http://{display}:{PORT}"
|
||||
|
||||
|
||||
def main():
|
||||
bind_host = "::" if HOST in {"", "*", "0.0.0.0"} else HOST
|
||||
try:
|
||||
server = DualStackHTTPServer((bind_host, PORT), Handler)
|
||||
listen_mode = "IPv4 + IPv6 全部网卡"
|
||||
except OSError as exc:
|
||||
# 只在系统不支持 IPv6/双栈时退回 IPv4；端口被占用仍应直接暴露错误。
|
||||
if exc.errno not in {
|
||||
errno.EADDRNOTAVAIL,
|
||||
errno.EAFNOSUPPORT,
|
||||
errno.EPROTONOSUPPORT,
|
||||
}:
|
||||
raise
|
||||
server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
|
||||
listen_mode = "IPv4 全部网卡"
|
||||
|
||||
LOG.info("快捷键训练系统已启动，监听：%s", listen_mode)
|
||||
for address in discover_local_addresses():
|
||||
LOG.info("学生端：%s", server_url(address))
|
||||
LOG.info("教师端：%s/teacher", server_url(address))
|
||||
try:
|
||||
server.serve_forever()
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
finally:
|
||||
server.server_close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user