"""run_juece13_mine_farthest.py — 角色【新角色】→ 比奇矿区 → 走到最远点 → 静止5分钟。 流程: [1-4] 登录 → 选角 → 进游戏(与 kill30 系列同链) [5] 喝药保护确认 [6] 进矿区: 出生=盟重土城 → KillBehavior 传送链复用 (tp_goto: A* 走到传送员(342,338) → CS1010{UserID:10002,index:2} → SC643 菜单 → 连点 @jiuxian_move_1,1 → @jiuxian_move_1,1,1(比奇矿区) → @act_move,1(进入地图) — 2026-09-26 tp 实测) 出生=比奇矿区 → 直接进入下一阶段 [7] 可达域 BFS: 从落点出发, 找地图上最远可达点(d401.map 权威墙格) [8] AutoMoveBehavior: A* 路径 + CS5013 START + 3013 跑链走到最远点 [9] 静止不动(仅心跳+收帧), 5 分钟后自动退出 用法: uv run python scripts/run_juece13_mine_farthest.py -u longwu988 -p Longwu1988 \ --zone 14570166 --name 新角色 [--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 login_gate_with_retry from pbot.types import ZoneId from pbot.world.mirror import WorldMirror _LOG = logging.getLogger("juece13_far") _PORTS: tuple[int, ...] = (8021, 8061, 8001, 8041) MINE_MAP = "比奇矿区" TOWN_MAP = "盟重土城" RENWU_PHASES = frozenset({"renwu_3031", "renwu_1011", "wait_map"}) # 矿区 = d401.map 200x200(2026-09-29 定案; 旧 621/400x102 判定作废) MINE_W, MINE_H = 200, 200 _MINE_MAP_FILE = "assets/maps/d401.map" def setup_logging() -> None: """全量 INFO 日志。""" logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)-7s %(name)s: %(message)s", stream=sys.stdout, ) def banner(msg: str, *args: object) -> None: """阶段分隔条日志。""" _LOG.info("======== %s ========", msg % args if args else msg) class MineEnterBehavior(KillBehavior): """盟重→矿区传送 + 矿内不打怪不游走; 落地矿区即 DONE。 复用父类 tp_goto/tp_dialog 相位(走到传送员+三连点), 但: - 矿内(意外提前落地): 永不 renwu, 直接 DONE - renwu 链在盟重失败 → 父类自动走 tp_goto 兜底(v1/v3 实测可达) """ @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 == MINE_MAP: _LOG.info("[矿区] 出生已在 %s @(%d,%d)", MINE_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: # 盟重: 传送员就在 7 格外, 不走父类 A* 路径腿(_run_toward 在 # 盟重实测对直线路径反复弹回死循环) — 改用 1 格步进 + 滑行 # (Crystal 直走逻辑), 到达后父类 tp_goto 自然接管点对话 self._allow_reroll = False self._enter("tp_goto") self._slip_dir: int | None = 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 == MINE_MAP: _LOG.info("[矿区✓] 落地 @(%d,%d) — 传送阶段完成", world.me.x, world.me.y) return TickResult.DONE # 盟重 tp_goto: 1 格步进 + 撞墙滑行(Crystal 直走逻辑), 替代父类 # A* 路径腿(实测直线路径反复弹回死循环) if self._phase == "tp_goto" and world.me.known() \ and not world.me.map_name == MINE_MAP: return self._tick_tp_goto_slip(conn, world) result = super().tick(conn, world) if world.me.known() and world.me.map_name == MINE_MAP: _LOG.info("[矿区✓] 落地 @(%d,%d)(传送中检测) — 完成", world.me.x, world.me.y) return TickResult.DONE _ = conn return result def _tick_tp_goto_slip(self, conn: GatewayConnection, world: WorldMirror) -> TickResult: """1 格步进走向传送员(342,338); 撞墙顺/逆 45° 滑行; 到达转 tp_dialog。""" from pbot.bot.wander import DIRS as WDIRS from pbot.bot.kill import TELEPORTER_POS, TELEPORTER_AID 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_goto") self._slip_dir = None return super().tick(conn, world) # 父类到达分支: 发 1010 开菜单 if self._pending_rej: 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 # 事件消费: SC1 确认清滑行锁; 仅 pull_back=真弹回才触发滑行 # (anchor_echo 是锚点申报回显=位置确认, 非拒绝) for e in world.events: if e.kind == "self_move_confirmed": self._slip_dir = None elif e.kind == "pull_back": self._slip_blocked = 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 # 滑行中锁定方向直到 SC1 确认 elif self._slip_blocked: # 原版滑行: 顺 45° → 逆 45°, 轮流试 cand = next_dir(want) if self._slip_try % 2 == 0 \ else prev_dir(want) self._slip_try += 1 d = cand _LOG.info("[盟重·滑行] want=%d blocked → dir=%d", want, d) self._slip_blocked = False dx, dy = WDIRS[d] fx, fy = me.x + dx, me.y + dy 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=(fy << 16) | fx) self._last_step_t = time.time() self._pending_rej = True self._pending_rej_t = self._last_step_t _LOG.info("[盟重·走] (%d,%d) dir=%d → (%d,%d)", me.x, me.y, d, fx, fy) 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)) from pbot.bot.wander import DIRS as WDIRS return next((i for i, (ox, oy) in enumerate(WDIRS) if ox == dx and oy == dy), 4) def __init__(self, *args: object, **kwargs: object) -> None: super().__init__(*args, **kwargs) # type: ignore[arg-type] self._pending_rej = False self._pending_rej_t = 0.0 self._slip_blocked = False self._slip_try = 0 self._slip_dir = None @override def _pick_target(self, world: WorldMirror) -> object: """永不打怪。""" return None @override def _credit_kill(self, world: WorldMirror, aid: int, name: str, signal: str) -> None: """永不计杀(打怪被禁)。""" _ = world, aid, name, signal _D401_WALLS: set[tuple[int, int]] | None = None def _d401_walls() -> set[tuple[int, int]]: """加载 d401.map 墙格(权威地形, 2026-09-29 定案)。""" global _D401_WALLS if _D401_WALLS is None: _D401_WALLS = load_walls_from_map(_MINE_MAP_FILE) return _D401_WALLS def find_farthest(start: tuple[int, int]) -> tuple[tuple[int, int], int, bool]: """矿区可达域 BFS(d401 墙): 返回(最远可达点, 路径步数, 是否飞地落点)。 d401 定案后恢复真 BFS — 墙数据 100% 吻合实测, 最远点即主连通块 内 BFS 深度最大的格子, 不再需要"全图开阔区任意远点"的放水兜底。 """ walls = _d401_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] < MINE_W and 0 <= nxt[1] < MINE_H): continue if nxt in walls: continue visited.add(nxt) dq.append((nxt, d + 1)) return far, far_d, len(visited) < 50 _SC28_PATH = "assets/mine_sc28_walls.json" def _load_sc28_walls() -> set[tuple[int, int]]: """加载跨会话累积的 SC28 实测墙格(服务器权威地形)。""" import json from pathlib import Path try: data = json.loads(Path(_SC28_PATH).read_text()) if data.get("map") == MINE_MAP: return {(int(x), int(y)) for x, y in data.get("walls", [])} except (OSError, ValueError): pass return set() def _save_sc28_walls(walls: set[tuple[int, int]]) -> None: """回写 SC28 实测墙格(与旧数据合并)。""" import json from pathlib import Path old = _load_sc28_walls() merged = sorted(old | walls) Path(_SC28_PATH).write_text(json.dumps( {"map": MINE_MAP, "updated": time.strftime("%Y-%m-%dT%H:%M"), "count": len(merged), "walls": merged}, ensure_ascii=False, indent=0)) _LOG.info("[MirWalk] SC28 实测墙回写 %d 格(累计 %d) → %s", len(walls), len(merged), _SC28_PATH) def next_dir(d: int) -> int: """顺时针 45°(0上1右上2右3右下4下5左下6左7左上)。""" return (d + 1) % 8 def prev_dir(d: int) -> int: """逆时针 45°。""" return (d - 1) % 8 DIR_VECS: tuple[tuple[int, int], ...] = ( (0, -1), (1, -1), (1, 0), (1, 1), (0, 1), (-1, 1), (-1, 0), (-1, -1)) class MirWalkExecutor(Behavior): """传奇原版 AutoPath 忠实复刻(Suprcode/Crystal@0e315fe GameScene.cs:11715): 每 tick: 1. 消费事件: SC28 弹回格 → grid.mark_blocked(唯一的"学习"机制) 2. A* 重算 当前位置 → 终点(200×200 网格实测 0-5ms, 非负担) 3. path 为空 → 终点不可达, 等待(可能弹回学习后恢复) 4. 取 path[0] 方向: CanRun(方向 2 格全可走)→ 跑; CanWalk → 走 无路径游标/滑行锁/振荡计数 — 原版就是每帧重算, 墙数据保证 A* 每次给出 当前最优路, 弹回一次下次自动绕开(此前打转 = 游标卡死 + 补丁状态机)。 """ name = "mir_walk" def __init__(self, sid: int, dest: tuple[int, int], start_pt: tuple[int, int], allow_replan: bool = True, walls: set[tuple[int, int]] | None = None, learned_walls: set[tuple[int, int]] | None = None, map_name: str = "") -> None: del allow_replan, start_pt self._sid = sid self._dest = dest self._map_name = map_name self._died = False self.steps = 0 self.rejected = 0 self.slips = 0 self._pending = False self._pending_t = 0.0 self._last_t = 0.0 self._no_path_n = 0 self._replans = 0 self._last_target: tuple[int, int] | None = None self._learned: set[tuple[int, int]] = set() self._grid = LearnedGrid() # 墙源参数化: 默认 d401(矿区); 其他图传自己的墙集(如 d601 蜈蚣洞穴) # — .map 定案后墙数据即权威地形, MirWalkExecutor 本身地图无关 map_walls = walls if walls is not None else _d401_walls() self._map_walls = map_walls self._grid.mark_from_map(map_walls) _LOG.info("[MirWalk] 权威墙格 %d(硬边界)", len(map_walls)) # SC28 实测墙(动态补充): 仅采信权威墙内的格(滤超速弹回误标) — # 服务器真新增动态墙会经实时弹回重新学到 learned = (learned_walls or set()) & map_walls if learned: self._grid.mark_from_map(learned) self._learned = set(learned) _LOG.info("[MirWalk] SC28 实测墙 %d 格(已滤误标)", len(learned)) def on_enter(self, conn: GatewayConnection, world: WorldMirror) -> None: _ = conn self._last_t = time.time() _LOG.info("[MirWalk] 启动 (%d,%d) → (%d,%d) — 原版 AutoPath 模式", world.me.x, world.me.y, self._dest[0], self._dest[1]) def tick(self, conn: GatewayConnection, world: WorldMirror) -> TickResult: me = world.me if not me.known(): return TickResult.RUNNING # 换图守卫: 死亡回城/意外传送(蜈蚣洞穴 run3 实证: 25级被怪围殴致死 # → 回盟重, executor 拿盟重坐标在 d601 网格上乱标墙) — 立即退出 if self._map_name and me.map_name != self._map_name: _LOG.warning("[MirWalk✗] 换图 %s → %s(死亡回城/意外传送) — 退出 " "(发腿=%d 弹回=%d 滑行=%d)", self._map_name, me.map_name, self.steps, self.rejected, self.slips) return TickResult.DONE # 到达判定: 距终点 ≤2(跑步 2 格步长容差) if max(abs(self._dest[0] - me.x), abs(self._dest[1] - me.y)) <= 2: _LOG.info("[MirWalk✓] 到达 (%d,%d) 共 %d 腿 弹回 %d 滑行 %d", me.x, me.y, self.steps, self.rejected, self.slips) return TickResult.DONE # 事件消费: 仅真弹回(pull_back)标墙 — anchor_echo(p3=CS1 魔数)是 # 锚点申报回显=位置确认, 标墙会产生误标(2026-09-29 定案的 19 格 # 误标即部分来源于此) for e in world.events: if e.kind == "self_move_confirmed": self._pending = False elif e.kind == "pull_back": self.rejected += 1 self._pending = False if self._last_target is not None: self._grid.mark_blocked(self._last_target) self._learned.add(self._last_target) if self.rejected % 20 == 0: _save_sc28_walls(self._learned) _LOG.info("[MirWalk] 弹回 → 标墙 %s(累计学 %d)", self._last_target, len(self._learned)) world.events.clear() now = time.time() if self._pending: if now - self._pending_t < 1.0: return TickResult.RUNNING self._pending = False if now - self._last_t < 0.6: return TickResult.RUNNING # ★ 原版核心: 每 tick 重算 A*(当前位置 → 终点) path = astar((me.x, me.y), self._dest, self._grid) if not path: self._no_path_n += 1 # 无路自救: 只清 SC28 学习增量, 绝不动 d401 权威墙(清真墙=必撞) # (学习数据若含飞地误标, 豁免半径渐进扩大; d401 与服务器 98% 吻合, # 落点必在可达域 — 无路的根因几乎只剩学习误标) if self._no_path_n % 10 == 0: r = 15 + self._no_path_n // 10 * 10 cleared = 0 for dx in range(-r, r + 1): for dy in range(-r, r + 1): c = (me.x + dx, me.y + dy) if c in self._grid.blocked and c not in self._map_walls: self._grid.blocked.discard(c) cleared += 1 _LOG.info("[MirWalk] 无路#%d — 豁免半径 %d(清学习墙 %d, d401 权威墙保留)", self._no_path_n, r, cleared) if self._no_path_n >= 60: _LOG.warning("[MirWalk] 连续无路 60 次 — 放弃") return TickResult.DONE self._last_t = now return TickResult.RUNNING self._no_path_n = 0 # path[0] = 下一步落点(2 格步长 A*) nxt = path[1] if len(path) > 1 else path[0] want = self._dir_to(me.x, me.y, nxt[0], nxt[1]) cand = self._pick_dir(me.x, me.y, want) if cand is None: # 三向全墙(数据矛盾) — 把路径首格标墙, 下 tick A* 自动绕 self._grid.mark_blocked(nxt) self.slips += 1 _LOG.info("[MirWalk] 三向堵 → 标墙 %s 下 tick 绕行", nxt) self._last_t = now return TickResult.RUNNING run2 = self._can_run(me.x, me.y, cand) self._last_target = (me.x + DIR_VECS[cand][0] * (2 if run2 else 1), me.y + DIR_VECS[cand][1] * (2 if run2 else 1)) self._send_leg(conn, world, cand, run2) self.steps += 1 if cand != want: self.slips += 1 _LOG.info("[MirWalk] 滑行: 期望dir=%d 实际dir=%d", want, cand) return TickResult.RUNNING def _pick_dir(self, x: int, y: int, want: int) -> int | None: """原版 CanWalk(dir,out): 直向 → 顺 45° → 逆 45°。""" if self._can_run(x, y, want) or self._can_walk(x, y, want): return want for nd in (next_dir(want), prev_dir(want)): if self._can_walk(x, y, nd): return nd return None def _can_walk(self, x: int, y: int, d: int) -> bool: dx, dy = DIR_VECS[d] return (x + dx, y + dy) not in self._grid.blocked def _can_run(self, x: int, y: int, d: int) -> bool: """原版 CanRun: 方向上 2 格全部可走才跑。""" dx, dy = DIR_VECS[d] return all((x + dx * i, y + dy * i) not in self._grid.blocked for i in (1, 2)) def _send_leg(self, conn: GatewayConnection, world: WorldMirror, d: int, run: bool) -> None: me = world.me dx, dy = DIR_VECS[d] dist2 = 2 if run else 1 fx, fy = me.x + dx * dist2, me.y + dy * dist2 from pbot.game.messages import protocol_map_name mn = protocol_map_name(me.map_id, me.map_name) _ = conn.send_frame(3013 if run else 3011, 0, d, (me.y << 16) | me.x, self._sid, mn.encode(), pre=(fy << 16) | fx) self._pending = True self._pending_t = time.time() self._last_t = self._pending_t _LOG.info("[%s] (%d,%d) dir=%d → (%d,%d) #%d", "跑" if run else "走", me.x, me.y, d, fx, fy, self.steps + 1) def _dir_to(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, help="到达最远点后静止时长(秒)") _ = ap.add_argument("--pre-wait", type=float, default=60.0, help="落地矿区后静置时长(秒), 0=立即出发") _ = ap.add_argument("--login-port", type=int, default=0) return ap.parse_args() def main() -> int: """登录 → 进矿区 → 最远点规划 → 自动走过去 → 静止5分钟退出。""" a = parse_args() setup_logging() banner("[0] 启动: 区=%d 角色=%s 静止时长=%.0fs", a.zone, a.name, a.hold_seconds) 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) # ---- [6] 进矿区(如需要) ---- if world.me.map_name == MINE_MAP: banner("[6] 出生已在 %s @(%d,%d) — 跳过传送", MINE_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 = MineEnterBehavior(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 != MINE_MAP: _LOG.error("[✗] 未能进入矿区(看上方日志), 退出") return 1 # 落地静置(ACK 后服务器建视野) time.sleep(3.0) _ = pump(conn, world, updater, 2.0) # ---- [6.5] 落地静置(仅心跳+收帧) ---- 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] 可达域分析: 起点(%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])) # ---- [8] 传奇式走到最远点(A* 一次规划 + 撞墙滑行) ---- banner("[8] 传奇式寻路 → (%d,%d)", far[0], far[1]) executor = MirWalkExecutor( sid=sid, dest=far, start_pt=start_pt, allow_replan=not is_enclave, ) # 时间预算按实测弹回率放宽: 理想 0.75s/腿, 弹回重试约 2.5s/有效腿 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) if executor._learned: _save_sc28_walls(executor._learned) 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) # ---- [9] 静止不动 ---- banner("[9] 静止 %.0f 秒(仅心跳+收帧)", a.hold_seconds) deadline = time.time() + a.hold_seconds while time.time() < deadline: _ = pump(conn, world, updater, 0.5) remain = world.me.map_name _LOG.info("[静止结束] 最终 map=%s @(%d,%d)", remain, world.me.x, world.me.y) top = sorted(world.frame_counts.items(), key=lambda kv: -kv[1])[:15] _LOG.info("收帧Top15: %s", top) 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)