#!/usr/bin/env python3 """ 課本變遊戲 — 本機後端（含 AI 代理） ================================================= - 提供靜態網頁（study-maker.html 等） - /v1/chat/completions 代理到本機 Ollama（朋友只需連到你的網站，不必自己裝模型） - /api/health 讓前端偵測「伺服器模式」與可用模型 - /api/models 列出本機模型 環境變數： STUDY_ACCESS_CODE 設定後，呼叫 AI 需要帶 X-Access-Code 標頭（建議分享時設定） STUDY_PORT 監聽埠（預設 5000，配合 Tailscale Funnel） STUDY_MAX_CONCURRENT 同時最多幾個 AI 請求（預設 2，其餘排隊） OLLAMA_URL Ollama 位址（預設 http://127.0.0.1:11434） """ import json import os import secrets import threading import time import urllib.error import urllib.request from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler ROOT = os.path.dirname(os.path.abspath(__file__)) SETS_DIR = os.path.join(ROOT, "sets") MAX_SETS = int(os.environ.get("STUDY_MAX_SETS", "500")) OLLAMA = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434").rstrip("/") ACCESS_CODE = os.environ.get("STUDY_ACCESS_CODE", "").strip() PORT = int(os.environ.get("STUDY_PORT", "5000")) MAX_CONCURRENT = int(os.environ.get("STUDY_MAX_CONCURRENT", "2")) SEM = threading.Semaphore(MAX_CONCURRENT) ACTIVE = threading.Lock() ACTIVE_N = [0] def list_models(): try: with urllib.request.urlopen(OLLAMA + "/api/tags", timeout=3) as r: data = json.load(r) return [m["name"] for m in data.get("models", [])], True except Exception: return [], False def pick(models, *needles): for n in needles: for m in models: if n in m: return m return None class Handler(SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=ROOT, **kwargs) def log_message(self, fmt, *args): print("[web] %s %s" % (self.address_string(), fmt % args), flush=True) # ---------- helpers ---------- 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 _authorized(self): if not ACCESS_CODE: return True return self.headers.get("X-Access-Code", "") == ACCESS_CODE def _relay(self, code, body, ctype="application/json; charset=utf-8"): self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) # ---------- routes ---------- def do_GET(self): path = self.path.split("?")[0] if path.startswith("/api/health"): return self.health() if path.startswith("/api/models"): models, ok = list_models() return self._json(200, {"ok": ok, "models": models}) if path.startswith("/api/sets/"): return self.get_set(path.rsplit("/", 1)[-1]) return super().do_GET() def do_POST(self): path = self.path.split("?")[0].rstrip("/") if path.endswith("/v1/chat/completions"): return self.chat() if path.endswith("/api/sets"): return self.save_set() return self._json(404, {"error": {"message": "not found"}}) # ---------- 分享題庫（短網址） ---------- def save_set(self): try: length = int(self.headers.get("Content-Length", "0")) data = json.loads(self.rfile.read(length)) bank = data.get("bank") or {} if not isinstance(bank, dict): raise ValueError("bad bank") except Exception: return self._json(400, {"error": {"message": "題庫格式錯誤"}}) os.makedirs(SETS_DIR, exist_ok=True) sid = secrets.token_urlsafe(6).replace("-", "a").replace("_", "b")[:10] record = { "id": sid, "name": str(data.get("name") or "分享題庫")[:80], "savedAt": int(time.time() * 1000), "knowledge": data.get("knowledge") or {}, "bank": bank, } try: with open(os.path.join(SETS_DIR, sid + ".json"), "w", encoding="utf-8") as f: json.dump(record, f, ensure_ascii=False) except Exception as e: return self._json(500, {"error": {"message": "儲存失敗：%s" % e}}) self._prune_sets() self._json(200, {"ok": True, "id": sid, "name": record["name"], "count": sum( len(bank.get(k, [])) for k in ("mc", "tf", "order", "short"))}) def get_set(self, sid): if not sid or not all(c.isalnum() for c in sid): return self._json(400, {"error": {"message": "無效的分享碼"}}) path = os.path.join(SETS_DIR, sid + ".json") if not os.path.isfile(path): return self._json(404, {"error": {"message": "找不到這份分享題庫（可能已過期）"}}) try: with open(path, encoding="utf-8") as f: return self._json(200, json.load(f)) except Exception as e: return self._json(500, {"error": {"message": "讀取失敗：%s" % e}}) def _prune_sets(self): try: files = [os.path.join(SETS_DIR, f) for f in os.listdir(SETS_DIR) if f.endswith(".json")] if len(files) <= MAX_SETS: return files.sort(key=os.path.getmtime) for f in files[:len(files) - MAX_SETS]: os.remove(f) except Exception: pass def health(self): models, ok = list_models() vis = pick(models, "qwen2.5vl", "minicpm-v", "llava", "vision") txt = pick(models, "qwen3:30b", "qwen3", "qwen2.5", "llama3") with ACTIVE: active = ACTIVE_N[0] self._json(200, { "ok": ok, "ollama": OLLAMA, "models": models, "visionModel": vis, "textModel": txt, "requiresCode": bool(ACCESS_CODE), "maxConcurrent": MAX_CONCURRENT, "active": active, }) def chat(self): if not self._authorized(): return self._json(401, {"error": {"message": "需要存取碼（Access code）才能使用，請向分享者查詢。"}}) try: length = int(self.headers.get("Content-Length", "0")) raw = self.rfile.read(length) json.loads(raw) # 驗證是合法 JSON except Exception: return self._json(400, {"error": {"message": "請求格式錯誤"}}) if not SEM.acquire(timeout=600): return self._json(503, {"error": {"message": "伺服器忙碌，請稍後再試"}}) with ACTIVE: ACTIVE_N[0] += 1 try: req = urllib.request.Request( OLLAMA + "/v1/chat/completions", data=raw, headers={"Content-Type": "application/json"}, ) with urllib.request.urlopen(req, timeout=1800) as r: body = r.read() ctype = r.headers.get("Content-Type", "application/json; charset=utf-8") self._relay(200, body, ctype) except urllib.error.HTTPError as e: body = e.read() self._relay(e.code, body) except Exception as e: self._json(502, {"error": {"message": "無法連接本機模型：%s" % e}}) finally: with ACTIVE: ACTIVE_N[0] -= 1 SEM.release() if __name__ == "__main__": print("=" * 60) print("🚑 課本變遊戲 — 本機伺服器") print(" 目錄 ：%s" % ROOT) print(" 監聽 ：http://127.0.0.1:%d" % PORT) print(" Ollama ：%s" % OLLAMA) print(" 存取碼 ：%s" % ("已設定 ✅" if ACCESS_CODE else "未設定（任何人可用，測試用）")) print(" 併發 ：最多 %d 個請求（其餘排隊）" % MAX_CONCURRENT) print("=" * 60) print("📄 本機開啟：http://localhost:%d/study-maker.html" % PORT) print("🌐 對外分享：https://david-ms-s1-max.tailcd3383.ts.net/study-maker.html") print(" （Tailscale Funnel 已代理 127.0.0.1:%d）" % PORT) print(" 停止：Ctrl+C") print("=" * 60) ThreadingHTTPServer(("127.0.0.1", PORT), Handler).serve_forever()