"""auto_pathfind.py — 专用自动寻路计时脚本(无战斗分支, 纯跑图)。 复刻真实客户端【自动寻路】实现思路: 1. 传送落地(任务链 CS3031 → CS1011 @act_move,N) 2. 加载地图数据(d401.map 等), 从落脚点 BFS 全图可走连通域 3. 选离落脚点欧氏距离最远的可达可走格作为目标(保证图内+可达) 4. 用真客户端还原的 A*(cpp_reversed/path_find.py: aiManager::FindPath 8向/直线10/对角14/欧氏²启发/防切角)按 48x32 包围盒分段拼接规划路径 5. CS5013(act=1,X,Y) 注册寻路会话 → 沿路径逐腿 CS3013 跑 (同向连吃 2 格对齐客户端 checkNextMovePos 跑步步长, pre=预测落点) 6. 到达 → CS5013(act=3 FINISH); 全程计时+帧统计 用法: uv run python scripts/auto_pathfind.py -u <账号> -p <密码> \ --zone 14558825 --name 凌风踏雪 --map 地牢一层东 --act act_move,2 可选: --target x,y 指定目标(默认自动选最远可达点); --dry-run 只规划不发包。 """ from __future__ import annotations import argparse import logging 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.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 GatewayConnection, login_gate_with_retry # noqa: E402 from pbot.types import ZoneId # noqa: E402 from pbot.world.update import WorldUpdater # noqa: E402 from map_parser import MapData, load as load_map # noqa: E402 from path_find import find_path, path_is_valid # noqa: E402 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)-7s %(name)s: %(message)s", stream=sys.stdout) _LOG = logging.getLogger("auto_pathfind") DIRS = ((0, -1), (1, -1), (1, 0), (1, 1), (0, 1), (-1, 1), (-1, 0), (-1, -1)) DIS_X, DIS_Y = 24, 16 # 真客户端 updateLimits 包围盒(PathFindAStar.lua:52-53) RUN_STEP_S = 0.60 # 步频(真客户端 0.55-0.6s, M2 封速门 1.4s 由 SC1 门控保证) PENDING_TIMEOUT_S = 1.0 ARRIVE_DIST = 1 MAP_FILES = { "地牢一层东": "d401.map", "天然洞穴一层": "d011.map", "比奇大城": "0.map", "盟重土城": "3.map", } PORTS = (8021, 8061, 8001, 8041) def bfs_walkable(m: MapData, start: tuple[int, int]) -> dict[tuple[int, int], int]: """从 start 做 8 向 BFS(含防切角), 返回 {格: 步数} — 服务器可走域近似。""" walk = [[m.get_cell(x, y).is_walkable for x in range(m.width)] for y in range(m.height)] dist: dict[tuple[int, int], int] = {start: 0} q: deque[tuple[int, int]] = deque([start]) while q: x, y = q.popleft() for dx, dy in DIRS: nx, ny = x + dx, y + dy if not (0 <= nx < m.width and 0 <= ny < m.height): continue if not walk[ny][nx] or (nx, ny) in dist: continue if dx and dy and (not walk[y][x + dx] or not walk[y + dy][x]): continue # 防切角: 真客户端 A* 同款规则 dist[(nx, ny)] = dist[(x, y)] + 1 q.append((nx, ny)) return dist def pick_farthest(dist: dict[tuple[int, int], int], start: tuple[int, int]) -> tuple[int, int]: """欧氏距离最远的可达格(相同取步数多者)。""" return max(dist, key=lambda p: ((p[0] - start[0]) ** 2 + (p[1] - start[1]) ** 2, dist[p])) def plan_multiseg(m: MapData, beg: tuple[int, int], end: tuple[int, int]) -> list[tuple[int, int]] | None: """分段 A*: 每段 waypoint 限制在 48x32 包围盒内(真客户端 PathFindController 多段调用拼接思路), 段终点被墙挡时螺旋扰动重试。 """ full: list[tuple[int, int]] = [beg] cur = beg while cur != end: dx, dy = end[0] - cur[0], end[1] - cur[1] d = max(abs(dx), abs(dy)) if d == 0: break step = min(20, d) wx = cur[0] + round(dx * step / d) wy = cur[1] + round(dy * step / d) wx = max(cur[0] - DIS_X, min(cur[0] + DIS_X, wx)) wy = max(cur[1] - DIS_Y, min(cur[1] + DIS_Y, wy)) got: list[tuple[int, int]] | None = None for r in range(9): cands = [(wx + ox, wy + oy) for ox in range(-r, r + 1) for oy in range(-r, r + 1) if max(abs(ox), abs(oy)) == r] for c in cands: if not (0 <= c[0] < m.width and 0 <= c[1] < m.height): continue seg = find_path(m, cur[0], cur[1], c[0], c[1]) if seg: got = seg break if got: break if not got: return None full.extend(got[1:]) cur = got[-1] if len(full) > 3000: return None return full def compress_run_legs(path: list[tuple[int, int]]) -> list[tuple[int, int]]: """沿路径同向连吃 2 格(客户端 checkNextMovePos 跑步=2 格同向合并)。""" legs: list[tuple[int, int]] = [] i = 0 while i < len(path) - 1: if i + 2 < len(path): d1 = (path[i + 1][0] - path[i][0], path[i + 1][1] - path[i][1]) d2 = (path[i + 2][0] - path[i + 1][0], path[i + 2][1] - path[i + 1][1]) if d1 == d2 and d1 in DIRS: legs.append(path[i + 2]) i += 2 continue legs.append(path[i + 1]) i += 1 return legs def heading_of(a: tuple[int, int], b: tuple[int, int]) -> int: dx = max(-1, min(1, b[0] - a[0])) dy = max(-1, min(1, b[1] - a[1])) return next((i for i, (ox, oy) in enumerate(DIRS) if ox == dx and oy == dy), 4) def send_move(conn: GatewayConnection, sid: int, me_x: int, me_y: int, heading: int, land_x: int, land_y: int, proto: str) -> None: """CS3013 跑一腿: recog=0, p1=dir, p2=起点(Y<<16|X), p3=sid, pre=落点。""" _ = conn.send_frame(3013, 0, heading, (me_y << 16) | me_x, sid, proto.encode(), pre=(land_y << 16) | land_x) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("-u", "--user", required=True) ap.add_argument("-p", "--password", required=True) ap.add_argument("--zone", type=int, required=True) ap.add_argument("--name", required=True) ap.add_argument("--map", default="地牢一层东") ap.add_argument("--act", default="act_move,2", help="传送按钮 Act(不带@)") ap.add_argument("--target", default="", help="x,y 指定目标(默认自动最远)") ap.add_argument("--dry-run", action="store_true", help="只规划不发帧") a = ap.parse_args() map_file = ROOT / "protocol_bot" / "assets" / "maps" / MAP_FILES[a.map] _LOG.info("[地图] %s → %s", a.map, map_file.name) # ---- 登录进游戏 ---- http = requests.Session() http.trust_env = False http.verify = False jwt, uid = login(Account(username=a.user, password=a.password)) _LOG.info("[登录] uid=%d", uid) _ = enter_game_client(http, jwt, a.zone) key32, sid, rname, tkey, gport = login_gate_with_retry( a.zone, jwt, uid, a.name, PORTS) _LOG.info("[票据] sid=%d 角色=%s 端口=%d", sid, rname, gport) stop = [False] conn, world, updater = enter_world(a.zone, rname, tkey, sid, gport, uid, stop=stop) _LOG.info("[出生] %s @(%d,%d)", world.me.map_name, world.me.x, world.me.y) try: # ---- 传送(已在目标图则跳过); 失败重试(任务态冷却 3-5 分钟可解) ---- if world.me.map_name != a.map: for attempt in range(6): _LOG.info("[传送] %s → CS1011 @%s (第%d次)", world.me.map_name, a.act, attempt + 1) _ = conn.send_frame( 3031, 0, 0, 0, sid, (ROOT / "protocol_bot" / "assets" / "cs3031_full.bin") .read_bytes()) time.sleep(1.0) _ = conn.send_frame( 1011, 0, 0, 0, sid, f'{{"UserID":"999999999","Act":"@{a.act}","index":999999999}}' .encode()) deadline = time.time() + 25.0 while time.time() < deadline: for f in conn.recv_frames(0.2): updater.apply(world, f) if world.me.map_name == a.map: break if world.me.map_name == a.map: break _LOG.warning("[传送] 未受理(可能任务态冷却), 60s 后重试") time.sleep(60.0) if world.me.map_name != a.map: _LOG.error("[传送] 6 次未受理 — 退出") return 1 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") _LOG.info("[落地] %s @(%d,%d)", world.me.map_name, world.me.x, world.me.y) time.sleep(2.0) deadline = time.time() + 3.0 while time.time() < deadline: for f in conn.recv_frames(0.2): updater.apply(world, f) start = (world.me.x, world.me.y) _LOG.info("[落脚点] (%d,%d)", *start) # ---- 地图数据必须与实际所在图一致 ---- actual = world.me.map_name if actual != a.map: _LOG.error("[地图] 实际在 %s 与目标 %s 不符 — 退出", actual, a.map) return 1 if actual not in MAP_FILES: _LOG.error("[地图] %s 无 .map 数据 — 退出", actual) return 1 m = load_map(str(ROOT / "protocol_bot" / "assets" / "maps" / MAP_FILES[actual])) if not (0 <= start[0] < m.width and 0 <= start[1] < m.height): _LOG.error("[地图] 落脚点 (%d,%d) 超出 %s (%dx%d) — 退出", start[0], start[1], MAP_FILES[actual], m.width, m.height) return 1 _LOG.info("[地图数据] %s (%dx%d) 已加载", MAP_FILES[actual], m.width, m.height) # ---- 选目标: 图内 + 可达 + 最远 ---- t0 = time.time() dist = bfs_walkable(m, start) if a.target: tx, ty = (int(v) for v in a.target.split(",")) assert (tx, ty) in dist, f"目标 ({tx},{ty}) 不可达或不可走" goal = (tx, ty) else: goal = pick_farthest(dist, start) _LOG.info("[目标] (%d,%d) BFS步数=%d 可达域=%d格 选择耗时 %.2fs", goal[0], goal[1], dist[goal], len(dist), time.time() - t0) # ---- 规划(真客户端 A* 分段拼接) ---- t1 = time.time() path = plan_multiseg(m, start, goal) if path is None: _LOG.error("[规划] 无路 — 退出") return 1 legs = compress_run_legs(path) assert path_is_valid(m, path) _LOG.info("[规划] %d 步 → %d 跑腿, 规划耗时 %.2fs", len(path), len(legs), time.time() - t1) if a.dry_run: for i in range(0, len(path), 20): _LOG.info("[路径] %s", path[i:i + 20]) return 0 # ---- CS5013 START ---- _ = 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 = 0 pending = False pending_t = 0.0 li = 0 last_t = 0.0 while li < len(legs): now = time.time() for f in conn.recv_frames(0.02): before = len(world.events) updater.apply(world, f) for e in world.events: if e.kind == "self_move_confirmed": confirmed += 1 pending = False elif e.kind in ("pull_back", "anchor_echo"): rejected += 1 pending = False if len(world.events) > before: world.events.clear() if pending and now - pending_t >= PENDING_TIMEOUT_S: pending = False silent += 1 if pending: continue if now - last_t < RUN_STEP_S: continue # 位置可能被服务器纠正过 — 从真实位置接续最近腿 me = (world.me.x, world.me.y) while li < len(legs) and max(abs(legs[li][0] - me[0]), abs(legs[li][1] - me[1])) <= 2: li += 1 if li >= len(legs): break if max(abs(legs[li][0] - me[0]), abs(legs[li][1] - me[1])) > 4: # 偏离路径(拉回): 从当前位置重新规划剩余路 _LOG.info("[偏航] 我(%d,%d) vs 腿%s — 重规划", *me, legs[li]) rest = plan_multiseg(m, me, goal) if rest is None: _LOG.error("[重规划] 无路 — ABORT") break legs = compress_run_legs(rest) li = 0 continue nxt = legs[li] h = heading_of(me, nxt) dist2 = 2 if (abs(nxt[0] - me[0]) == 2 or abs(nxt[1] - me[1]) == 2) else 1 dxu = max(-1, min(1, nxt[0] - me[0])) dyu = max(-1, min(1, nxt[1] - me[1])) send_move(conn, sid, me[0], me[1], h, me[0] + dxu * dist2, me[1] + dyu * dist2, protocol_map_name(world.me.map_id, world.me.map_name)) _LOG.info("[跑] #%d (%d,%d) dir=%d → (%d,%d)", li + 1, me[0], me[1], h, nxt[0], nxt[1]) pending = True pending_t = now last_t = now li += 1 elapsed = time.time() - t_start _ = conn.send_frame(5013, 0, 0, 3, sid, b"") final = (world.me.x, world.me.y) arrived = max(abs(final[0] - goal[0]), abs(final[1] - goal[1])) <= ARRIVE_DIST + 1 _LOG.info("=" * 60) _LOG.info("[结果] %s | 起点(%d,%d) → 目标(%d,%d) 终点(%d,%d)", "✓到达" if arrived else "✗未达", start[0], start[1], goal[0], goal[1], final[0], final[1]) _LOG.info("[计时] 总耗时 %.1fs (%.1f分钟) | 计划 %d 腿 实跑 %d 腿", elapsed, elapsed / 60, len(legs), li) _LOG.info("[帧统计] SC1确认=%d SC28拒=%d 静默=%d", confirmed, rejected, silent) _LOG.info("[5013] FINISH 已发") return 0 if arrived else 2 finally: stop[0] = True conn.close() if __name__ == "__main__": sys.exit(main())