"""tp_probe_pathfind.py — 探测传送员真实标识 → 对话传送 → 自动寻路计时。 前置实证(2026-09-29): - 走位自愈链路已通(goto 拉回采信循环, 13s 内到传送员旁) - 静态 aid=10002 + index=2 的 CS1010 全部无响应(9/27 同参数 0.18s 即回 SC643) - quest_bot3 定案: CS1010 的 UserID=index 来自 NPC 的 SC10 推送(动态, 每角色不同) → 本脚本: 站到传送员旁收集 SC10 原始帧, 找 body 含"传送"的实体, 用其真实 (info[0], p1) 做对话候选; 全部失败则穷举视野实体 × index 候选。 """ from __future__ import annotations import importlib.util import json import logging import re import sys import time from collections import deque from pathlib import Path ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT / "cpp_reversed")) sys.path.insert(0, str(ROOT / "protocol_bot" / "src")) import requests # noqa: E402 from pbot.codec.bitbuf import decode6 # noqa: E402 from pbot.config import Account # noqa: E402 from pbot.game.messages import protocol_map_name # noqa: E402 from pbot.game.session import enter_world # noqa: E402 from pbot.login.box_client import enter_game_client, login # noqa: E402 from pbot.transport.gateway import login_gate_with_retry # noqa: E402 from pbot.world.update import WorldUpdater, decode_body_variants # noqa: E402 from map_parser import load as load_map # noqa: E402 from path_find import path_is_valid # noqa: E402 # 复用已验证的 goto/pump/规划 helpers _spec = importlib.util.spec_from_file_location( "wt", ROOT / "protocol_bot" / "scripts" / "watch_tp_pathfind.py") wt = importlib.util.module_from_spec(_spec) sys.modules["wt"] = wt _spec.loader.exec_module(wt) _LOG = logging.getLogger("tp_probe") TARGET_MAP = "地牢一层东" LINK_RE = re.compile(r"link=@([^|<>]+)") def _nearest_walkable(m, pos: tuple[int, int]) -> tuple[int, int] | None: """pos 周围最近的可走格(螺旋), 供墙格出生点逃逸。""" if 0 <= pos[0] < m.width and 0 <= pos[1] < m.height \ and m.get_cell(pos[0], pos[1]).is_walkable: return pos for rr in range(1, 6): for dx in range(-rr, rr + 1): for dy in range(-rr, rr + 1): if max(abs(dx), abs(dy)) != rr: continue x, y = pos[0] + dx, pos[1] + dy if 0 <= x < m.width and 0 <= y < m.height \ and m.get_cell(x, y).is_walkable: return (x, y) return None def collect_sc10(conn, world, updater, seconds: float) -> list: """收帧 seconds; 返回 SC10 原始帧列表(同时喂 updater)。""" out = [] t0 = time.time() while time.time() - t0 < seconds: for f in conn.recv_frames(0.2): if f.msg_id == 10: out.append(f) updater.apply(world, f) world.events.clear() return out def sc10_info(f) -> dict | None: """解析 SC10: 返回 {race, key, x, y, aid, name, raw} 或 None。""" race = f.recog & 0xFFFF x, y = f.p1 & 0xFFFF, (f.p1 >> 16) & 0xFFFF info: list[str] = [] raw = "" for blob in decode_body_variants(f.body): try: txt = blob.decode("utf-8") except UnicodeDecodeError: continue raw = txt try: j = json.loads(txt) except ValueError: continue if isinstance(j, dict) and isinstance(j.get("info"), list): info = [str(v) for v in j["info"]] break return {"race": race, "key": str(f.p1), "x": x, "y": y, "aid": info[0] if info else "", "name": info[1] if len(info) > 1 else "", "raw": raw, "p2": f.p2, "p3": f.p3} def main() -> int: http = requests.Session() http.trust_env = False http.verify = False jwt, uid = login(Account(username="longwu988", password="Longwu1988")) _ = enter_game_client(http, jwt, 14558825) key32, sid, rname, tkey, gport = login_gate_with_retry( 14558825, jwt, uid, "凌风踏雪", wt.PORTS) stop = [False] conn, world, updater = enter_world(14558825, rname, tkey, sid, gport, uid, stop=stop) _LOG.info("[出生] %s @(%d,%d)", world.me.map_name, world.me.x, world.me.y) try: if world.me.map_name != TARGET_MAP: time.sleep(8.0) wt.pump(conn, world, updater, 1.0) # 回城落地必须 ACK 盟重(5012/5027, README 铁律), 否则服务器 # 不下发本图视野推送(SC10) — 拿不到传送员动态 aid/index proto3 = protocol_map_name(world.me.map_id, world.me.map_name) for _ in range(2): _ = conn.send_frame(5012, 0, 0, 0, sid, proto3.encode()) _ = conn.send_frame(5027, 0, 0, 0, sid, proto3.encode()) _ = conn.send_frame(5012, 0, 0, 0, 32611848, b"cTh") _LOG.info("[盟重ACK] body=%r", proto3) t0 = time.time() while time.time() - t0 < 6: wt.pump(conn, world, updater, 0.3) m3 = load_map(str(ROOT / "protocol_bot" / "assets" / "maps" / "3.map")) # 出生点可能落在墙格(服务器随机落点): 先逃到最近可走格再规划 me0 = (world.me.x, world.me.y) if not m3.get_cell(me0[0], me0[1]).is_walkable: near = _nearest_walkable(m3, me0) if near and wt.goto(conn, sid, world, updater, m3, near): _LOG.info("[出墙] %s(墙) → %s", me0, near) if not wt.goto(conn, sid, world, updater, m3, wt.TELEPORTER_POS): _LOG.error("[走位] 失败") return 1 _LOG.info("[到位] (%d,%d)", world.me.x, world.me.y) # ---- 探测: 收集 SC10, 找传送员 ---- sc10s = collect_sc10(conn, world, updater, 12.0) _LOG.info("[SC10] 共 %d 帧", len(sc10s)) parsed = [p for p in (sc10_info(f) for f in sc10s) if p] for p in parsed[:20]: _LOG.info(" SC10 race=%d key=%s pos=(%d,%d) aid=%s name=%s", p["race"], p["key"], p["x"], p["y"], p["aid"][:16], p["name"][:16]) tp = [p for p in parsed if "传送" in p["name"] or "九" in p["name"] or "传送" in p["raw"]] _LOG.info("[探测] 含'传送'实体: %d 个", len(tp)) # ---- 构造对话候选 ---- cands: list[tuple[str, int]] = [] for p in tp: cands.append((p["aid"], int(p["key"]) if p["key"].isdigit() else 2)) cands.append((p["aid"], 2)) if not cands: cands = [(wt.TELEPORTER_AID, 2)] # 兜底穷举: 视野实体 aid × 常见 index extra = [(p["aid"], idx) for p in parsed if p["aid"].isdigit() for idx in (2, 35, 34)] for c in extra: if c not in cands: cands.append(c) _LOG.info("[候选] %s", cands[:12]) # ---- 逐候选点对话 ---- dlg = None for i, (aid_c, idx_c) in enumerate(cands[:10]): body = json.dumps({"UserID": aid_c, "index": idx_c}, separators=(",", ":")).encode() _ = conn.send_frame(1010, 0, 0, 0, sid, body) _LOG.info("[对话] 候选%d aid=%s index=%d", i + 1, aid_c[:16], idx_c) t0 = time.time() while time.time() - t0 < 4: wt.pump(conn, world, updater, 0.2) if world.npc_dialog: dlg = world.npc_dialog break if dlg is not None: _LOG.info("[弹窗✓] 候选%d 命中", i + 1) break if dlg is None: _LOG.error("[对话] 全部候选无弹窗 — dump 终态") return 1 smsg = str(dlg.get("sMsg", "")) links = LINK_RE.findall(smsg) _LOG.info("[SC643] UserID=%s index=%s 按钮: %s", dlg.get("UserID"), dlg.get("index"), links[:24]) # 连点全部非 exit 按钮(9/27 模式: QFunction 一次给全, 顺序点) for act in [a for a in links if a not in ("exit", "close")]: _ = conn.send_frame( 1011, 0, 0, 0, sid, json.dumps({"UserID": str(dlg.get("UserID", "999999999")), "Act": f"@{act}", "index": int(dlg.get("index", 999999999))}, separators=(",", ":")).encode()) _LOG.info("[点] @%s", act) t0 = time.time() while time.time() - t0 < 5: wt.pump(conn, world, updater, 0.2) if world.me.map_name == TARGET_MAP: break if world.me.map_name == TARGET_MAP: break if world.me.map_name != TARGET_MAP: _LOG.error("[传送] 未落地 — 终态 %s", world.me.map_name) return 1 # ---- 地牢: ACK + 自动寻路计时 ---- proto = protocol_map_name(world.me.map_id, world.me.map_name) for _ in range(2): _ = conn.send_frame(5012, 0, 0, 0, sid, proto.encode()) _ = conn.send_frame(5027, 0, 0, 0, sid, proto.encode()) _ = conn.send_frame(5012, 0, 0, 0, 32611848, b"cTh") t0 = time.time() while time.time() - t0 < 4: wt.pump(conn, world, updater, 0.2) _LOG.info("[落地] %s @(%d,%d)", world.me.map_name, world.me.x, world.me.y) start = (world.me.x, world.me.y) md = load_map(str(ROOT / "protocol_bot" / "assets" / "maps" / wt.MAP_FILES[TARGET_MAP])) if not (0 <= start[0] < md.width and 0 <= start[1] < md.height): _LOG.error("[地图] 落脚点越界 %s", start) return 1 dist = wt.bfs_walkable(md, start) goal = max(dist, key=lambda p: ((p[0] - start[0]) ** 2 + (p[1] - start[1]) ** 2, dist[p])) _LOG.info("[目标] (%d,%d) 可达域 %d 格 曼哈顿 %d", goal[0], goal[1], len(dist), abs(goal[0] - start[0]) + abs(goal[1] - start[1])) t1 = time.time() path = wt.plan_multiseg(md, start, goal) if path is None or not path_is_valid(md, path): _LOG.error("[规划] 失败") return 1 legs = wt._compress(path) _LOG.info("[规划] %d 步 → %d 腿, %.2fs", len(path), len(legs), time.time() - t1) _ = conn.send_frame(5013, goal[0], goal[1], 1, sid, b"") _LOG.info("[5013] START → (%d,%d)", *goal) t_start = time.time() confirmed = rejected = silent = replan = 0 li = 0 cur_legs = legs while li < len(cur_legs) and replan < 40: me = (world.me.x, world.me.y) while li < len(cur_legs) and max(abs(cur_legs[li][0] - me[0]), abs(cur_legs[li][1] - me[1])) <= 2: li += 1 if li >= len(cur_legs): break if max(abs(cur_legs[li][0] - me[0]), abs(cur_legs[li][1] - me[1])) > 4: rest = wt.plan_multiseg(md, me, goal) if rest is None: _LOG.error("[重规划] 无路 — ABORT") break cur_legs = wt._compress(rest) li = 0 replan += 1 continue nxt = cur_legs[li] dx = max(-1, min(1, nxt[0] - me[0])) dy = max(-1, min(1, nxt[1] - me[1])) heading = next((i for i, (ox, oy) in enumerate(wt.DIRS) if ox == dx and oy == dy), 4) d2 = 2 if (abs(nxt[0] - me[0]) == 2 or abs(nxt[1] - me[1]) == 2) else 1 r = wt.walk_one_leg(conn, sid, world, updater, heading, d2, proto) if r == "ok": confirmed += 1 li += 1 elif r == "push": rejected += 1 rest = wt.plan_multiseg(md, (world.me.x, world.me.y), goal) if rest is None: break cur_legs = wt._compress(rest) li = 0 replan += 1 else: silent += 1 li += 1 n = confirmed + rejected if n and n % 40 == 0: _LOG.info("[进度] #%d @(%d,%d) ✓%d ✗%d 静默%d", n, world.me.x, world.me.y, goal[0], goal[1], confirmed, rejected, silent) elapsed = time.time() - t_start _ = conn.send_frame(5013, 0, 0, 3, sid, b"") final = (world.me.x, world.me.y) ok = max(abs(final[0] - goal[0]), abs(final[1] - goal[1])) <= 3 _LOG.info("=" * 60) _LOG.info("[结果] %s 起点(%d,%d)→目标(%d,%d) 终点(%d,%d)", "✓到达" if ok else "✗未达", start[0], start[1], goal[0], goal[1], final[0], final[1]) _LOG.info("[计时] %.1fs (%.1f 分钟)", elapsed, elapsed / 60) _LOG.info("[帧统计] SC1=%d SC28=%d 静默=%d 重规划=%d", confirmed, rejected, silent, replan) return 0 if ok else 2 finally: stop[0] = True conn.close() if __name__ == "__main__": sys.exit(main())