chuanqi/protocol_bot/scripts/run_juece13_wugong_farthest.py

#!/usr/bin/env python3
"""run_juece13_wugong_farthest.py — 角色【新角色】→ 蜈蚣洞穴(地牢一层东) → 走到最远点 → 静止。

验证 2026-09-30 凌晨映射修正(地牢一层东=d601)的实跑效果:
  登录 → 盟重传送员(342,338) → @jiuxian_move_1,1,2(蜈蚣洞穴) + @act_move,1
  → ACK 5012+5027(body='地牢一层东') → 落地 → d601 墙 BFS 最远点
  → MirWalkExecutor 跑图 → 静止 → 退出

用法:
    uv run python scripts/run_juece13_wugong_farthest.py -u longwu988 -p Longwu1988 \
        --zone 14570166 --name 新角色 [--pre-wait 60] [--hold-seconds 300]
"""
from __future__ import annotations

import argparse
import logging
import math
import sys
import time
from collections import deque
from typing import override

import requests

from pbot.bot.behavior import Behavior, TickResult
from pbot.bot.kill import KillBehavior
from pbot.bot.pathfind import LearnedGrid, astar, load_walls_from_map
from pbot.bot.runner import pump, run
from pbot.config import Account
from pbot.errors import PbotError
from pbot.game.session import enter_world
from pbot.login.box_client import enter_game_client, login
from pbot.transport.gateway import GatewayConnection, login_gate_with_retry
from pbot.types import ZoneId
from pbot.world.mirror import WorldMirror

sys.path.insert(0, "scripts")
from run_juece13_mine_farthest import (  # noqa: E402
    DIR_VECS, MirWalkExecutor, next_dir, prev_dir, setup_logging, banner,
)

_LOG = logging.getLogger("juece13_wugong")

_PORTS: tuple[int, ...] = (8021, 8061, 8001, 8041)
CAVE_MAP = "地牢一层东"
TOWN_MAP = "盟重土城"
CAVE_W, CAVE_H = 250, 250
_CAVE_MAP_FILE = "assets/maps/d601.map"
# 真机抓包(cap_live2, 21040 帧): 地牢一层东移动帧 body = 显示名(文件ID=中文名)
_ACK_BODY = "地牢一层东"
_TELEPORT_ACTS: tuple[str, ...] = ("@jiuxian_move_1,1,2", "@act_move,1")

SC28_PATH = "assets/cave_sc28_walls.json"

_CAVE_WALLS: set[tuple[int, int]] | None = None


def _cave_walls() -> set[tuple[int, int]]:
    """加载 d601.map 墙格(权威地形, sourid 实抓定案)."""
    global _CAVE_WALLS
    if _CAVE_WALLS is None:
        _CAVE_WALLS = load_walls_from_map(_CAVE_MAP_FILE)
    return _CAVE_WALLS


def find_farthest(start: tuple[int, int]) -> tuple[tuple[int, int], int, bool]:
    """d601 墙 BFS 最远可达点."""
    walls = _cave_walls()
    visited: set[tuple[int, int]] = {start}
    dq = deque([(start, 0)])
    far, far_d = start, 0
    while dq:
        (x, y), d = dq.popleft()
        if d > far_d:
            far, far_d = (x, y), d
        for dx, dy in DIR_VECS:
            nxt = (x + dx, y + dy)
            if nxt in visited or not (0 <= nxt[0] < CAVE_W and 0 <= nxt[1] < CAVE_H):
                continue
            if nxt in walls:
                continue
            visited.add(nxt)
            dq.append((nxt, d + 1))
    return far, far_d, len(visited) < 50


