linux-practice/server.py

#!/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")
PRIVACY_FILE = os.path.join(os.path.dirname(__file__), "privacy.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 ["/", "/privacy", "/privacy.html", "/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 in {"/privacy", "/privacy.html"}:
            self.send_file(PRIVACY_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()