330 lines
13 KiB
Python
330 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""90CN GM 页本地托管 + 接口代理。
|
|
|
|
为什么需要这一层：
|
|
1. 客户端里那个窗口是 IE11 内核，页面要同源才能发 XHR，跨域到 18111 会被拦；
|
|
2. admin token 不应该写进页面 JS（会被任何人读到），由本进程在服务端注入。
|
|
|
|
用法：
|
|
python serve.py # 默认 127.0.0.1:18112
|
|
python serve.py --port 18112
|
|
python serve.py --runtime Z:\\90server\\Server\\runtime
|
|
|
|
页面访问：http://127.0.0.1:18112/
|
|
接口转发：/api/<path> -> http://127.0.0.1:18111/<path>
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import uuid
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from urllib import error, request
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
MIME = {
|
|
".html": "text/html; charset=utf-8",
|
|
".js": "application/javascript; charset=utf-8",
|
|
".css": "text/css; charset=utf-8",
|
|
".png": "image/png",
|
|
".jpg": "image/jpeg",
|
|
".jpeg": "image/jpeg",
|
|
".gif": "image/gif",
|
|
".webp": "image/webp",
|
|
".bmp": "image/bmp",
|
|
".svg": "image/svg+xml",
|
|
".ico": "image/x-icon",
|
|
}
|
|
|
|
CONFIG = {"admin": "127.0.0.1:18111", "token": ""}
|
|
|
|
# 关键：本机回环不能走系统/环境代理。环境里存在 http_proxy 时 urllib 会照单全收，
|
|
# 结果连 127.0.0.1:18111 都被转发到代理，报 "upstream connect failed"。
|
|
OPENER = request.build_opener(request.ProxyHandler({}))
|
|
|
|
|
|
# --------------------------------------------------------------------- 发现
|
|
|
|
def _candidate_runtimes() -> list[Path]:
|
|
out: list[Path] = []
|
|
env = os.environ.get("DNF90_RUNTIME")
|
|
if env:
|
|
out.append(Path(env))
|
|
out += [
|
|
Path(r"Z:\90server\Server\runtime"),
|
|
HERE.parent / "server" / "runtime",
|
|
HERE.parent / "runtime",
|
|
]
|
|
return out
|
|
|
|
|
|
def discover_admin() -> tuple[str, str]:
|
|
"""从 instance.json / dnfbridge.toml 里读 adminListen 与 adminToken。"""
|
|
token_env = os.environ.get("DNF90_ADMIN_TOKEN")
|
|
for root in _candidate_runtimes():
|
|
inst = root / "config" / "instance.json"
|
|
if inst.is_file():
|
|
try:
|
|
cfg = json.loads(inst.read_text(encoding="utf-8"))
|
|
srv = cfg.get("server", {})
|
|
listen = srv.get("adminListen") or CONFIG["admin"]
|
|
token = token_env or srv.get("adminToken") or ""
|
|
if token:
|
|
return listen, token
|
|
except (OSError, ValueError):
|
|
pass
|
|
toml = root / "configs" / "dnfbridge.toml"
|
|
if toml.is_file():
|
|
try:
|
|
txt = toml.read_text(encoding="utf-8", errors="ignore")
|
|
except OSError:
|
|
continue
|
|
m = re.search(r'^\s*listen\s*=\s*"([^"]+)"', txt, re.M)
|
|
listen = m.group(1) if m else CONFIG["admin"]
|
|
m = re.search(r'^\s*token\s*=\s*"([^"]+)"', txt, re.M)
|
|
token = token_env or (m.group(1) if m else "")
|
|
if token:
|
|
return listen, token
|
|
return CONFIG["admin"], token_env or ""
|
|
|
|
|
|
# --------------------------------------------------------------------- 代理
|
|
|
|
def proxy(method: str, path: str, body: bytes | None) -> tuple[int, bytes, str]:
|
|
url = "http://%s%s" % (CONFIG["admin"], path)
|
|
headers = {"X-Admin-Token": CONFIG["token"]}
|
|
if method == "POST":
|
|
headers["Content-Type"] = "application/json"
|
|
# 服务端对写操作要求确认头 + 幂等键
|
|
headers["X-Admin-Confirm"] = "gmweb"
|
|
headers["X-Admin-Idempotency-Key"] = str(uuid.uuid4())
|
|
req = request.Request(url, data=body, headers=headers, method=method)
|
|
try:
|
|
with OPENER.open(req, timeout=20) as resp:
|
|
return resp.status, resp.read(), resp.headers.get("Content-Type", "application/json")
|
|
except error.HTTPError as exc:
|
|
return exc.code, exc.read(), exc.headers.get("Content-Type", "application/json")
|
|
except error.URLError as exc:
|
|
msg = json.dumps({"success": False, "message": "服务端未响应：%s" % exc.reason},
|
|
ensure_ascii=False).encode()
|
|
return 502, msg, "application/json; charset=utf-8"
|
|
|
|
|
|
# --------------------------------------------------------------------- 换装配置
|
|
|
|
def _loadout_config_path() -> Path:
|
|
"""LoadoutTool/config.json，优先部署根目录，其次运行目录。"""
|
|
roots = [HERE.parent]
|
|
env = os.environ.get("DNF90_SERVER_ROOT")
|
|
if env:
|
|
roots.insert(0, Path(env))
|
|
for root in roots:
|
|
p = root / "LoadoutTool" / "config.json"
|
|
if p.is_file():
|
|
return p
|
|
return roots[0] / "LoadoutTool" / "config.json"
|
|
|
|
|
|
def _default_loadout_config() -> dict:
|
|
return {
|
|
"ServerUrl": "http://127.0.0.1:18111",
|
|
"AdminToken": "",
|
|
"Pages": [{"Name": "默认页", "Schemes": []}],
|
|
"ActivePage": 0,
|
|
}
|
|
|
|
|
|
def _read_loadout_config() -> dict:
|
|
p = _loadout_config_path()
|
|
try:
|
|
if p.is_file():
|
|
data = json.loads(p.read_text(encoding="utf-8-sig"))
|
|
if isinstance(data, dict):
|
|
if not isinstance(data.get("Pages"), list) or not data["Pages"]:
|
|
data["Pages"] = [{"Name": "默认页", "Schemes": []}]
|
|
data["ActivePage"] = max(0, min(int(data.get("ActivePage") or 0), len(data["Pages"]) - 1))
|
|
# AdminToken 只留在本地文件，网页端永远拿不到。
|
|
data["AdminToken"] = ""
|
|
return data
|
|
except (OSError, ValueError, TypeError):
|
|
pass
|
|
return _default_loadout_config()
|
|
|
|
|
|
def _write_loadout_config(data: dict) -> tuple[int, bytes, str]:
|
|
if not isinstance(data, dict):
|
|
return 400, b'{"ok":false,"error":"config must be an object"}', "application/json"
|
|
pages = data.get("Pages")
|
|
if not isinstance(pages, list) or not pages:
|
|
return 400, b'{"ok":false,"error":"Pages must be a non-empty list"}', "application/json"
|
|
p = _loadout_config_path()
|
|
try:
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
# 网页提交的配置里没有 token；保留磁盘上原来的 token。
|
|
existing = {}
|
|
if p.is_file():
|
|
try:
|
|
existing = json.loads(p.read_text(encoding="utf-8-sig"))
|
|
except (OSError, ValueError):
|
|
existing = {}
|
|
data = dict(data)
|
|
data["AdminToken"] = existing.get("AdminToken", "") if isinstance(existing, dict) else ""
|
|
tmp = p.with_suffix(".json.tmp")
|
|
tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
tmp.replace(p)
|
|
except OSError as exc:
|
|
return 500, json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False).encode(), "application/json"
|
|
return 200, b'{"ok":true}', "application/json"
|
|
|
|
|
|
def _loadout_hotkeys() -> dict:
|
|
"""Return the active page's configured global hotkeys for the UI hint."""
|
|
cfg = _read_loadout_config()
|
|
pages = cfg.get("Pages") or []
|
|
try:
|
|
page = pages[int(cfg.get("ActivePage") or 0)]
|
|
except (IndexError, TypeError, ValueError):
|
|
page = pages[0] if pages else {}
|
|
out = []
|
|
for scheme in (page.get("Schemes") or []):
|
|
keycode = int(scheme.get("KeyCode") or 0)
|
|
if keycode <= 0:
|
|
continue
|
|
out.append({
|
|
"name": scheme.get("Name") or "",
|
|
"character": scheme.get("CharacterName") or "",
|
|
"keycode": keycode,
|
|
"modifiers": int(scheme.get("Modifiers") or 0),
|
|
})
|
|
return {"page": page.get("Name") or "", "schemes": out}
|
|
|
|
|
|
def _loadout_characters() -> tuple[int, bytes, str]:
|
|
"""Character list for the loadout page.
|
|
|
|
The admin endpoint without a selector is bound to the logged-in account and
|
|
can be empty when the game client is still at the login screen. The
|
|
launcher's local DB API already exposes the full local character list, so
|
|
reuse it here and keep the loadout page usable before entering the game.
|
|
"""
|
|
try:
|
|
req = request.Request("http://127.0.0.1:19527/api/characters",
|
|
headers={"Accept": "application/json"})
|
|
with OPENER.open(req, timeout=8) as resp:
|
|
data = json.loads(resp.read().decode("utf-8"))
|
|
chars = data.get("data") if isinstance(data, dict) else None
|
|
out = []
|
|
for c in chars or []:
|
|
if not isinstance(c, dict):
|
|
continue
|
|
out.append({
|
|
"character_id": str(c.get("character_id") or ""),
|
|
"name": c.get("name") or "",
|
|
"level": c.get("level") or 0,
|
|
"online": bool(c.get("online")),
|
|
})
|
|
return 200, json.dumps({"ok": True, "data": out}, ensure_ascii=False).encode(), "application/json; charset=utf-8"
|
|
except (OSError, ValueError) as exc:
|
|
return 502, json.dumps({"ok": False, "error": "读取角色失败：%s" % exc},
|
|
ensure_ascii=False).encode(), "application/json; charset=utf-8"
|
|
|
|
|
|
# --------------------------------------------------------------------- HTTP
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
server_version = "90CN-GMWeb/1.0"
|
|
|
|
def log_message(self, fmt, *args): # 静默，避免刷屏
|
|
pass
|
|
|
|
def _send(self, code: int, payload: bytes, ctype: str) -> None:
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", ctype)
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.end_headers()
|
|
try:
|
|
self.wfile.write(payload)
|
|
except OSError:
|
|
pass
|
|
|
|
def _static(self, path: str) -> None:
|
|
rel = "index.html" if path in ("/", "") else path.lstrip("/")
|
|
target = (HERE / rel).resolve()
|
|
if HERE not in target.parents and target != HERE:
|
|
self._send(403, b"forbidden", "text/plain")
|
|
return
|
|
if not target.is_file():
|
|
self._send(404, b"not found", "text/plain")
|
|
return
|
|
self._send(200, target.read_bytes(), MIME.get(target.suffix.lower(), "application/octet-stream"))
|
|
|
|
def do_GET(self) -> None:
|
|
route = self.path.split("?", 1)[0]
|
|
if route == "/api/loadout/config":
|
|
payload = json.dumps({"ok": True, "data": _read_loadout_config()}, ensure_ascii=False).encode()
|
|
self._send(200, payload, "application/json; charset=utf-8")
|
|
return
|
|
if route == "/api/loadout/characters":
|
|
code, payload, ctype = _loadout_characters()
|
|
self._send(code, payload, ctype)
|
|
return
|
|
if route == "/api/loadout/hotkeys":
|
|
payload = json.dumps({"ok": True, "data": _loadout_hotkeys()}, ensure_ascii=False).encode()
|
|
self._send(200, payload, "application/json; charset=utf-8")
|
|
return
|
|
if self.path.startswith("/api/"):
|
|
code, data, ctype = proxy("GET", self.path[4:], None)
|
|
self._send(code, data, ctype)
|
|
return
|
|
self._static(route)
|
|
|
|
def do_POST(self) -> None:
|
|
route = self.path.split("?", 1)[0]
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
body = self.rfile.read(length) if length else b"{}"
|
|
if route == "/api/loadout/config":
|
|
try:
|
|
data = json.loads(body.decode("utf-8"))
|
|
except ValueError:
|
|
self._send(400, b'{"ok":false,"error":"invalid json"}', "application/json")
|
|
return
|
|
code, payload, ctype = _write_loadout_config(data)
|
|
self._send(code, payload, ctype)
|
|
return
|
|
if not self.path.startswith("/api/"):
|
|
self._send(404, b"not found", "text/plain")
|
|
return
|
|
code, data, ctype = proxy("POST", self.path[4:], body)
|
|
self._send(code, data, ctype)
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
ap = argparse.ArgumentParser(description="90CN GM 页本地托管 + 接口代理")
|
|
ap.add_argument("--port", type=int, default=18112)
|
|
ap.add_argument("--host", default="127.0.0.1")
|
|
args = ap.parse_args(argv)
|
|
|
|
listen, token = discover_admin()
|
|
CONFIG["admin"], CONFIG["token"] = listen, token
|
|
if not token:
|
|
print("警告：没找到 adminToken，接口会 401。可用 DNF90_ADMIN_TOKEN 环境变量指定。", file=sys.stderr)
|
|
|
|
srv = ThreadingHTTPServer((args.host, args.port), Handler)
|
|
print("GM 页: http://%s:%d/" % (args.host, args.port))
|
|
print("代理到: http://%s/api/ -> http://%s/" % (args.host + ":" + str(args.port), listen))
|
|
try:
|
|
srv.serve_forever()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv[1:]))
|