class CaveEnterBehavior(KillBehavior):
    """盟重→蜈蚣洞穴传送(菜单第 2 项); 落地地牢一层东即 DONE.

    复用 mine_farthest.MineEnterBehavior 的盟重 1 格步进+滑行相位,
    仅替换传送 Act 序列与落地判定图名/ACK body(父类 tp_dialog 为硬编码
    比奇矿区序列, 这里接管 _tick_tp_dialog 相位自点).
    """

    def __init__(self, *args: object, **kwargs: object) -> None:
        super().__init__(*args, **kwargs)
        self._cave_act_i = 0
        self._cave_phase_t = 0.0
        self._pending_rej = False
        self._pending_rej_t = 0.0
        self._slip_dir: int | None = None
        self._slip_try = 0

    @override
    def on_enter(self, conn: GatewayConnection, world: WorldMirror) -> None:
        self._world = world
        self._last_step_t = time.time()
        if self._bound is None:
            self._bound = ((world.me.x, world.me.y), 25)
        self._load_map_if_needed(world)
        if world.me.map_name == CAVE_MAP:
            _LOG.info("[洞穴] 出生已在 %s @(%d,%d)", CAVE_MAP,
                      world.me.x, world.me.y)
            self._phase = "done"
            return
        self._declare_guaji(conn)
        if world.me.map_name == TOWN_MAP or "土城" in world.me.map_name:
            self._allow_reroll = False
            self._enter("tp_goto")
            self._slip_dir = None
            _LOG.info("[盟重] @(%d,%d) — 1格步进+滑行走向传送员", world.me.x, world.me.y)
        else:
            self._force_act = False
            self._allow_reroll = True
            _LOG.info("[%s] @(%d,%d) — 走默认任务链/传送员兜底",
                      world.me.map_name, world.me.x, world.me.y)
            self._enter("renwu_3031")

    @override
    def tick(self, conn: GatewayConnection, world: WorldMirror) -> object:
        if world.me.known() and world.me.map_name == CAVE_MAP:
            _LOG.info("[洞穴✓] 落地 %s @(%d,%d)", CAVE_MAP, world.me.x, world.me.y)
            return TickResult.DONE
        if self._phase == "tp_goto" and world.me.known():
            return self._tick_tp_goto_cave(conn, world)
        # run16b 定案: _enter("tp_dialog") 后父类主 tick 的 tp_dialog 分支
        # 会抢先接管(读 kill.py 矿区 TELEPORT_ACTS+ACK='比奇矿区') — 菜单
        # 点击其实已定目的地, 但洞穴/沃玛第 1 步不同, 必须全程自管该相位
        if self._phase == "tp_dialog":
            return self._tick_tp_dialog_cave(conn, world, time.time())
        result = super().tick(conn, world)
        if world.me.known() and world.me.map_name == CAVE_MAP:
            _LOG.info("[洞穴✓] 落地 %s @(%d,%d)(传送中检测)", CAVE_MAP,
                      world.me.x, world.me.y)
            return TickResult.DONE
        _ = conn
        return result

    def _tick_tp_goto_cave(self, conn: GatewayConnection,
                           world: WorldMirror) -> TickResult:
        """盟重 1 格步进+滑行到传送员 → 到达转自点洞穴菜单."""
        from pbot.bot.wander import DIRS as WDIRS
        from pbot.bot.kill import TELEPORTER_POS
        me = world.me
        now = time.time()
        tx, ty = TELEPORTER_POS
        if max(abs(tx - me.x), abs(ty - me.y)) <= 2:
            _LOG.info("[盟重✓] 已到传送员旁 @(%d,%d) — 开菜单点蜈蚣洞穴",
                      me.x, me.y)
            self._enter("tp_dialog")
            self._cave_act_i = 0
            self._cave_phase_t = now
            return self._tick_tp_dialog_cave(conn, world, now)
        if getattr(self, "_pending_rej", False):
            if now - self._pending_rej_t < 0.9:
                return TickResult.RUNNING
            self._pending_rej = False
        if now - self._last_step_t < 0.9:
            return TickResult.RUNNING
        for e in world.events:
            if e.kind == "self_move_confirmed":
                self._slip_dir = None
            elif e.kind == "pull_back":
                self._pending_rej = True
        world.events.clear()
        want = self._dir8(me.x, me.y, tx, ty)
        d = want
        if self._slip_dir is not None:
            d = self._slip_dir
        elif self._pending_rej:
            cand = next_dir(want) if getattr(self, "_slip_try", 0) % 2 == 0 \
                else prev_dir(want)
            self._slip_try = getattr(self, "_slip_try", 0) + 1
            d = cand
            _LOG.info("[盟重·滑行] want=%d → dir=%d", want, d)
        self._pending_rej = False
        dx, dy = WDIRS[d]
        from pbot.game.messages import protocol_map_name
        mn = protocol_map_name(me.map_id, me.map_name)
        _ = conn.send_frame(3011, 0, d, (me.y << 16) | me.x, self._sid,
                            mn.encode(), pre=((me.y + dy) << 16) | (me.x + dx))
        self._pending_rej = True
        self._pending_rej_t = time.time()
        self._last_step_t = self._pending_rej_t
        return TickResult.RUNNING

    def _tick_tp_dialog_cave(self, conn: GatewayConnection,
                             world: WorldMirror, now: float) -> TickResult:
        """点 @jiuxian_move_1,1,2(蜈蚣洞穴) → @act_move,1 → ACK(中文图名)."""
        dlg = world.npc_dialog
        if dlg is None:
            if now - self._cave_phase_t > 15.0:
                self._cave_phase_t = now
                import json as _json
                from pbot.bot.kill import TELEPORTER_AID
                body = _json.dumps({"UserID": TELEPORTER_AID, "index": 2},
                                   separators=(",", ":")).encode()
                _ = conn.send_frame(1010, 0, 0, 0, self._sid, body)
            return TickResult.RUNNING
        if self._cave_act_i < len(_TELEPORT_ACTS):
            if now - self._cave_phase_t >= (1.2 if self._cave_act_i else 0.8):
                act = _TELEPORT_ACTS[self._cave_act_i]
                self._cave_act_i += 1
                import json as _json
                body = _json.dumps({"UserID": dlg.get("UserID", "999999999"),
                                    "index": dlg.get("index", 999999999),
                                    "Act": act},
                                   separators=(",", ":")).encode()
                _ = conn.send_frame(1011, 0, 0, 0, self._sid, body)
                _LOG.info("[传送] 点 %s (%d/%d)", act, self._cave_act_i,
                          len(_TELEPORT_ACTS))
                self._cave_phase_t = now
                if self._cave_act_i == len(_TELEPORT_ACTS):
                    _ = conn.send_frame(5012, 0, 0, 0, self._sid,
                                        _ACK_BODY.encode())
                    _ = conn.send_frame(5027, 0, 0, 0, self._sid,
                                        _ACK_BODY.encode())
                    _LOG.info("[传送] ACK 5012+5027('%s')已发 — 等落地", _ACK_BODY)
            return TickResult.RUNNING
        return TickResult.RUNNING

    def _dir8(self, x: int, y: int, tx: int, ty: int) -> int:
        dx = max(-1, min(1, tx - x))
        dy = max(-1, min(1, ty - y))
        return next((i for i, (ox, oy) in enumerate(DIR_VECS)
                     if ox == dx and oy == dy), 4)


