小助手内嵌 GM 页：网页重定向 + 物品分页 + 装备强化增幅发放
- 90CN.dll 新增 IE 控件接管，把腾讯小助手页面重定向到本地 GM 页 - gmweb：物品列表改分页（每页 10/20/50，列表固定 6 行高） - 发放走 grant-options，支持强化 0-31、增幅属性+等级、锻造 0-8、品质 - 服务端补齐 amplify_level：等级统一写入 ext_data0/packed_flag_byte/reinforce/upgrade_level - 增幅与强化互斥校验，避免写出客户端无法创建的装备
This commit is contained in:
hanfuit committed
1 parent
6e570fe820
commit
18e98ca7c9
13 files changed
+1206
-11
No files matched your search
+181
@@ -0,0 +1,181 @@
|
||||
#!/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",
|
||||
".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"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------- 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:
|
||||
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])
|
||||
|
||||
def do_POST(self) -> None:
|
||||
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)
|
||||
|
||||
|
||||
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:]))
|
||||
Reference in new issue
Block a user