228 lines
8.1 KiB
Python
Executable File
228 lines
8.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
Linux 学习平台 Server（知识导向版）
|
||
- 提供课程结构、练习判题、沙盒执行
|
||
- 课程模型：module -> lesson -> exercise
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import http.server
|
||
import json
|
||
import os
|
||
import urllib.parse
|
||
from typing import Any
|
||
|
||
from sandbox import LinuxSandbox
|
||
|
||
USERS = {
|
||
"admin": hashlib.sha256(b"safe_linux_2026").hexdigest(),
|
||
}
|
||
|
||
TASKS_FILE = os.path.join(os.path.dirname(__file__), "COURSE_TASKS.json")
|
||
HTML_FILE = os.path.join(os.path.dirname(__file__), "index.html")
|
||
SANDBOX = LinuxSandbox()
|
||
|
||
|
||
def load_course() -> dict[str, Any]:
|
||
try:
|
||
with open(TASKS_FILE, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception as e:
|
||
print(f"Error loading course: {e}")
|
||
return {"meta": {}, "modules": []}
|
||
|
||
|
||
COURSE = load_course()
|
||
|
||
|
||
def flatten_exercises() -> list[dict[str, Any]]:
|
||
rows: list[dict[str, Any]] = []
|
||
for module in COURSE.get("modules", []):
|
||
for lesson in module.get("lessons", []):
|
||
for exercise in lesson.get("exercises", []):
|
||
item = dict(exercise)
|
||
item["module_id"] = module.get("id")
|
||
item["module_title"] = module.get("title")
|
||
item["lesson_id"] = lesson.get("id")
|
||
item["lesson_title"] = lesson.get("title")
|
||
item["lesson_goal"] = lesson.get("goal")
|
||
item["lesson_command"] = lesson.get("command")
|
||
rows.append(item)
|
||
return rows
|
||
|
||
|
||
def find_exercise(ex_id: str) -> dict[str, Any] | None:
|
||
for item in flatten_exercises():
|
||
if item.get("id") == ex_id:
|
||
return item
|
||
return None
|
||
|
||
|
||
class LinuxLearningHandler(http.server.BaseHTTPRequestHandler):
|
||
def log_message(self, format, *args):
|
||
pass
|
||
|
||
def send_json(self, data: Any, status=200):
|
||
raw = json.dumps(data, ensure_ascii=False).encode("utf-8")
|
||
self.send_response(status)
|
||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||
self.send_header("Content-Length", str(len(raw)))
|
||
self.end_headers()
|
||
self.wfile.write(raw)
|
||
|
||
def send_file(self, path: str, content_type: str):
|
||
with open(path, "rb") as f:
|
||
body = f.read()
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", content_type)
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def check_auth(self, auth_header: str, token: str) -> bool:
|
||
if self.client_address[0] == "127.0.0.1":
|
||
return True
|
||
if token == "safe_linux_2026":
|
||
return True
|
||
if auth_header.startswith("Bearer ") and auth_header[7:] == "safe_linux_2026":
|
||
return True
|
||
return False
|
||
|
||
def do_GET(self):
|
||
parsed = urllib.parse.urlparse(self.path)
|
||
path = parsed.path
|
||
|
||
if path not in ["/", "/api/login", "/api/logout", "/api/course", "/api/health"]:
|
||
auth_header = self.headers.get("Authorization", "")
|
||
token = self.headers.get("X-Token", "")
|
||
if not self.check_auth(auth_header, token) and "xiaoxiaoluohao.indevs.in" in self.headers.get("Host", ""):
|
||
self.send_json({"error": "Authentication required"}, 401)
|
||
return
|
||
|
||
if path == "/":
|
||
self.send_file(HTML_FILE, "text/html; charset=utf-8")
|
||
return
|
||
|
||
if path == "/api/health":
|
||
self.send_json({"ok": True, "cwd": SANDBOX.cwd, "user": SANDBOX.user, "modules": len(COURSE.get("modules", []))})
|
||
return
|
||
|
||
if path == "/api/course":
|
||
self.send_json(COURSE)
|
||
return
|
||
|
||
if path == "/api/run":
|
||
query = urllib.parse.parse_qs(parsed.query)
|
||
cmd = query.get("cmd", [""])[0]
|
||
if not cmd:
|
||
self.send_json({"error": "No command provided"}, 400)
|
||
return
|
||
self.send_json(SANDBOX.execute(cmd))
|
||
return
|
||
|
||
if path == "/api/check":
|
||
query = urllib.parse.parse_qs(parsed.query)
|
||
ex_id = query.get("exercise_id", [""])[0]
|
||
cmd = query.get("last_cmd", [""])[0]
|
||
output = query.get("output", [""])[0]
|
||
if not ex_id:
|
||
self.send_json({"error": "exercise_id required"}, 400)
|
||
return
|
||
exercise = find_exercise(ex_id)
|
||
if not exercise:
|
||
self.send_json({"error": "Exercise not found"}, 404)
|
||
return
|
||
state = {
|
||
"cmd": cmd,
|
||
"output": output,
|
||
"cwd": SANDBOX.cwd,
|
||
"exists": SANDBOX.exists,
|
||
"is_executable": SANDBOX.is_executable,
|
||
}
|
||
success, reason = self.evaluate_exercise(exercise, state)
|
||
self.send_json({
|
||
"exercise_id": ex_id,
|
||
"success": success,
|
||
"message": exercise.get("success_msg", "✅ 练习通过") if success else reason,
|
||
"hint": exercise.get("hint"),
|
||
"lesson_title": exercise.get("lesson_title"),
|
||
"module_title": exercise.get("module_title"),
|
||
"next_suggestion": self.build_next_suggestion(ex_id),
|
||
})
|
||
return
|
||
|
||
self.send_response(404)
|
||
self.end_headers()
|
||
self.wfile.write(b"Not Found")
|
||
|
||
def do_POST(self):
|
||
parsed = urllib.parse.urlparse(self.path)
|
||
path = parsed.path
|
||
content_length = int(self.headers.get("Content-Length", 0))
|
||
raw = self.rfile.read(content_length).decode() if content_length else "{}"
|
||
|
||
if path == "/api/login":
|
||
try:
|
||
data = json.loads(raw)
|
||
except Exception:
|
||
self.send_json({"success": False, "error": "Invalid JSON"}, 400)
|
||
return
|
||
username = data.get("username", "")
|
||
password = data.get("password", "")
|
||
if username in USERS and hashlib.sha256(password.encode()).hexdigest() == USERS[username]:
|
||
self.send_json({"success": True, "token": "safe_linux_2026", "message": "✅ 登录成功！"})
|
||
return
|
||
self.send_json({"success": False, "error": "❌ 用户名或密码错误"}, 401)
|
||
return
|
||
|
||
if path == "/api/logout":
|
||
self.send_json({"success": True, "message": "👋 已退出登录"})
|
||
return
|
||
|
||
if path == "/api/reset":
|
||
SANDBOX.reset()
|
||
self.send_json({"success": True, "message": "♻️ 沙盒环境已重置"})
|
||
return
|
||
|
||
self.send_response(404)
|
||
self.end_headers()
|
||
self.wfile.write(b"Not Found")
|
||
|
||
def evaluate_exercise(self, exercise: dict[str, Any], state: dict[str, Any]) -> tuple[bool, str]:
|
||
ex_type = exercise.get("type")
|
||
if ex_type in {"understanding", "scenario"}:
|
||
return False, "📝 这是理解类练习，请先阅读讲解并思考答案。"
|
||
|
||
cmd = state["cmd"].strip()
|
||
if exercise.get("solution"):
|
||
for sol in exercise["solution"]:
|
||
if cmd == sol.strip():
|
||
return True, ""
|
||
|
||
success_test = exercise.get("success_test")
|
||
if not success_test:
|
||
return False, "❌ 暂未命中练习要求"
|
||
|
||
try:
|
||
ok = bool(eval(success_test, {"__builtins__": {}}, state))
|
||
return ok, "❌ 结果还没达到练习要求，再试一次"
|
||
except Exception:
|
||
return False, "❌ 当前命令没有通过判定，建议对照示例重新尝试"
|
||
|
||
def build_next_suggestion(self, current_ex_id: str) -> str | None:
|
||
rows = flatten_exercises()
|
||
for i, item in enumerate(rows):
|
||
if item.get("id") == current_ex_id and i + 1 < len(rows):
|
||
nxt = rows[i + 1]
|
||
return f"继续下一练：{nxt.get('title') or nxt.get('question') or nxt.get('id')}"
|
||
return None
|
||
|
||
|
||
if __name__ == "__main__":
|
||
port = 8084
|
||
print(f"🐧 Linux 学习平台启动中... http://127.0.0.1:{port}")
|
||
print("📚 线上地址: https://linux.xiaoxiaoluohao.indevs.in")
|
||
http.server.ThreadingHTTPServer(("127.0.0.1", port), LinuxLearningHandler).serve_forever()
|