def parse_args() -> argparse.Namespace:
    """命令行."""
    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("--hold-seconds", type=float, default=300.0)
    _ = ap.add_argument("--pre-wait", type=float, default=60.0)
    _ = ap.add_argument("--login-port", type=int, default=0)
    return ap.parse_args()


def main() -> int:
    """登录 → 进蜈蚣洞穴 → 最远点跑图 → 静止 → 退出."""
    a = parse_args()
    setup_logging()
    banner("[0] 启动: 区=%d 角色=%s 静止=%.0fs 预跑=%.0fs",
           a.zone, a.name, a.hold_seconds, a.pre_wait)

    http = requests.Session()
    http.trust_env = False
    http.verify = False

    banner("[1] 盒子登录")
    jwt, uid = login(Account(username=a.user, password=a.password))
    _LOG.info("[✓] uid=%d", uid)

    banner("[2] EnterGameClient zone=%d", a.zone)
    for attempt in range(4):
        try:
            _ = enter_game_client(http, jwt, ZoneId(a.zone))
            _LOG.info("[✓] 上报成功")
            break
        except Exception as e:
            if attempt == 3:
                raise
            _LOG.warning("[!] 第%d次失败(%s), 12s 后重试", attempt + 1, e)
            time.sleep(12)

    banner("[3] 登录服选角: %s", a.name)
    ports = (a.login_port,) if a.login_port else _PORTS
    key32, sid, real_name, ticket_key, game_port = login_gate_with_retry(
        ZoneId(a.zone), jwt, uid, a.name, ports)
    _LOG.info("[票据] key32=%s sid=%d 角色=%s 游戏端口=%d",
              key32.decode(errors="replace"), sid, real_name, game_port)

    banner("[4] 进游戏")
    stop: list[bool] = [False]
    conn, world, updater = enter_world(ZoneId(a.zone), real_name, ticket_key,
                                       sid, game_port, uid, stop=stop)
    _LOG.info("[✓] 出生 map=%s @(%d,%d)", world.me.map_name, world.me.x,
              world.me.y)

    try:
        banner("[5] 喝药保护确认")
        from pbot.game.heal import enable_auto_heal
        state = enable_auto_heal(conn, sid, want_on=True, world=world)
        _LOG.info("[%s] 喝药状态=%s", "✓" if state == "on" else "!", state)

        if world.me.map_name == CAVE_MAP:
            banner("[6] 出生已在 %s @(%d,%d) — 跳过传送",
                   CAVE_MAP, world.me.x, world.me.y)
        else:
            banner("[6] 进蜈蚣洞穴: 当前 %s @(%d,%d)",
                   world.me.map_name, world.me.x, world.me.y)
            enter_bh = CaveEnterBehavior(sid=sid, kill_target=0)
            r = run(conn, world, updater, enter_bh, duration_s=300.0)
            _LOG.info("[传送结果] result=%s map=%s @(%d,%d)",
                      r.value, world.me.map_name, world.me.x, world.me.y)
            if world.me.map_name != CAVE_MAP:
                _LOG.error("[✗] 未能进入蜈蚣洞穴(看上方日志), 退出")
                return 1
            time.sleep(3.0)
            _ = pump(conn, world, updater, 2.0)

        if a.pre_wait > 0:
            banner("[6.5] 落地静置 %.0f 秒再出发", a.pre_wait)
            deadline = time.time() + a.pre_wait
            while time.time() < deadline:
                _ = pump(conn, world, updater, 0.5)

        start_pt = (world.me.x, world.me.y)
        banner("[7] d601 可达域分析: 起点(%d,%d)", *start_pt)
        far, d_steps, is_enclave = find_farthest(start_pt)
        _LOG.info("[最远点] (%d,%d) 路径距离 %d 步(直线 %.0f)",
                  far[0], far[1], d_steps,
                  math.hypot(far[0] - start_pt[0], far[1] - start_pt[1]))

        banner("[8] MirWalk 跑图 → (%d,%d)", far[0], far[1])
        cave_walls = _cave_walls()
        executor = MirWalkExecutor(sid=sid, dest=far, start_pt=start_pt,
                                    walls=cave_walls, map_name=CAVE_MAP)
        est_s = d_steps * 0.75 * 4 + 180.0
        _LOG.info("预计跑动上限 %.0fs(弹回冗余×4)", est_s)
        r = run(conn, world, updater, executor, duration_s=est_s)
        final_d = max(abs(far[0] - world.me.x), abs(far[1] - world.me.y))
        _LOG.info("[寻路结束] result=%s 位置=(%d,%d) 距目标 %d 格 map=%s "
                  "(发腿=%d 弹回=%d 滑行=%d)",
                  r.value, world.me.x, world.me.y, final_d,
                  world.me.map_name, executor.steps, executor.rejected,
                  executor.slips)

        banner("[9] 静止 %.0f 秒", a.hold_seconds)
        deadline = time.time() + a.hold_seconds
        while time.time() < deadline:
            _ = pump(conn, world, updater, 0.5)
        _LOG.info("[静止结束] 最终 map=%s @(%d,%d)", world.me.map_name,
                  world.me.x, world.me.y)
        return 0
    finally:
        stop[0] = True
        conn.close()
        _LOG.info("连接已关闭")


if __name__ == "__main__":
    try:
        sys.exit(main())
    except PbotError as e:
        _LOG.error("失败: %s", e.detail)
        sys.exit(1)