gmweb:游戏内一键换装、方案配置与全局热键说明
This commit is contained in:
1 parent
3b240ec176
commit
cc546031b0
5 files changed
+586
-3
No files matched your search
+145
-3
@@ -115,6 +115,125 @@ def proxy(method: str, path: str, body: bytes | None) -> tuple[int, bytes, str]:
|
||||
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):
|
||||
@@ -146,18 +265,41 @@ class Handler(BaseHTTPRequestHandler):
|
||||
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(self.path.split("?")[0])
|
||||
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
|
||||
length = int(self.headers.get("Content-Length") or 0)
|
||||
body = self.rfile.read(length) if length else b"{}"
|
||||
code, data, ctype = proxy("POST", self.path[4:], body)
|
||||
self._send(code, data, ctype)
|
||||
|
||||
|
||||
Reference in new issue
Block a user