# -*- 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 <模块名> [参数1] [参数2] [参数3]") print("可用模块见 config.json（gadget_loader / world_mod / train / bowling / balloon / addbtn）") return 1 # 三个参数够用（如 findobj <关键字> ）；不足就补 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