server.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""CodeGym — AI 可填充的编程教学框架：stdlib HTTP + SQLite + 子进程判题。
设计铁律：进度只由证据解锁。学生无法自标完成。"""
import hashlib
import hmac
import json
import sqlite3
import urllib.parse
import subprocess
import sys
import tempfile
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

ROOT = Path(__file__).parent
COURSES = ROOT / "courses"
import importlib
import os
PORT = int(os.environ.get("CODEGYM_PORT", "8775"))
TIMEOUT_S = 10
RETENTION_STAGES = [1, 3, 7, 14, 30]  # 冷启动通过后的变体回访间隔（天）

_TLS = threading.local()

# ---- 口令门（局域网暴露必须有门，家里规矩）：config.json 顶层 code，X-Code 头比对 ----
_GATE_CODE = ""
_FAIL_LOCK_UNTIL = 0.0
_FAIL_COUNT = 0
_GATE_LOCK = threading.Lock()


def _load_gate_code():
    global _GATE_CODE
    f = Path(__file__).parent / "config.json"
    if not f.exists():
        return
    try:
        cfg = json.loads(f.read_text(encoding="utf-8"))
        code = cfg.get("code")
        if isinstance(code, str) and code:
            _GATE_CODE = code
    except (json.JSONDecodeError, OSError) as e:
        print(f"[CodeGym] config.json 读不了，口令门不开：{e}", flush=True)


def _gate_check(headers):
    global _FAIL_LOCK_UNTIL, _FAIL_COUNT
    if not _GATE_CODE:
        return None
    now = time.time()
    with _GATE_LOCK:
        if now < _FAIL_LOCK_UNTIL:
            return 429, f"试错太多次，锁 {int(_FAIL_LOCK_UNTIL - now) + 1} 秒"
    supplied = headers.get("X-Code", "")
    if supplied and hmac.compare_digest(supplied.encode("utf-8"), _GATE_CODE.encode("utf-8")):
        with _GATE_LOCK:
            _FAIL_COUNT = 0
        return None
    with _GATE_LOCK:
        _FAIL_COUNT += 1
        if _FAIL_COUNT >= 5:
            _FAIL_LOCK_UNTIL = now + 120
            _FAIL_COUNT = 0
            return 429, "试错太多次，锁 120 秒"
    return 401, "口令不对，或没带 X-Code 头"


_CARDS = {}      # cid -> card dict
_CARD_ORDER = []

# ---------------------------------------------------------------- content 加载

_MODULES = {}   # course:mid -> {"id","title",...,"lang","items","primaries"}
_ITEMS = {}     # item_id -> dict（含 lang）
    # Courses: courses/<id>/course.json 声明 {"id","lang","title"}
_COURSES = {}


def load_content():
    _MODULES.clear()
    _ITEMS.clear()
    _COURSES.clear()
    order = []
    for cdir in sorted(COURSES.glob("*")):
        cj = cdir / "course.json"
        if not cdir.is_dir() or not cj.exists():
            continue
        try:
            course = json.loads(cj.read_text(encoding="utf-8"))
        except (json.JSONDecodeError, OSError) as e:
            print(f"[CodeGym] 跳过坏课程 {cdir.name}: {e}", flush=True)
            continue
        cid_course = course["id"]
        lang = course.get("lang", "python")
        _COURSES[cid_course] = course
        course["dir"] = cdir
        for mdir in sorted(cdir.glob("m*")):
            if not mdir.is_dir():
                continue
            try:
                meta = json.loads((mdir / "module.json").read_text(encoding="utf-8"))
            except (json.JSONDecodeError, OSError) as e:
                print(f"[CodeGym] 跳过未完成模块 {cdir.name}/{mdir.name}: {e}", flush=True)
                continue
            mid = f"{cid_course}:{meta['id']}"
            meta["id"] = mid
            meta["dir"] = mdir
            meta["lang"] = lang
            meta["course"] = cid_course
            meta["items"] = []
            meta["primaries"] = []
            _MODULES[mid] = meta
            order.append(mid)
            for cf in sorted(mdir.glob("c*.json")):
                try:
                    c = json.loads(cf.read_text(encoding="utf-8"))
                except (json.JSONDecodeError, OSError) as e:
                    print(f"[CodeGym] 跳过坏内容文件 {cdir.name}/{mdir.name}/{cf.name}: {e}", flush=True)
                    continue
                c["module"] = mid
                c["lang"] = lang
                cid = c["id"]
                _ITEMS[cid] = c
                meta["items"].append(cid)
                meta["primaries"].append(cid)
                for i, v in enumerate(c.get("variants", [])):
                    vid = f"{cid}#v{i + 1}"
                    vi = dict(c)
                    vi.update(v)
                    vi["id"] = vid
                    vi["parent"] = cid
                    vi["is_variant"] = True
                    _ITEMS[vid] = vi
                    meta["items"].append(vid)
    _MODULES["__order__"] = order


def load_cards():
    f = ROOT / "content" / "cards" / "cards.json"
    if not f.exists():
        return
    _CARDS.clear()
    data = json.loads(f.read_text(encoding="utf-8"))
    for c in data["cards"]:
        _CARDS[c["id"]] = c
    globals()["_CARD_ORDER"] = [c["id"] for c in data["cards"]]


def db():
    if not hasattr(_TLS, "conn"):
        _TLS.conn = sqlite3.connect(ROOT / "progress.db")
        _TLS.conn.row_factory = sqlite3.Row
        _TLS.conn.executescript(
            """CREATE TABLE IF NOT EXISTS progress(
                 item_id TEXT PRIMARY KEY, attempts INT DEFAULT 0,
                 hints_used INT DEFAULT 0, ref_viewed INT DEFAULT 0,
                 cold_pass INT DEFAULT 0, last_attempt_at REAL, cold_pass_at REAL);
               CREATE TABLE IF NOT EXISTS retention(
                 item_id TEXT, stage INT, due_at REAL, done INT DEFAULT 0,
                 PRIMARY KEY(item_id, stage));
               CREATE TABLE IF NOT EXISTS gates(
                 mid TEXT PRIMARY KEY, passed INT DEFAULT 0, passed_at REAL,
                 pending TEXT);
               CREATE TABLE IF NOT EXISTS cards_state(
                 cid TEXT PRIMARY KEY, box INT DEFAULT 1, due_at REAL DEFAULT 0,
                 last_at REAL, reviews INT DEFAULT 0);
               CREATE TABLE IF NOT EXISTS attempts(
                 id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, item_id TEXT,
                 ok INT, cold INT, ms REAL, detail TEXT);""")
        _TLS.conn.commit()
    return _TLS.conn


# ---------------------------------------------------------------- 卡片（Leitner 盒子）

BOX_STEPS = RETENTION_STAGES  # [1,3,7,14,30] 天


def _card_row(cid):
    c = db().execute("SELECT * FROM cards_state WHERE cid=?", (cid,)).fetchone()
    if c is None:
        db().execute("INSERT INTO cards_state(cid,box,due_at) VALUES(?,1,0)", (cid,))
        db().commit()
        return db().execute("SELECT * FROM cards_state WHERE cid=?", (cid,)).fetchone()
    return c


def card_grade(cid, remembered):
    """remembered: True 记得 / None 模糊 / False 忘了"""
    r = _card_row(cid)
    box = r["box"]
    if remembered is True:
        box = min(len(BOX_STEPS), box + 1)
    elif remembered is False:
        box = 1
    due = time.time() + BOX_STEPS[box - 1] * 86400
    db().execute("""UPDATE cards_state SET box=?, due_at=?, last_at=?, reviews=reviews+1 WHERE cid=?""",
                 (box, due, time.time(), cid))
    db().commit()
    return {"box": box, "due_days": BOX_STEPS[box - 1],
            "mastered": box >= len(BOX_STEPS) and remembered is True}


def card_code_submit(cid, code):
    """算法卡默写：跑卡片自带 assert，通过=自动记'记得'。"""
    card = _CARDS.get(cid)
    if not card or not card.get("asserts"):
        return {"ok": False, "error": "这张卡没有可判题的代码"}
    with tempfile.TemporaryDirectory(prefix="cg-card-") as td:
        td = Path(td)
        (td / "solution.py").write_text(code, encoding="utf-8")
        tests = [{"name": f"assert #{i+1}", "code": a} for i, a in enumerate(card["asserts"])]
        (td / "_tests.json").write_text(json.dumps({"tests": tests}, ensure_ascii=False), encoding="utf-8")
        (td / "_grade.py").write_text(_GRADE_FUNC, encoding="utf-8")
        try:
            proc = subprocess.run([sys.executable, "_grade.py"], cwd=td,
                                  capture_output=True, text=True, timeout=TIMEOUT_S)
        except subprocess.TimeoutExpired:
            return {"ok": False, "fail": 1, "pass": 0,
                    "error": "超时：10 秒没跑完（检查死循环）"}
    out = proc.stdout.strip()
    try:
        data = json.loads(out.splitlines()[-1]) if out else {}
    except (json.JSONDecodeError, IndexError):
        return {"ok": False, "error": (proc.stderr or "判题异常")[-300:]}
    if "load_error" in data:
        res = {"ok": False, "failed_count": len(tests), "passed_count": 0, "error": data["load_error"]}
        return res
    npass = sum(1 for x in data["results"] if x["ok"])
    nfail = len(data["results"]) - npass
    ok = nfail == 0
    res = {"ok": ok, "pass": npass, "fail": nfail, "results": data["results"]}
    if ok:
        g = card_grade(cid, True)
        res["auto"] = f"默写通过，已记入盒子 {g['box']}（{g['due_days']} 天后再见）"
    return res


def cards_summary():
    now = time.time()
    decks = {}
    for cid in _CARD_ORDER:
        c = _CARDS[cid]
        d = decks.setdefault(c["deck_cn"], {"deck": c["deck"], "deck_cn": c["deck_cn"],
                                            "total": 0, "due": 0, "mastered": 0})
        d["total"] += 1
        r = db().execute("SELECT * FROM cards_state WHERE cid=?", (cid,)).fetchone()
        if r is None:
            d["due"] += 1  # 没学过=待初记
        else:
            if r["due_at"] <= now:
                d["due"] += 1
            if r["box"] >= len(BOX_STEPS):
                d["mastered"] += 1
    return [decks[k] for k in sorted(decks)]


# ---------------------------------------------------------------- 判题

_GRADE_FUNC = r'''
import json, sys
spec = json.load(open("_tests.json", encoding="utf-8"))
ns = {}
try:
    src = open("solution.py", encoding="utf-8").read()
    exec(compile(src, "solution.py", "exec"), ns)
except SystemExit:
    print(json.dumps({"load_error": "代码里调用了 exit/quit——提交的应是定义，不该直接退出"})); raise
except BaseException as e:
    print(json.dumps({"load_error": "%s: %s" % (type(e).__name__, e)})); raise
results = []
for i, t in enumerate(spec["tests"]):
    try:
        exec(compile(t["code"], "<test%d>" % i, "exec"), ns)
        results.append({"i": i, "name": t["name"], "ok": True})
    except AssertionError:
        results.append({"i": i, "name": t["name"], "ok": False, "err": "断言不成立"})
    except BaseException as e:
        results.append({"i": i, "name": t["name"], "ok": False,
                        "err": "%s: %s" % (type(e).__name__, e)})
print(json.dumps({"results": results}, ensure_ascii=False))
'''


def _norm(s):
    return "\n".join(line.rstrip() for line in s.strip().splitlines())


def _adapter(lang):
    """语言适配器注册表：新语言 = adapters/ 下加一个 *_adapter.py + 这里登记。"""
    mods = {"python": "python_adapter", "javascript": "javascript_adapter"}
    return importlib.import_module(f"adapters.{mods.get(lang, 'python')}")


def grade(item, code):
    """跑用户代码，返回统一结构 {ok, results:[{name,ok,err}], load_error, stdout}。"""
    try:
        return _grade_inner(item, code)
    except subprocess.TimeoutExpired:
        return {"ok": False,
                "load_error": "超时：10 秒没跑完（八成是死循环，检查 while 的退出条件）",
                "ms": TIMEOUT_S * 1000}


def _grade_inner(item, code):
    a = _adapter(item.get("lang", "python"))
    if item["mode"] == "func":
        res = a.grade_func(item, code)
    else:
        res = a.grade_io(item, code)
    # 适配器超时/解析失败形状归一成 load_error，前端只认这个
    if res.get("error") and not res.get("results"):
        res = {"ok": False, "load_error": res["error"], "ms": TIMEOUT_S * 1000}
    return res


# ---------------------------------------------------------------- 渐进披露与进度

def _row(item_id):
    c = db().execute("SELECT * FROM progress WHERE item_id=?", (item_id,)).fetchone()
    if c is None:
        db().execute("INSERT INTO progress(item_id) VALUES(?)", (item_id,))
        db().commit()
        return db().execute("SELECT * FROM progress WHERE item_id=?", (item_id,)).fetchone()
    return c


def submit(item_id, code):
    item = _ITEMS[item_id]
    res = grade(item, code)
    r = _row(item_id)
    attempts = r["attempts"] + 1
    cold = int(res["ok"] and r["hints_used"] == 0 and r["ref_viewed"] == 0)
    db().execute("""UPDATE progress SET attempts=?, last_attempt_at=?, cold_pass=MAX(cold_pass,?),
                    cold_pass_at=CASE WHEN ?=1 AND cold_pass_at IS NULL THEN ? ELSE cold_pass_at END
                    WHERE item_id=?""",
                 (attempts, time.time(), cold, cold, time.time(), item_id))
    db().execute("INSERT INTO attempts(ts,item_id,ok,cold,ms,detail) VALUES(?,?,?,?,?,?)",
                 (time.time(), item_id, int(res["ok"]), cold, res["ms"],
                  json.dumps({"fails": [x["name"] for x in res.get("results", []) if not x["ok"]]},
                             ensure_ascii=False)))
    db().commit()
    if cold:
        if not item.get("is_variant"):
            _schedule_retention(item_id)
        _maybe_gate_progress(item_id)
    # 渐进披露
    res["attempts"] = attempts
    res["cold"] = bool(cold)
    res["hints_unlockable"] = attempts >= 4
    res["ref_unlockable"] = attempts >= 6 or r["hints_used"] >= len(item.get("hints", []))
    if attempts == 1:
        for x in res.get("results", []):
            x.pop("err", None)
    return res


def take_hint(item_id):
    item = _ITEMS[item_id]
    r = _row(item_id)
    hints = item.get("hints", [])
    if r["hints_used"] >= len(hints):
        return {"hint": None, "exhausted": True}
    if r["attempts"] < 4:
        return {"hint": None, "locked": True, "need_attempts": 4 - r["attempts"]}
    h = hints[r["hints_used"]]
    db().execute("UPDATE progress SET hints_used=? WHERE item_id=?", (r["hints_used"] + 1, item_id))
    db().commit()
    return {"hint": h, "index": r["hints_used"] + 1, "total": len(hints)}


def get_ref(item_id):
    item = _ITEMS[item_id]
    r = _row(item_id)
    if r["attempts"] < 6 and r["hints_used"] < len(item.get("hints", [])):
        return {"ref": None, "locked": True}
    if not r["ref_viewed"]:
        db().execute("UPDATE progress SET ref_viewed=1 WHERE item_id=?", (item_id,))
        db().commit()
    return {"ref": item["ref"], "note": "看了参考解：本题永远不算冷启动通过。把窗口关了，自己重写一遍——写出来才算。"
            if not r["cold_pass"] else "你已冷启动通过，参考解仅供对照。"}


def _schedule_retention(item_id):
    now = time.time()
    for stage in RETENTION_STAGES:
        db().execute("""INSERT OR REPLACE INTO retention(item_id,stage,due_at,done)
                        VALUES(?,?,?,COALESCE((SELECT done FROM retention WHERE item_id=? AND stage=?),0))""",
                     (item_id, stage, now + stage * 86400, item_id, stage))
    db().commit()


def _retention_done(item_id, stage):
    db().execute("UPDATE retention SET done=1 WHERE item_id=? AND stage=?", (item_id, stage))
    db().commit()


def retention_due():
    now = time.time()
    rows = db().execute(
        "SELECT item_id, stage, due_at FROM retention WHERE done=0 AND due_at<=? ORDER BY due_at", (now,)).fetchall()
    out = []
    for r in rows:
        variants = [i for i in _ITEMS.get(r["item_id"], {}).get("variants", [])]
        practice = (f"{r['item_id']}#v1" if variants else r["item_id"])
        out.append({"item_id": r["item_id"], "stage": r["stage"],
                    "practice_id": practice,
                    "overdue_days": round((now - r["due_at"]) / 86400, 1)})
    return out


def _maybe_gate_progress(item_id):
    """item 冷启动通过时，检查是否有模块门考等待这两题。"""
    for g in db().execute("SELECT mid, pending FROM gates WHERE passed=0 AND pending IS NOT NULL").fetchall():
        pending = json.loads(g["pending"])
        if item_id in pending:
            pending = [p for p in pending if p != item_id]
            if not pending:
                db().execute("UPDATE gates SET passed=1, passed_at=?, pending=NULL WHERE mid=?",
                             (time.time(), g["mid"]))
            else:
                db().execute("UPDATE gates SET pending=? WHERE mid=?", (json.dumps(pending), g["mid"]))
            db().commit()


def module_state(mid):
    m = _MODULES[mid]
    items = m["primaries"]
    n_cold = sum(1 for i in items if _row(i)["cold_pass"])
    gate = db().execute("SELECT * FROM gates WHERE mid=?", (mid,)).fetchone()
    lessons = m.get("lessons", [])
    gate_ready = n_cold == len(items)
    state = "locked"
    if mid == _MODULES["__order__"][0]:
        state = "open"
    prev = None
    for i, x in enumerate(_MODULES["__order__"]):
        if x == mid:
            prev = _MODULES["__order__"][i - 1] if i else None
    if prev:
        pg = db().execute("SELECT passed FROM gates WHERE mid=?", (prev,)).fetchone()
        if prev == "m11":
            pass
        if pg and pg["passed"]:
            state = "open"
        state = "open"
    if gate and gate["passed"]:
        state = "done"
    elif gate and gate["pending"]:
        state = "gating"
    elif gate_ready:
        state = "gate_ready"
    return {"id": mid, "title": m["title"], "desc": m.get("desc", ""), "state": state,
            "n_items": len(items), "n_cold": n_cold,
            "lessons": [{"n": i + 1, "title": l["title"]} for i, l in enumerate(lessons)]}


def _gate_passed(mid):
    g = db().execute("SELECT passed FROM gates WHERE mid=?", (mid,)).fetchone()
    return bool(g and g["passed"])


def start_gate(mid):
    import random
    if not all(_row(i)["cold_pass"] for i in _MODULES[mid]["primaries"]):
        return {"error": "还有挑战没冷启动通过，门考不开。"}
    pool = [i for i in _MODULES[mid]["items"] if "#" in i]
    if len(pool) < 2:
        return {"error": "本模块变体不足，内容问题，联系管理员。"}
    picks = random.sample(pool, 2)
    db().execute("""INSERT INTO gates(mid,passed,pending) VALUES(?,0,?)
                    ON CONFLICT(mid) DO UPDATE SET pending=excluded.pending, passed=0""",
                 (mid, json.dumps(picks)))
    db().commit()
    return {"gate_items": picks, "rule": "两题一次会话内零提示冷启动通过，即过模块门。"}

# ---------------------------------------------------------------- HTTP

class Handler(BaseHTTPRequestHandler):
    server_version = "CodeGym/1.0"

    def log_message(self, fmt, *args):
        """访问日志：默认静音，设 CODEGYM_DEBUG=1 时打 stderr。"""
        import os
        if os.environ.get("CODEGYM_DEBUG") or os.environ.get("PYGYM_DEBUG"):
            sys.stderr.write("[访问] %s - %s\n" % (self.address_string(), fmt % args))

    def _json(self, code, obj):
        body = json.dumps(obj, ensure_ascii=False).encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _static(self, path):
        if path in ("/", "/index.html"):
            fp, ct = ROOT / "web" / "index.html", "text/html; charset=utf-8"
        else:
            return self._json(404, {"error": "not found"})
        body = fp.read_bytes()
        self.send_response(200)
        self.send_header("Content-Type", ct)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-cache")
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        try:
            self._route_get()
        except Exception as e:  # noqa: BLE001
            self._json(500, {"error": f"{type(e).__name__}: {e}"})

    def _route_get(self):
        p, _, qs = self.path.partition("?")
        self._qs = urllib.parse.parse_qs(qs)
        p = urllib.parse.unquote(p)
        if p == "/api/health":
            n = len(_ITEMS)
            return self._json(200, {"ok": True, "service": "codegym",
                                    "modules": len(_MODULES.get("__order__", [])), "items": n,
                                    "gate": bool(_GATE_CODE)})
        if p.startswith("/api"):
            gate = _gate_check(self.headers)
            if gate:
                code, msg = gate
                return self._json(code, {"error": msg, "code": "GATE_REJECTED"})
        if p == "/api/outline":
            return self._json(200, {
                "modules": [module_state(m) for m in _MODULES["__order__"]],
                "retention": retention_due(),
                "graduated": False})
        if p.startswith("/api/lesson/"):
            parts = p.split("/")
            mid, n = parts[3], parts[4]
            mid = mid if mid in _MODULES else ":".join([_COURSES and list(_COURSES)[0], mid])
            m = _MODULES[mid]
            lesson = m["lessons"][int(n) - 1]
            text = (m["dir"] / lesson["file"]).read_text(encoding="utf-8")
            return self._json(200, {"title": lesson["title"], "md": text})
        if p.startswith("/api/item/"):
            iid = p.split("/")[-1]
            it = _ITEMS[iid]
            r = _row(iid)
            return self._json(200, {
                "id": iid, "title": it["title"], "skill": it.get("skill", ""),
                "mode": it["mode"], "spec": it["spec"], "starter": it.get("starter", ""),
                "attempts": r["attempts"], "hints_used": r["hints_used"],
                "cold_pass": bool(r["cold_pass"]),
                "n_tests": len(it.get("tests") or it.get("tests_io", [])),
                "n_hints": len(it.get("hints", []))})
        if p == "/api/cards/decks":
            return self._json(200, {"decks": cards_summary()})
        if p == "/api/cards/due":
            now = time.time()
            deck = (self._qs.get("deck") or [""])[0]
            ids = []
            for cid in _CARD_ORDER:
                c = _CARDS[cid]
                if deck and c["deck"] != deck:
                    continue
                r = db().execute("SELECT box,due_at FROM cards_state WHERE cid=?", (cid,)).fetchone()
                if r is None or r["due_at"] <= now:
                    ids.append(cid)
            return self._json(200, {"ids": ids[:50]})
        if p.startswith("/api/card/"):
            cid = p.split("/")[-1]
            c = _CARDS.get(cid)
            if not c:
                return self._json(404, {"error": "卡片不存在"})
            r = _card_row(cid)
            return self._json(200, {
                "id": cid, "title": c["title"], "deck_cn": c["deck_cn"],
                "one_liner": c["one_liner"], "sections": c["sections"],
                "has_code": c["has_code"], "code": c["code"], "asserts": len(c["asserts"]),
                "box": r["box"], "reviews": r["reviews"]})
        if p.startswith("/api/module/"):
            mid = p.split("/")[-1]
            if mid not in _MODULES:
                return self._json(404, {"error": "模块不存在"})
            items = []
            for iid in _MODULES[mid]["items"]:
                it = _ITEMS[iid]
                r = _row(iid)
                items.append({"id": iid, "title": it["title"], "skill": it.get("skill", ""),
                              "attempts": r["attempts"], "cold_pass": bool(r["cold_pass"]),
                              "is_variant": it.get("is_variant", False)})
            return self._json(200, {"items": items})
        if p.startswith("/api/ref/"):
            return self._json(200, get_ref(p.split("/")[-1]))
        if p == "/api/progress":
            rows = db().execute("SELECT * FROM progress").fetchall()
            due = retention_due()
            return self._json(200, {
                "n_items": len(_ITEMS),
                "n_cold": sum(1 for r in rows if r["cold_pass"]),
                "n_retention_done": db().execute("SELECT COUNT(*) c FROM retention WHERE done=1").fetchone()["c"],
                "retention_due": len(due),
                "gates": {g["mid"]: bool(g["passed"]) for g in
                          db().execute("SELECT mid,passed FROM gates").fetchall()}})
        if p.startswith("/api/hint/"):
            return self._json(200, take_hint(p.split("/")[-1]))
        return self._static(p)

    def do_POST(self):
        try:
            gate = _gate_check(self.headers)
            if gate:
                code, msg = gate
                return self._json(code, {"error": msg, "code": "GATE_REJECTED"})
            length = int(self.headers.get("Content-Length") or 0)
            payload = json.loads(self.rfile.read(length).decode("utf-8")) if length else {}
            if self.path == "/api/submit":
                iid = payload["id"]
                if iid not in _ITEMS:
                    return self._json(404, {"error": "没有这道题"})
                return self._json(200, submit(iid, payload.get("code", "")))
            if self.path == "/api/card/grade":
                g = card_grade(payload["id"], payload.get("remembered"))
                return self._json(200, g)
            if self.path == "/api/card/code":
                return self._json(200, card_code_submit(payload["id"], payload.get("code", "")))
            if self.path == "/api/scratch":
                code = payload.get("code", "")
                lang = payload.get("lang", "python")
                if not code.strip():
                    return self._json(400, {"error": "先写点代码"})
                return self._json(200, _adapter(lang).run_scratch(code))
            if self.path == "/api/retention/done":
                iid, stage = payload["item_id"], int(payload["stage"])
                variants = [i for i in _ITEMS.get(iid, {}).get("variants", [])]
                practice = f"{iid}#v1" if variants else iid
                if not _row(practice)["cold_pass"]:
                    return self._json(400, {"error": "变体还没冷启动通过，不算复习完成"})
                _retention_done(iid, stage)
                return self._json(200, {"done": True})
            if self.path.startswith("/api/gate/"):
                return self._json(200, start_gate(self.path.split("/")[-1]))
            return self._json(404, {"error": "not found"})
        except subprocess.TimeoutExpired:
            return self._json(200, {"ok": False, "load_error": "超时：10 秒没跑完（检查死循环）"})
        except Exception as e:  # noqa: BLE001
            self._json(500, {"error": f"{type(e).__name__}: {e}"})


def main():
    load_content()
    load_cards()
    _load_gate_code()
    db()
    print(f"[CodeGym] 内容: {len(_MODULES['__order__'])} 模块 / {len(_ITEMS)} 题", flush=True)
    server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
    print(f"[CodeGym] 0.0.0.0:{PORT} 就绪", flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass


if __name__ == "__main__":
    main()