1026 lines
50 KiB
Python
1026 lines
50 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
Ananta 注入器 · 只做两件事：保龄球瓶 + 气球
|
||
|
||
★ 这不是「统一加载器」（AnantaUnifiedLoader）那个全通道诊断版 —— 那份还在
|
||
C:\\Game\\Ananta_mainleakflow_keitaro_gg\\AnantaUnifiedLoader，开着是为了排查问题时
|
||
临时启用 A/B/D/E 那些通道。**日常跑服只用这一份**，它只挂 bowling + balloon 两个模块。
|
||
|
||
★ 会把自己的状态写进同目录 injector_status.json，供根目录的「一键启动」界面读：
|
||
starting 正在启动
|
||
waiting_game 已启动，等游戏
|
||
waiting_world 已启动，等玩家进入游戏世界（进入世界闸门）
|
||
attached 已挂上客户端
|
||
injected 已注入并生效（bowling + balloon 的 hook 都装好了）
|
||
stopped 已退出
|
||
|
||
一个 frida 会话 + 多个 script，把调研文档里「注入端能做到」的通道一次性全开：
|
||
|
||
| 模块 | 通道 | 做什么 |
|
||
|-----------------|----------|-----------------------------------------------------|
|
||
| gadget_world | A | 室外/建筑内 gadget 网格 |
|
||
| world_mod | B | 场景物品：路灯/街边道具/可破坏物/念动力目标 |
|
||
| train | A'+D(+E) | 地铁车厢与座位；自带 loadindoor / sectorstate |
|
||
| bowling | C | 保龄球球瓶（补表类：伪造 SceneItemBlob 查表结果） |
|
||
| balloon | C | 打气球的气球本体（补表类，需服务端同机写 side-channel）|
|
||
| addbtn | — | ★ 通用 F 提示按钮（InteractBtnMgr.AddBtn） |
|
||
|
||
★ addbtn 是**不可退让**的那一项：协议层没有任何 S2C 能 AddBtn，
|
||
服务端再怎么回 Ask* 也变不出 F 提示 —— 只有注入端能补。
|
||
|
||
★ 一键启动（2026-09-29）：双击同目录的 start_game_and_loader.cmd
|
||
—— 自提权 → 开 Ananta.exe → 等进程 → 自动挂本加载器。见 使用说明.md §2.5。
|
||
|
||
用法（源码模式，需先装 frida：pip install frida frida-tools）：
|
||
python unified_loader.py 常驻（推荐；先开它再开游戏也行）
|
||
python unified_loader.py diag 加载全部模块并各跑一次状态，打印后退出
|
||
python unified_loader.py scan 只读：列出身边可补按钮的物件
|
||
python unified_loader.py --no train,balloon 临时关掉某几个模块
|
||
|
||
⛑ 硬约束
|
||
1. 同一时刻只能有一个 frida 会话注入 Ananta.exe。
|
||
开本加载器前必须关掉 GymLoader / TrainSeatLoader / BowlingLoader /
|
||
BalloonLoader / GadgetLoader / WorldMod / PropScanner —— 本程序每轮都会查，
|
||
查到就让位等待，不会硬抢。
|
||
2. 必须以**管理员**运行（游戏是提权启动的，非管理员 attach 会永久阻塞）。
|
||
3. balloon 模块依赖服务端在同机写 side-channel 文件
|
||
%LOCALAPPDATA%\\AnantaBalloon\\balloon_push.json；
|
||
主线 v0.3 目前没有 GameRouter.Balloon.cs，写不出这个文件 ⇒ 气球不会出。
|
||
需要的话可把 v2.1 包里的 GameRouter.Balloon.cs + BalloonWireContracts.cs 移植过来。
|
||
"""
|
||
import sys, os, json, time, math, atexit, glob
|
||
|
||
# ⛑⛑ 2026-09-30 修「别的电脑上注入器一启动就崩」：
|
||
# 随包分发的便携 Python（tools\python）是 embeddable 版，带 python313._pth ⇒
|
||
# 运行时是**隔离模式**：不会把「脚本所在目录」加进 sys.path，_pth 里的 "." 也是相对
|
||
# **python.exe 所在目录**（tools\python），不是本脚本所在目录。
|
||
# ⇒ `from loader_common import ...` 直接 ModuleNotFoundError ⇒ 进程秒退，
|
||
# 启动器那边表现为「每 60 秒 spawn 一次、每次立刻死」。
|
||
# 开发机上用的是普通 venv python（没有 _pth），所以一直复现不出来。
|
||
# 这里显式把本脚本目录塞进 sys.path，不管用什么 python、从哪个 cwd 启动都能导入。
|
||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||
if _HERE not in sys.path:
|
||
sys.path.insert(0, _HERE)
|
||
|
||
from loader_common import (
|
||
Ctx, Mod, _LIVE, _atexit_close, other_loader_running,
|
||
resolve_pid, print, ts, sleep_slice, STOP,
|
||
acquire_instance_lock,
|
||
)
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
CFG_PATH = os.path.join(HERE, "config.json")
|
||
# ★ 状态文件：给「一键启动」界面读的，靠它显示「已启动 / 等待注入 / 已注入」，
|
||
# 而不是让界面去瞎猜（界面看不到注入器控制台里的字）。
|
||
STATUS_PATH = os.path.join(HERE, "injector_status.json")
|
||
|
||
|
||
def write_status(state, **kw):
|
||
try:
|
||
d = {"state": state, "ts": time.time(), "pid": os.getpid()}
|
||
d.update(kw)
|
||
tmp = STATUS_PATH + ".tmp"
|
||
with open(tmp, "w", encoding="utf-8") as fh:
|
||
json.dump(d, fh, ensure_ascii=False)
|
||
os.replace(tmp, STATUS_PATH) # 原子替换，界面不会读到半截文件
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def load_cfg():
|
||
try:
|
||
return json.loads(open(CFG_PATH, encoding="utf-8").read())
|
||
except Exception as e:
|
||
print("[!] 读 config.json 失败（%s）—— 用默认全开" % e)
|
||
return {"modules": [], "addbtnRadiusMeters": 6.0, "addbtnBatch": 3,
|
||
"addbtnEverySec": 2.0, "seatRadiusMeters": 8.0, "statusPollSec": 30.0}
|
||
|
||
|
||
def build_modules(cfg, disable=None):
|
||
"""★ 2026-10-01：新增 **分组开关**（`config.json` 的 `groups`）。
|
||
|
||
每个模块用 `"group": "xxx"` 声明组别；groups 里该组为 false 就整组不挂，
|
||
没有声明 group 的模块不受影响（保持老行为）。
|
||
⛔ traffic 组（vehicle_go.js）2026-10-01 晚已删除：它 hook 的
|
||
`AetherVehicleViewManager` 实机 8 探针全 0 命中、类根本没在用
|
||
⇒ 车流改走服务端 trafficCruiseDrive，别再往这条路上投时间。
|
||
⛑ 与既有两条过滤**叠加**（不是替换）：模块自身 `enabled=false` 仍然优先，
|
||
命令行 `--no a,b` 也仍然有效。
|
||
⛑ 没写 group 的模块不受分组影响（保持老行为）。
|
||
"""
|
||
groups = cfg.get("groups") or {}
|
||
out = []
|
||
for m in cfg.get("modules") or []:
|
||
name = m.get("name")
|
||
if disable and name in disable:
|
||
print(" · %s（本次 --no 已关闭）" % name)
|
||
continue
|
||
if not m.get("enabled", True):
|
||
print(" · %s（config.json 里 enabled=false）" % name)
|
||
continue
|
||
g = m.get("group")
|
||
if g and groups and not groups.get(g, True):
|
||
print(" · %s（groups.%s=false，本次不挂）" % (name, g))
|
||
continue
|
||
out.append(Mod(name, os.path.join(HERE, m.get("file"))))
|
||
return out
|
||
|
||
|
||
def fmt_status(name, rpc, raw):
|
||
"""把状态 RPC 的返回压成一行人能看的日志；解析不了就原样截断。"""
|
||
s = str(raw)
|
||
try:
|
||
d = json.loads(s)
|
||
except Exception:
|
||
return s[:200]
|
||
if name == "gadget_loader" and rpc == "status":
|
||
w = d.get("world") or {}
|
||
if d.get("warmupLeft"):
|
||
return "预热中，%d 秒后开始（gadgets=%s）" % (d.get("warmupLeft"), w.get("GadgetsInWorld"))
|
||
return ("cell=%s raid=%s loaded=%d(confirmed=%d) pend=%d giveup=%d ok=%d fail=%d "
|
||
"gadgets=%s speed=%.2f%s burst=%d retry=%d refresh=%d 到达救援=%d 已走=%.1f格 全扫=%d 静扫=%d"
|
||
% (d.get("cell"), d.get("raid"), d.get("loadedCells", 0), d.get("confirmed", 0),
|
||
d.get("pendingCells", 0), d.get("giveUpCells", 0), d.get("callsOk", 0),
|
||
d.get("callsFail", 0), w.get("GadgetsInWorld"), d.get("speed", 0),
|
||
" FAST" if d.get("fast") else "", d.get("burst", 0),
|
||
d.get("retryTotal", 0), d.get("refreshTotal", 0),
|
||
d.get("arriveCount", 0), d.get("movedCells", 0),
|
||
d.get("sweepAllCount", 0), d.get("idleSweepCount", 0)))
|
||
return s[:200]
|
||
|
||
|
||
def load_cell_sectors(path):
|
||
"""cells.json: {"cells": [[x, z, sectorId, ...], ...]} → {"x,z": [sectorId...]}
|
||
|
||
照抄 AnantaGadgetLoader/src/gadget_world.py:load_cell_sectors。
|
||
一个格子可能属于多个 sector —— 室内 Gadget 通常不在 sector 0。
|
||
"""
|
||
try:
|
||
data = json.loads(open(path, encoding="utf-8").read())
|
||
except Exception as e:
|
||
return {}, "读不了 %s（%s）" % (path, e)
|
||
rows = data.get("cells") if isinstance(data, dict) else data
|
||
if not rows:
|
||
return {}, "文件里没有 cells"
|
||
m = {}
|
||
for c in rows:
|
||
if not isinstance(c, (list, tuple)) or len(c) < 3:
|
||
continue
|
||
try:
|
||
m["%d,%d" % (int(c[0]), int(c[1]))] = [int(v) for v in c[2:]]
|
||
except Exception:
|
||
continue
|
||
return m, "%d 格" % len(m)
|
||
|
||
|
||
def run_setup(ctx, cfg):
|
||
"""按 config.json 的 setup 数组，把各模块的**启动序列**跑一遍。
|
||
|
||
★ 为什么必须有这一步：
|
||
原版加载器是 python 驱动的 —— agent 加载进去之后还要
|
||
setr/probe → setsectors → sectormode → tune → start 才真正开始加载。
|
||
只 load 不 start = 什么都不会发生（本加载器第一版就是这个症状）。
|
||
"""
|
||
for m in cfg.get("modules") or []:
|
||
mod = ctx.mod(m.get("name"))
|
||
if mod is None:
|
||
continue
|
||
for step in m.get("setup") or []:
|
||
rpc = step.get("rpc")
|
||
if not rpc:
|
||
continue
|
||
arg = step.get("arg")
|
||
if step.get("cellsFile"):
|
||
mapping, note = load_cell_sectors(os.path.join(HERE, step["cellsFile"]))
|
||
if not mapping:
|
||
print("[%s] · %s.%s 跳过（%s）" % (ts(), m.get("name"), rpc, note))
|
||
continue
|
||
arg = json.dumps(mapping)
|
||
try:
|
||
r = mod.call(rpc, arg, timeout=step.get("timeout", 30))
|
||
s = str(r)
|
||
print("[%s] · %s.%s => %s" % (ts(), m.get("name"), rpc,
|
||
s if len(s) <= 220 else s[:220] + "…"))
|
||
except Exception as e:
|
||
print("[%s] [!] %s.%s 失败：%s" % (ts(), m.get("name"), rpc, str(e)[:160]))
|
||
|
||
|
||
def apply_addbtn_cfg(ctx, cfg):
|
||
"""addbtn 的开关是**脚本内状态**，每次 attach 后要显式下发一次。
|
||
|
||
（每个 rpc 调用都是一次新的 attach + 新的 script realm，所以不写进
|
||
addbtn.js 的默认值，而是由 Python 这边按 config.json 下发。）
|
||
"""
|
||
m = ctx.mod("addbtn")
|
||
if m is None:
|
||
return
|
||
try:
|
||
m.call("abUseObjects", "true" if cfg.get("addbtnUseObjects") else "false", timeout=8)
|
||
except Exception as e:
|
||
print("[%s] [!] abUseObjects 下发失败：%s" % (ts(), str(e)[:120]))
|
||
return
|
||
try:
|
||
m.call("abObjRadius", str(cfg.get("addbtnObjectRadius", 5.0)), timeout=8)
|
||
except Exception as e:
|
||
print("[%s] [!] abObjRadius 下发失败：%s" % (ts(), str(e)[:120]))
|
||
|
||
|
||
def once_status(ctx, mod, names):
|
||
for n in names or []:
|
||
try:
|
||
r = mod.call(n, timeout=8)
|
||
return n, str(r)[:600]
|
||
except Exception:
|
||
continue
|
||
return None, None
|
||
|
||
|
||
def _newest_server_log(cfg):
|
||
"""服务端 console 日志里最新那份（每局一份 console-<时间戳>.log）。"""
|
||
g = cfg.get("attachGate") or {}
|
||
here = os.path.dirname(os.path.abspath(__file__))
|
||
d = os.path.normpath(os.path.join(here, g.get("serverLogDir", "../../logs")))
|
||
try:
|
||
files = glob.glob(os.path.join(d, g.get("serverLogGlob", "console-*.log")))
|
||
except Exception:
|
||
return None
|
||
return max(files, key=lambda p: os.path.getmtime(p)) if files else None
|
||
|
||
|
||
def gate_ok(cfg, st):
|
||
"""★ 进入世界闸门：返回 True 才允许 attach。
|
||
|
||
为什么需要（2026-09-29）：bowling_agent.js / balloon_fix.js 里装 hook 的那段
|
||
`Il2Cpp.perform(...)` **没有重试** —— 类拿不到就 `sendlog('ERR: ... not found'); return;`
|
||
永久放弃，直到重新 attach 才会再试。所以「游戏刚起就挂」是有真实风险的。
|
||
这里先等服务端日志出现「只有 world-ready 之后才会打」的行，再 attach，
|
||
等价于复现手动操作时「进世界 → 再开加载器」这条已验证的顺序。
|
||
|
||
⛑ 只扫**本次开始跟踪之后新增的字节** —— 否则会把上一局的历史内容当成「已进世界」。
|
||
"""
|
||
g = cfg.get("attachGate") or {}
|
||
mode = str(g.get("mode") or "serverLog").lower()
|
||
poll = float(g.get("pollSec", 2.0))
|
||
timeout = float(g.get("timeoutSec", 900))
|
||
# ★ 2026-09-30：命中 marker 后还要再等「爆发期」结束（delayAfterMarkerSec）。
|
||
# 命中瞬间正是客户端「进世界黑屏加载」的最热阶段：此刻 attach
|
||
# （V8 编译 ~440KB 两个脚本 + Il2Cpp 域遍历 + 在物件加载热路径上连挂 5 个 hook）
|
||
# 会把客户端钉死在黑屏上 —— 实测 14:51:27 命中、14:51:31 客户端日志停在
|
||
# 角色动作加载段，直到注入器断开（armedGuard 重挂）客户端才放行进世界。
|
||
# 等爆发期过了再挂：大世界已加载完、热路径流量低，装上无感（13:14 局实测如此）。
|
||
# 代价只是进世界后头 delay 秒里没有注入器 —— 反正走到保龄球馆/气球区也要时间。
|
||
delay = float(g.get("delayAfterMarkerSec", 0) or 0)
|
||
now = time.time()
|
||
|
||
pid = resolve_pid()
|
||
if pid is None:
|
||
# 换局了 / 游戏没开 ⇒ 闸门状态整个作废
|
||
if st.get("passed"):
|
||
st["passed"] = False
|
||
st["file"] = None
|
||
st["hit_t"] = None
|
||
if now - st.get("last", 0.0) >= 15.0:
|
||
st["last"] = now
|
||
print("[%s] [闸门] 等游戏启动（没找到 Ananta.exe）……" % ts())
|
||
return False
|
||
|
||
if st.get("pid") != pid: # 客户端重启 ⇒ 重新等一次
|
||
st["pid"] = pid
|
||
st["passed"] = False
|
||
st["file"] = None
|
||
st["hit_t"] = None
|
||
if st.get("passed"):
|
||
return True # 这一局已经进过世界了，重挂不用再等
|
||
# ⛑ mode 上面已经 lower() 过了，这里必须跟小写比 —— 写成 "serverLog" 会恒不相等
|
||
# ⇒ 闸门形同虚设（2026-09-29 单元测抓到）。
|
||
if mode != "serverlog":
|
||
return True
|
||
|
||
path = _newest_server_log(cfg)
|
||
if path is None:
|
||
if now - st.get("last", 0.0) >= 15.0:
|
||
st["last"] = now
|
||
print("[%s] [闸门] 还没看到服务端 console 日志，继续等（服务端起了吗？）" % ts())
|
||
return False
|
||
|
||
if st.get("file") != path:
|
||
st["file"] = path
|
||
try:
|
||
st["off"] = os.path.getsize(path)
|
||
except Exception:
|
||
st["off"] = 0
|
||
st["t0"] = now
|
||
print("[%s] [闸门] 跟踪 %s（从第 %d 字节之后开始看）"
|
||
% (ts(), os.path.basename(path), st["off"]))
|
||
|
||
try:
|
||
sz = os.path.getsize(path)
|
||
except Exception:
|
||
return False
|
||
off = int(st.get("off") or 0)
|
||
if sz > off:
|
||
try:
|
||
with open(path, "rb") as fh:
|
||
fh.seek(off)
|
||
chunk = fh.read(min(sz - off, 4 * 1024 * 1024))
|
||
except Exception:
|
||
chunk = b""
|
||
st["off"] = off + len(chunk)
|
||
txt = chunk.decode("utf-8", errors="ignore")
|
||
for mk in (g.get("markers") or []):
|
||
if mk and mk in txt:
|
||
# ★ 2026-09-30：命中后**不立刻放行** —— 先记 hit_t，等爆发期过去
|
||
# （见函数头上 delayAfterMarkerSec 的说明）。
|
||
if st.get("hit_t") is None:
|
||
st["hit_t"] = now
|
||
print("[%s] [闸门] 命中 %r —— 客户端还在进世界，等 %.0fs 爆发期过去再注入"
|
||
% (ts(), mk, delay))
|
||
break
|
||
hit_t = st.get("hit_t")
|
||
if hit_t is not None and now - hit_t >= delay:
|
||
st["hit_t"] = None
|
||
st["passed"] = True
|
||
print("[%s] [闸门] 爆发期已过（+%.0fs）—— 玩家已进世界，开始注入" % (ts(), now - hit_t))
|
||
return True
|
||
|
||
if st.get("t0") is None:
|
||
st["t0"] = now
|
||
waited = now - st["t0"]
|
||
if now - st.get("last", 0.0) >= 15.0:
|
||
st["last"] = now
|
||
print("[%s] [闸门] 等玩家进入世界…… 已等 %ds（%ds 后按 fallback=%s 处理）"
|
||
% (ts(), int(waited), int(timeout), g.get("fallback", "attachAnyway")))
|
||
if waited >= timeout:
|
||
print("[%s] [闸门] 超时 —— fallback=%s" % (ts(), g.get("fallback", "attachAnyway")))
|
||
st["passed"] = True # 别每秒重试，直接放行
|
||
return str(g.get("fallback", "attachAnyway")).lower() != "giveup"
|
||
return False
|
||
|
||
|
||
def check_armed(ctx, cfg, st):
|
||
"""★ armed 守卫：attach 之后确认 hook 真装上了。
|
||
|
||
闸门只保证「时机对」，不保证脚本里那次 `bfClass(...)` 成功。
|
||
两个模块的 bfStatus 都带 armed 字段：
|
||
armed=true → 正常（tpl=false 是正常的，那只是还没拿到模板，**不要**因此重挂）
|
||
armed=false → 脚本里 return 掉了 ⇒ 到点主动重挂
|
||
"""
|
||
ag = cfg.get("armedGuard") or {}
|
||
if not ag.get("enabled", True):
|
||
return "ok"
|
||
now = time.time()
|
||
if st.get("t0") is None:
|
||
st["t0"] = now
|
||
st["next"] = 0.0
|
||
if now < float(st.get("next") or 0.0):
|
||
return st.get("lastRes", "wait")
|
||
st["next"] = now + float(ag.get("recheckEverySec", 10))
|
||
|
||
bad = []
|
||
for m in (cfg.get("modules") or []):
|
||
mod = ctx.mod(m.get("name"))
|
||
if mod is None:
|
||
continue
|
||
try:
|
||
d = json.loads(str(mod.call("bfStatus", timeout=8)))
|
||
except Exception:
|
||
bad.append((m.get("name"), "bfStatus 拿不到"))
|
||
continue
|
||
if not d.get("armed"):
|
||
bad.append((m.get("name"), "armed=false"))
|
||
if not bad:
|
||
st["lastRes"] = "ok"
|
||
return "ok"
|
||
waited = now - st["t0"]
|
||
print("[%s] [守卫] hook 还没装上（%s）—— 已等 %ds"
|
||
% (ts(), ", ".join("%s:%s" % b for b in bad), int(waited)))
|
||
st["lastRes"] = "reattach" if waited >= float(ag.get("maxWaitSec", 60)) else "wait"
|
||
return st["lastRes"]
|
||
|
||
|
||
def run_diag(disable=None):
|
||
cfg = load_cfg()
|
||
mods = build_modules(cfg, disable)
|
||
o = other_loader_running()
|
||
if o:
|
||
print("[!] 另一个加载器在跑（%s）—— 请先关掉" % o)
|
||
return 1
|
||
if resolve_pid() is None:
|
||
print("[!] 没找到 Ananta.exe（游戏没开，或你不是管理员看不到它）")
|
||
return 1
|
||
ctx = Ctx(mods)
|
||
_LIVE.append(ctx)
|
||
if not ctx.attach():
|
||
print("[!] 连接/加载失败")
|
||
return 1
|
||
print("\n" + "=" * 60)
|
||
for m in cfg.get("modules") or []:
|
||
mod = ctx.mod(m.get("name"))
|
||
if mod is None:
|
||
continue
|
||
print("\n--- %s：%s" % (m.get("name"), m.get("desc", "")))
|
||
n, r = once_status(ctx, mod, m.get("statusRpc"))
|
||
if r is None:
|
||
print(" （无状态 RPC / 全部不可用；模块自驱动，看游戏里效果）")
|
||
else:
|
||
print(" %s => %s" % (n, r))
|
||
print("\n" + "=" * 60)
|
||
ctx.close()
|
||
return 0
|
||
|
||
|
||
def run_scan(disable=None):
|
||
cfg = load_cfg()
|
||
mods = build_modules(cfg, disable)
|
||
ctx = Ctx(mods)
|
||
_LIVE.append(ctx)
|
||
if not ctx.attach():
|
||
print("[!] 连接失败")
|
||
return 1
|
||
m = ctx.mod("addbtn")
|
||
if m is None:
|
||
print("[!] addbtn 模块没加载")
|
||
return 1
|
||
try:
|
||
r = json.loads(m.call("abScan", "8", timeout=20))
|
||
print("身边候选 %d 个（已有按钮 %d 个）：" % (r.get("count", 0), r.get("existing", 0)))
|
||
for row in r.get("rows") or []:
|
||
print(" %-46s hasBtn=%s d=%sm" % (str(row.get("n"))[:46], row.get("hasBtn"), row.get("d")))
|
||
except Exception as e:
|
||
print("abScan 失败: %s" % str(e)[:300])
|
||
ctx.close()
|
||
return 0
|
||
|
||
|
||
def run_rpc(rest, disable=None):
|
||
"""通用 RPC 直通：不写包装，直接调任意模块的任意导出。
|
||
|
||
例：
|
||
... unified_loader.py rpc gadget_loader sample
|
||
... unified_loader.py rpc gadget_loader spawn 1443076546,0,0
|
||
... unified_loader.py rpc gadget_loader indoorsectors
|
||
... unified_loader.py rpc addbtn abDiag
|
||
"""
|
||
if len(rest) < 3:
|
||
print("用法：unified_loader.py rpc <模块名> <RPC名> [参数1] [参数2] [参数3]")
|
||
print("可用模块见 config.json（gadget_loader / world_mod / train / bowling / balloon / addbtn）")
|
||
return 1
|
||
# 三个参数够用（如 findobj <关键字> <maxNodes> <maxDepth>）；不足就补 None。
|
||
args = (rest[3:6] + [None, None, None])[:3]
|
||
mod_name, rpc_name = rest[1], rest[2]
|
||
arg, arg2, arg3 = args[0], args[1], args[2]
|
||
cfg = load_cfg()
|
||
mods = build_modules(cfg, disable)
|
||
ctx = Ctx(mods)
|
||
_LIVE.append(ctx)
|
||
if not ctx.attach():
|
||
print("[!] 连接失败")
|
||
return 1
|
||
m = ctx.mod(mod_name)
|
||
if m is None:
|
||
print("[!] 没有模块 %s（没启用或加载失败）" % mod_name)
|
||
ctx.close()
|
||
return 1
|
||
apply_addbtn_cfg(ctx, cfg)
|
||
try:
|
||
if arg3 is not None:
|
||
r = m.call3(rpc_name, arg, arg2, arg3, timeout=40)
|
||
elif arg2 is not None:
|
||
r = m.call(rpc_name, arg, arg2, timeout=30)
|
||
else:
|
||
r = m.call(rpc_name, arg, timeout=30)
|
||
s = str(r)
|
||
print("%s.%s =>" % (mod_name, rpc_name))
|
||
# 排障时要看全量（如 abComp 拉几百个物件名），所以上限给到 400 KB
|
||
print(s if len(s) <= 400000 else s[:400000] + "\n...(截断，共 %d 字符)" % len(s))
|
||
except Exception as e:
|
||
print("%s.%s 失败：%s" % (mod_name, rpc_name, str(e)[:300]))
|
||
ctx.close()
|
||
return 0
|
||
|
||
|
||
def run_watch(disable=None):
|
||
cfg = load_cfg()
|
||
print("=" * 64)
|
||
print(" Ananta 注入器 · 保龄球瓶 + 气球")
|
||
print(" （全通道诊断版在 AnantaUnifiedLoader，日常不用它）")
|
||
print("=" * 64)
|
||
write_status("starting")
|
||
mods = build_modules(cfg, disable)
|
||
if not mods:
|
||
print("[!] 没有启用任何模块（config.json 全关或 --no 关完了）")
|
||
write_status("stopped", note="没有启用任何模块")
|
||
return 1
|
||
# ⛑ 常驻模式互斥：两个 frida 会话同时挂 Ananta.exe = 游戏必崩（踩过 3 次）。
|
||
# 改成「一键启动」之后重复双击的概率大增，这里必须程序自己挡。
|
||
owner = acquire_instance_lock()
|
||
if owner:
|
||
print("[X] 已经有另一个常驻加载器在跑（pid=%s）—— 同一时刻只能有一个 frida 会话。" % owner)
|
||
print(" 想重开请先关掉那一个（Ctrl+C 或直接关窗口）；")
|
||
print(" 若确认它已经死了，删掉 .loader.lock 再启动。")
|
||
return 1
|
||
print(" 已启用：%s" % ", ".join(m.name for m in mods))
|
||
# 2026-09-28：AddBtn 已降级为兜底（客户端自己会出 F），只在 config 里还配了 drive 时才这么说
|
||
_ab = next((m for m in (cfg.get("modules") or []) if m.get("name") == "addbtn"), None)
|
||
if _ab and _ab.get("drive"):
|
||
print(" ★ addbtn 会在你走近物件时自动补 F 提示按钮")
|
||
elif _ab:
|
||
print(" · addbtn 已降级为**兜底**（客户端自己会出 F）—— 不自动补")
|
||
print(" 要手动补：rpc addbtn abTick 3 2 收起：rpc addbtn abHide")
|
||
print(" 停止：Ctrl+C 或关窗口（都会先安全卸载注入）")
|
||
print("=" * 64)
|
||
|
||
ctx = Ctx(mods)
|
||
_LIVE.append(ctx)
|
||
|
||
ab_r = str(cfg.get("addbtnRadiusMeters", 6.0))
|
||
ab_b = str(cfg.get("addbtnBatch", 3))
|
||
seat_r = str(cfg.get("seatRadiusMeters", 8.0))
|
||
ab_every = float(cfg.get("addbtnEverySec", 2.0))
|
||
poll_every = float(cfg.get("statusPollSec", 30.0))
|
||
|
||
last_ab, last_poll, last_seat = 0.0, 0.0, 0.0
|
||
last_seatscan = 0.0
|
||
# 进入世界闸门 / armed 守卫 的状态（见 gate_ok / check_armed 的说明）
|
||
gate_state = {"pid": None, "file": None, "off": 0, "t0": None,
|
||
"last": 0.0, "passed": False}
|
||
armed_st = {"t0": None, "next": 0.0, "lastRes": "wait",
|
||
"attempts": 0, "cool": 0.0}
|
||
last_generic = {} # (模块名, rpc) -> 上次调用时间
|
||
drive_last = {} # 模块名 -> 最后一次状态（gadgets 数 / cell），自动收工用
|
||
idle_since = {} # 模块名 -> 上次看到的对象数
|
||
idle_t0 = {} # 模块名 -> 对象数开始没动静的时刻
|
||
stopped = {} # 模块名 -> 收工时所处格子（None = 正在跑）
|
||
# 2026-09-28 新增：移动后补货的救援状态（取代「不涨就收工」）
|
||
rescue_state = {} # 模块名 -> {prev_cell, moved, stable_since, peak, last_rescue}
|
||
rescue_ok = {} # 模块名 -> 该模块有没有 sweep RPC
|
||
rescue_cd = float(cfg.get("rescueCooldownSec", 25) or 25)
|
||
# ⛑ 这两个开关写在**模块**段里（和 autoStopAfterIdleSec 一起），不是顶层。
|
||
# 22:19 实机就栽在这：顶层读不到 ⇒ rescue_drop=0 ⇒ `cnt <= peak*1.0` 恒真
|
||
# ⇒ 每 25 秒无脑重扫一次。所以每个模块单独取，顶层值只作兜底。
|
||
rescue_cells = float(cfg.get("rescueMinCells", 3.0) or 3.0)
|
||
rescue_idle = float(cfg.get("rescueIdleSec", 3.0) or 3.0)
|
||
bad = 0
|
||
warned_busy = False
|
||
while True:
|
||
try:
|
||
if STOP["ask"]:
|
||
break
|
||
o = other_loader_running()
|
||
if o:
|
||
if ctx.session is not None:
|
||
print("[%s] [X] %s 也在注入 —— 立刻让出（同时注入会把游戏带崩）" % (ts(), o))
|
||
# ★ 不重建 Ctx：close() 已把 session/script 清空，下一轮直接重 attach。
|
||
# 重建再 append 会让 _LIVE 无限增长，退出时要逐个 close 一遍。
|
||
ctx.close()
|
||
if not warned_busy:
|
||
warned_busy = True
|
||
print("[%s] [i] %s 在跑，本程序等它退出（每 5 秒复查）" % (ts(), o))
|
||
sleep_slice(5.0)
|
||
continue
|
||
warned_busy = False
|
||
|
||
if ctx.session is None:
|
||
# ★ 进入世界闸门：没等到「玩家已进世界」就先不 attach
|
||
if not gate_ok(cfg, gate_state):
|
||
write_status("waiting_game" if resolve_pid() is None else "waiting_world")
|
||
sleep_slice(float((cfg.get("attachGate") or {}).get("pollSec", 2.0)))
|
||
continue
|
||
if not ctx.attach():
|
||
bad += 1
|
||
wait = min(3.0 * bad, 20.0)
|
||
if bad == 1 or bad % 10 == 0:
|
||
print("[%s] 没连上 Ananta.exe —— 等游戏启动（第 %d 次）" % (ts(), bad))
|
||
sleep_slice(wait)
|
||
continue
|
||
bad = 0
|
||
print("[%s] 已连接 pid=%d" % (ts(), ctx.pid))
|
||
run_setup(ctx, cfg)
|
||
apply_addbtn_cfg(ctx, cfg)
|
||
# 每次新建会话都重新计时：给 hook 安装 maxWaitSec 的机会
|
||
armed_st.update({"t0": time.time(), "next": 0.0,
|
||
"lastRes": "wait", "attempts": 0})
|
||
write_status("attached", game_pid=ctx.pid)
|
||
|
||
now = time.time()
|
||
|
||
# ★ 客户端崩了/重启之后，旧 script 会变成 "script has been destroyed"。
|
||
# 不主动 close() 的话 session 还在、但永远调不动，加载器就假死。
|
||
if ctx.dead:
|
||
print("[%s] [!] 注入已失效（客户端多半崩了/重启了）—— 准备重新注入" % ts())
|
||
ctx.close()
|
||
gate_state["passed"] = False # 新的一局 ⇒ 闸门重新等
|
||
write_status("waiting_game", note="客户端注入失效，准备重新注入")
|
||
bad = 0
|
||
sleep_slice(3.0)
|
||
continue
|
||
|
||
# ---- armed 守卫：hook 没装上 ⇒ 主动重挂，别干等 ----
|
||
_ar = check_armed(ctx, cfg, armed_st)
|
||
if _ar == "ok":
|
||
armed_st["attempts"] = 0
|
||
write_status("injected", game_pid=ctx.pid,
|
||
modules=[m.name for m in mods])
|
||
elif _ar == "reattach":
|
||
ag = cfg.get("armedGuard") or {}
|
||
if time.time() < float(armed_st.get("cool") or 0.0):
|
||
sleep_slice(2.0)
|
||
continue
|
||
armed_st["attempts"] = int(armed_st.get("attempts") or 0) + 1
|
||
if armed_st["attempts"] > int(ag.get("maxAttempts", 5)):
|
||
armed_st["cool"] = time.time() + float(ag.get("backoffSec", 120))
|
||
armed_st["attempts"] = 0
|
||
print("[%s] [守卫] 连试 %s 次都没装上 —— 退避 %ss 再试"
|
||
% (ts(), ag.get("maxAttempts", 5), ag.get("backoffSec", 120)))
|
||
else:
|
||
print("[%s] [守卫] 主动重挂（第 %d 次）" % (ts(), armed_st["attempts"]))
|
||
ctx.close()
|
||
armed_st.update({"t0": None, "next": 0.0, "lastRes": "wait"})
|
||
write_status("attached", note="hook 没装上，正在重挂")
|
||
sleep_slice(3.0)
|
||
continue
|
||
else:
|
||
sleep_slice(2.0)
|
||
continue
|
||
|
||
# ---- train：地铁车厢与座位（通道 D）----
|
||
# ⛑⛑ 2026-09-29 00:1x：这一段原来和 addbtn 一样是**无条件**跑的（硬编码每 1 秒
|
||
# 一次 seatfix），压根不读 config 的 drive ⇒ 单通道逐个测时关不掉、参数也改不动
|
||
# （同一类坑：addbtn 因此重复 AddBtn 把客户端刷崩，见上面那段注释）。
|
||
# 现在改成**以 config 的 drive 为唯一开关**：drive 为空 ⇒ 一次都不跑，
|
||
# 只留手动 `rpc train seatfix <半径>`。
|
||
tr_drive, tr_status = None, []
|
||
for _m in (cfg.get("modules") or []):
|
||
if _m.get("name") != "train":
|
||
continue
|
||
for _d in (_m.get("drive") or []):
|
||
if _d.get("rpc") == "seatfix":
|
||
tr_drive = _d
|
||
for _s in (_m.get("statusRpc") or []):
|
||
if _s in ("seatscan", "seattree", "trainnear", "trainhier", "where"):
|
||
tr_status.append(_s)
|
||
tr = ctx.mod("train")
|
||
if tr is not None and tr_drive and now - last_seat >= float(tr_drive.get("everySec", 1.0)):
|
||
last_seat = now
|
||
_sr = str(tr_drive.get("arg", seat_r))
|
||
try:
|
||
raw = tr.call("seatfix", _sr, timeout=15)
|
||
try:
|
||
r = json.loads(raw)
|
||
except Exception:
|
||
r = {}
|
||
# ★ 关键计数：total=缓存座位总数 / alreadyInited=已初始化 / fixed=本轮新开 /
|
||
# far=被 R_ALL(=max(半径,30)m) 门槛跳过的 —— 「整列车有一节坐不了」多半就是
|
||
# 那节的座位落进了 far。所以这几个数必须打出来，不能只打新开的那几行。
|
||
_bad = [x for x in (r.get("rows") or []) if x.get("err")]
|
||
print("[%s] [座椅] total=%s 已开=%s 本轮新开=%s 超距跳过=%s 扫描=%sms%s" % (
|
||
ts(), r.get("total"), r.get("alreadyInited"), r.get("fixed"),
|
||
r.get("far"), r.get("scanMs"),
|
||
(" Init报错=%d" % len(_bad)) if _bad else ""))
|
||
for _b in _bad[:3]:
|
||
print("[%s] ! %s err=%s" % (ts(), _b.get("n"), str(_b.get("err"))[:120]))
|
||
for row in r.get("rows") or []:
|
||
if row.get("after") == "true":
|
||
print("[%s] [座椅] 已开启 %s(%.1fm)" % (ts(), row.get("n"), row.get("d") or -1))
|
||
except Exception as e:
|
||
if "seatfix" not in str(e):
|
||
print("[%s] [座椅] %s" % (ts(), str(e)[:140]))
|
||
# 状态：seatscan / seattree / trainnear（通道 D 的判据看这里）
|
||
if tr is not None and tr_status and now - last_seatscan >= float(cfg.get("trainStatusEverySec", 5.0)):
|
||
last_seatscan = now
|
||
for _sr in tr_status:
|
||
try:
|
||
_raw = str(tr.call(_sr, timeout=15))
|
||
except Exception as e:
|
||
print("[%s] [train] %s %s" % (ts(), _sr, str(e)[:140]))
|
||
continue
|
||
# 原始扫描落盘：数「每节车厢到底几个座位」要靠它（seatscan 可能把
|
||
# 名为 Seats 的分组节点本身也算进去，必须按名字核）
|
||
try:
|
||
with open("_seats_last_%s.json" % _sr, "w", encoding="utf-8") as _fh:
|
||
_fh.write(_raw)
|
||
except Exception:
|
||
pass
|
||
# seattree / seatscan 都带 rows[].path ⇒ 按车厢汇总，一眼看出哪节没起来
|
||
_sum = None
|
||
try:
|
||
_j = json.loads(_raw)
|
||
_rows = _j.get("rows") or []
|
||
if _rows and isinstance(_rows[0], dict) and (_rows[0].get("path") or _rows[0].get("p")):
|
||
_cars = {}
|
||
for _rw in _rows:
|
||
_p = str(_rw.get("path") or _rw.get("p") or "")
|
||
_parts = _p.split("/")
|
||
_car = "/".join(_parts[:2]) if len(_parts) >= 2 else _parts[0]
|
||
_s = _cars.setdefault(_car, [0, 0])
|
||
_s[0] += 1
|
||
if str(_rw.get("ini")) == "true":
|
||
_s[1] += 1
|
||
_sum = "count=%d " % _j.get("count", len(_rows)) + " ".join(
|
||
"%s:%d/%d" % (k, v[1], v[0]) for k, v in sorted(_cars.items()))
|
||
except Exception:
|
||
_sum = None
|
||
if _sum:
|
||
print("[%s] [train·%s] %s" % (ts(), _sr, _sum))
|
||
else:
|
||
print("[%s] [train·%s] %s" % (ts(), _sr, _raw[:400]))
|
||
|
||
# ---- addbtn：补 F 按钮 ----
|
||
# ⛑⛑ 2026-09-28 23:0x：这一段原来是**无条件**跑的，压根不读 config 的 drive。
|
||
# 所以上一轮把 drive 清空之后它照样每 4 秒 abTick 一轮 ⇒
|
||
# 客户端 23:02:22 刷 `添加了重复的按钮 ... pid 14013633775178039662（衣柜）`
|
||
# —— 客户端自己已经出了那颗按钮，我们又 AddBtn 一个同 pid 的，
|
||
# btnInfos[pid] 的 List 无限增长，这是已记录的必崩项。
|
||
# 现在改成和别的模块一样：**以 config 的 drive 为唯一开关**。
|
||
# drive 为空 ⇒ 一个都不补，只留手动 `rpc addbtn abTick <半径> <批量>`。
|
||
ab_drive = None
|
||
for _m in (cfg.get("modules") or []):
|
||
if _m.get("name") != "addbtn":
|
||
continue
|
||
for _d in (_m.get("drive") or []):
|
||
if _d.get("rpc") == "abTick":
|
||
ab_drive = _d
|
||
break
|
||
if ab_drive:
|
||
ab_every = float(ab_drive.get("everySec", ab_every))
|
||
ab_r = str(ab_drive.get("arg", ab_r))
|
||
ab_b = str(ab_drive.get("arg2", ab_b))
|
||
ab = ctx.mod("addbtn")
|
||
if ab is not None and ab_drive and now - last_ab >= ab_every:
|
||
last_ab = now
|
||
try:
|
||
raw = ab.call("abTick", ab_r, ab_b, timeout=20)
|
||
try:
|
||
r = json.loads(raw)
|
||
except Exception:
|
||
r = {}
|
||
if r.get("err"):
|
||
# 模板按钮没拿到是最常见的初始状态，别每 2 秒刷屏
|
||
if now - last_poll >= poll_every:
|
||
last_poll = now
|
||
print("[%s] [F按钮] %s" % (ts(), str(r.get("err"))[:160]))
|
||
continue
|
||
for row in r.get("made") or []:
|
||
# ★ 光看「已补」没用 —— 按钮造出来不等于会显示。
|
||
# 把客户端自己对这颗按钮的判定（CanShow / InternalCanShow / 原因）打出来。
|
||
ck = row.get("check") or {}
|
||
print("[%s] [F按钮] 已补 %s（%.1fm）mode=%s | canShow=%s internal=%s isShow=%s dis=%s text=%r reason=%s"
|
||
% (ts(), row.get("n") or row.get("uid"), row.get("d") or -1, row.get("mode"),
|
||
ck.get("canShow"), ck.get("internalCanShow"), ck.get("isShow"),
|
||
ck.get("dis"), ck.get("text"), ck.get("reason")))
|
||
for row in r.get("fail") or []:
|
||
print("[%s] [F按钮] 失败 %s：%s" % (ts(), row.get("n"), str(row.get("err"))[:100]))
|
||
except Exception as e:
|
||
if now - last_poll >= poll_every:
|
||
last_poll = now
|
||
print("[%s] [F按钮] %s" % (ts(), str(e)[:160]))
|
||
|
||
# ---- 其余模块：按 config 的 drive 节奏驱动 ----
|
||
for m in cfg.get("modules") or []:
|
||
nm = m.get("name")
|
||
if nm in ("train", "addbtn"):
|
||
continue # 这两个上面有专门分支
|
||
mod = ctx.mod(nm)
|
||
if mod is None:
|
||
continue
|
||
|
||
# ================= 移动后补货：网格救援 =================
|
||
# ⛔ 原来「物件数不涨就 stop()」的做法已作废（2026-09-28）：它与「移动之后
|
||
# 周围的网格也要加载出物件」根本冲突 —— 收工后玩家走到新区域，身边那批
|
||
# 格子再也没人重试。现在流一直开着（靠 tune 节流），出问题用 sweep 主动重扫。
|
||
if nm not in rescue_ok:
|
||
try:
|
||
rescue_ok[nm] = (mod.script is not None
|
||
and hasattr(mod.script.exports_sync, "sweep"))
|
||
except Exception:
|
||
rescue_ok[nm] = False
|
||
|
||
st0 = drive_last.get(nm) or {}
|
||
cnt0 = st0.get("gadgets")
|
||
cell0 = st0.get("cell")
|
||
rs = rescue_state.get(nm)
|
||
if rs is None:
|
||
rs = rescue_state[nm] = {"prev_cell": cell0, "moved": 0.0,
|
||
"stable_since": None, "peak": cnt0 or 0,
|
||
"last_rescue": 0.0}
|
||
if isinstance(cell0, (list, tuple)) and len(cell0) == 2:
|
||
pc0 = rs.get("prev_cell")
|
||
if isinstance(pc0, (list, tuple)) and len(pc0) == 2:
|
||
d0 = ((cell0[0] - pc0[0]) ** 2 + (cell0[1] - pc0[1]) ** 2) ** 0.5
|
||
if d0 > 0:
|
||
rs["moved"] += d0
|
||
rs["stable_since"] = None
|
||
if rs.get("prev_cell") != cell0:
|
||
rs["prev_cell"] = cell0
|
||
if rs.get("stable_since") is None:
|
||
rs["stable_since"] = now
|
||
if isinstance(cnt0, int):
|
||
rs["peak"] = max(rs.get("peak") or 0, cnt0)
|
||
|
||
if rescue_ok.get(nm) and cell0 and now - rs.get("last_rescue", 0.0) >= rescue_cd:
|
||
stable_enough = (rs.get("stable_since") is not None
|
||
and now - rs["stable_since"] >= rescue_idle)
|
||
# ⛑ 模块段优先（这两个开关和 autoStopAfterIdleSec 写在一起），顶层作兜底
|
||
drop_pct = float(m.get("rescueOnDropPct",
|
||
cfg.get("rescueOnDropPct", 0)) or 0)
|
||
why = None
|
||
if rs.get("moved", 0.0) >= rescue_cells and stable_enough:
|
||
why = "移动 %.1f 格后停下（JS 到达救援没响，Python 兜底）" % rs["moved"]
|
||
elif (stable_enough and drop_pct > 0 and isinstance(cnt0, int)
|
||
and rs.get("peak", 0) >= 10
|
||
and cnt0 < rs["peak"] * (1.0 - drop_pct / 100.0)):
|
||
why = ("物件数从峰值 %d 掉到 %d（-%.0f%%）"
|
||
% (rs["peak"], cnt0,
|
||
(1.0 - cnt0 / float(rs["peak"])) * 100))
|
||
if why:
|
||
try:
|
||
out = mod.call("sweep", str(int(m.get("rescueSweepRadius", 3))),
|
||
timeout=20)
|
||
rs["last_rescue"] = now
|
||
rs["moved"] = 0.0
|
||
rs["peak"] = cnt0 if isinstance(cnt0, int) else 0
|
||
print("[%s] [网格救援] %s ⇒ %s" % (ts(), why, str(out)[:200]))
|
||
except Exception as e:
|
||
print("[%s] [网格救援] sweep 失败：%s" % (ts(), str(e)[:140]))
|
||
|
||
# ★ 自动收工：物件数不再增长 ⇒ 调 stop()，别让「空地重试」无限刷
|
||
# ⛔ 默认 0 = 关闭（与「移动后继续补货」冲突）。真要防崩请调 tune 的节流参数。
|
||
idle_sec = float(m.get("autoStopAfterIdleSec") or 0)
|
||
if idle_sec > 0:
|
||
st = drive_last.get(nm)
|
||
cnt = (st or {}).get("gadgets")
|
||
cell = (st or {}).get("cell")
|
||
if stopped.get(nm):
|
||
# 玩家换格子了 ⇒ 重新开一轮
|
||
if m.get("restartOnCellChange") and cell and cell != stopped.get(nm):
|
||
print("[%s] [%s] 玩家移动到 %s —— 重新开始加载" % (ts(), nm, cell))
|
||
stopped[nm] = None
|
||
idle_since.pop(nm, None)
|
||
try:
|
||
mod.call("start", str(cfg.get("worldRaidId", 23300888)), timeout=15)
|
||
except Exception as e:
|
||
print("[%s] [%s] start 失败：%s" % (ts(), nm, str(e)[:120]))
|
||
elif cnt is not None:
|
||
if idle_since.get(nm) != cnt:
|
||
idle_since[nm] = cnt
|
||
idle_t0[nm] = now
|
||
elif now - idle_t0.get(nm, now) >= idle_sec:
|
||
try:
|
||
mod.call("stop", timeout=15)
|
||
stopped[nm] = cell
|
||
print("[%s] [%s] 物件数 %s 已 %ss 没增长 —— 收工（走动后自动重开）"
|
||
% (ts(), nm, cnt, int(idle_sec)))
|
||
except Exception as e:
|
||
print("[%s] [%s] stop 失败：%s" % (ts(), nm, str(e)[:120]))
|
||
if stopped.get(nm):
|
||
continue # 已收工，本轮不再驱动
|
||
|
||
for d in m.get("drive") or []:
|
||
rpc = d.get("rpc")
|
||
if not rpc:
|
||
continue
|
||
every = float(d.get("everySec", 5.0))
|
||
key = (nm, rpc)
|
||
if now - last_generic.get(key, 0.0) < every:
|
||
continue
|
||
last_generic[key] = now
|
||
try:
|
||
raw = mod.call(rpc, d.get("arg"), timeout=d.get("timeout", 20))
|
||
line = fmt_status(nm, rpc, raw)
|
||
if line:
|
||
print("[%s] [%s] %s" % (ts(), nm, line))
|
||
# 记下最新状态，供「自动收工」判断物件数有没有增长
|
||
try:
|
||
sd = json.loads(str(raw))
|
||
# ⛑ 别让 gorender 这类「不报状态」的 RPC 把 cell/gadgets 冲成 None
|
||
# （救援逻辑靠这两个值判断玩家有没有移动）
|
||
if not (isinstance(sd, dict) and ("world" in sd or "cell" in sd)):
|
||
continue
|
||
w = sd.get("world") or {}
|
||
# ⛑ cell 存原始 list 而不是 str —— 救援逻辑要算两格之间的距离
|
||
drive_last[nm] = {"gadgets": w.get("GadgetsInWorld"),
|
||
"cell": sd.get("cell"),
|
||
"confirmed": sd.get("confirmed"),
|
||
"giveUp": sd.get("giveUpCells"),
|
||
"pending": sd.get("pendingCells"),
|
||
"arrive": sd.get("arriveCount")}
|
||
except Exception:
|
||
pass
|
||
except Exception as e:
|
||
print("[%s] [%s] %s 失败：%s" % (ts(), nm, rpc, str(e)[:140]))
|
||
|
||
# ---- 其余自驱动模块：定期轮询一次状态 ----
|
||
if now - last_poll >= poll_every:
|
||
last_poll = now
|
||
# addbtn 只在「补成功/失败」时才说话，平时一声不响，
|
||
# 排障时根本不知道是「没候选」还是「造不出来」。所以每轮报一次家底。
|
||
abm = ctx.mod("addbtn")
|
||
if abm is not None:
|
||
try:
|
||
print("[%s] [F按钮·状态] %s" % (ts(), str(abm.call("abStatus", timeout=10))[:400]))
|
||
except Exception as e:
|
||
print("[%s] [F按钮·状态] %s" % (ts(), str(e)[:140]))
|
||
try:
|
||
sc = json.loads(str(abm.call("abScan", ab_r, timeout=15)))
|
||
print("[%s] [F按钮·候选] %d 个（已有按钮 %d 个）%s" % (
|
||
ts(), sc.get("count", 0), sc.get("existing", 0),
|
||
", ".join("%s@%sm" % (str(r.get("n"))[:24], r.get("d"))
|
||
for r in (sc.get("rows") or [])[:6])))
|
||
except Exception as e:
|
||
print("[%s] [F按钮·候选] %s" % (ts(), str(e)[:140]))
|
||
try:
|
||
dg = json.loads(str(abm.call("abDiag", timeout=20)))
|
||
print("[%s] [F按钮·体检] pos=%s holdTotal=%s holdNear10m=%s useful=%s ctor=%s" % (
|
||
ts(), dg.get("playerPos"), dg.get("holdTotal"), dg.get("holdNear10m"),
|
||
dg.get("usefulListCount"), dg.get("ctorParamCounts")))
|
||
except Exception as e:
|
||
print("[%s] [F按钮·体检] %s" % (ts(), str(e)[:140]))
|
||
# 现有按钮逐个问客户端「你为什么不让这个显示」
|
||
try:
|
||
cs = json.loads(str(abm.call("abCanShow", timeout=20)))
|
||
g = cs.get("global") or {}
|
||
print("[%s] [F按钮·可见性] 全局=%s/%s 按钮数=%s" % (
|
||
ts(), g.get("globalInteractAble"), g.get("hideName"), cs.get("count")))
|
||
for row in (cs.get("rows") or [])[:5]:
|
||
print("[%s] pid=%s %s canShow=%s internal=%s dis=%s reason=%s"
|
||
% (ts(), row.get("pid"), row.get("cls"),
|
||
row.get("canShow"), row.get("internalCanShow"),
|
||
row.get("dis"), row.get("reason")))
|
||
except Exception as e:
|
||
print("[%s] [F按钮·可见性] %s" % (ts(), str(e)[:140]))
|
||
for m in cfg.get("modules") or []:
|
||
if m.get("name") in ("train", "addbtn"):
|
||
continue
|
||
mod = ctx.mod(m.get("name"))
|
||
if mod is None:
|
||
continue
|
||
n, r = once_status(ctx, mod, m.get("statusRpc"))
|
||
if r:
|
||
print("[%s] [%s] %s" % (ts(), m.get("name"), r[:300]))
|
||
|
||
sleep_slice(0.5)
|
||
except KeyboardInterrupt:
|
||
break
|
||
except Exception as e:
|
||
try:
|
||
ctx.close()
|
||
except Exception:
|
||
pass
|
||
bad += 1
|
||
wait = min(3.0 * bad, 20.0)
|
||
print("[%s] 连接中断（%s）—— 第 %d 次，%.0f 秒后重连" % (ts(), str(e)[:130], bad, wait))
|
||
sleep_slice(wait)
|
||
|
||
print("[%s] 正在退出并卸载注入……" % ts())
|
||
write_status("stopped")
|
||
try:
|
||
ctx.close()
|
||
except Exception:
|
||
pass
|
||
STOP["done"] = True
|
||
return 0
|
||
|
||
|
||
def main():
|
||
atexit.register(_atexit_close)
|
||
argv = sys.argv[1:]
|
||
disable = None
|
||
rest = []
|
||
i = 0
|
||
while i < len(argv):
|
||
if argv[i] in ("--no", "--disable") and i + 1 < len(argv):
|
||
disable = set(x.strip() for x in argv[i + 1].split(",") if x.strip())
|
||
i += 2
|
||
continue
|
||
rest.append(argv[i])
|
||
i += 1
|
||
|
||
a0 = (rest[0].lower() if rest else "")
|
||
if a0 in ("diag", "debug", "check", "test"):
|
||
return run_diag(disable)
|
||
if a0 in ("scan", "abscan"):
|
||
return run_scan(disable)
|
||
if a0 in ("rpc", "call"):
|
||
return run_rpc(rest, disable) # rest 已含 "rpc" 本身
|
||
if a0 in ("-h", "--help", "help", "?"):
|
||
print(__doc__)
|
||
return 0
|
||
return run_watch(disable)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
sys.exit(main())
|
||
except Exception as e:
|
||
print("[X] %s" % e)
|
||
try:
|
||
import traceback
|
||
traceback.print_exc()
|
||
except Exception:
|
||
pass
|
||
if getattr(sys.stdin, None) and sys.stdin.isatty():
|
||
try:
|
||
input("\n按回车退出……")
|
||
except Exception:
|
||
pass
|