chuanqi/protocol_bot/scripts/auto_pathfind.py

"""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())