"""run_juece13_cave30_v2.py — 角色【新角色】→ 天然洞穴一层 → 杀 30 怪(全在洞内)。 v2 修复(v1 实测教训 2026-09-29 14:41): - v1 病根: 进洞后 allow_reroll=True → 视野无怪 45s 就重走 renwu 链"换落点" → 角色频繁被任务跳转切换落脚点, 从未认真跑图找怪 - v2 铁律: 【进洞后绝不再点任务跳转】allow_reroll=False + 覆写换落点分支 → 无怪 = 认真跑图(CS5284 问服务器怪位 + 新颖度锚点 + 90s 换远锚) - 进洞手段: 出生异图时 force_act 强制 renwu_mov,2(不读弹窗 — 任务指针 已推进到任务三, 弹窗返回矿区 act_move,1 会被 v1 误点) → 落地洞穴即停用 - 计数闸门: 只有 天然洞穴一层 内击杀计入 30; 异图击杀仅记日志 用法: uv run python scripts/run_juece13_cave30_v2.py -u longwu988 -p Longwu1988 \ --zone 14570166 --name 新角色 --kill 30 """ from __future__ import annotations import argparse import logging import sys import time from typing import override import requests from pbot.bot.kill import LAND_QUIET_S, KillBehavior from pbot.bot.runner import 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_cave_v2") _PORTS: tuple[int, ...] = (8021, 8061, 8001, 8041) CAVE_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 CaveStayKillBehavior(KillBehavior): """洞穴驻留杀怪: 进洞后永不点任务跳转; 无怪纯跑图找怪; 仅洞内计杀。""" def _in_cave(self, world: WorldMirror) -> bool: """是否在目标洞穴图。""" return world.me.map_name == CAVE_MAP @override def on_enter(self, conn: GatewayConnection, world: WorldMirror) -> None: """异图 → force_act 强制 renwu_mov,2 进洞(一次); 洞内 → 直接开打。""" self._world = world self._last_step_t = time.time() self._declare_guaji(conn) if self._bound is None: self._bound = ((world.me.x, world.me.y), 25) self._load_map_if_needed(world) if self._in_cave(world): self._force_act = True # 洞内: 后续所有 1011 都强指 renwu_mov,2 self._allow_reroll = False # ★ 铁律: 进洞后绝不允许"换落点重走链" _LOG.info("[洞穴] 出生即在 %s @(%d,%d) — 静置%.1fs后开打(任务跳转已禁用)", CAVE_MAP, world.me.x, world.me.y, LAND_QUIET_S) self._enter("fight") self._phase_t = time.time() + LAND_QUIET_S else: # 异图(等待图/比奇/矿区): force_act=True 强制 renwu_mov,2, # 不读任务弹窗(弹窗是任务三 act_move,1 → 会传矿区, v1 死循环根因) self._force_act = True self._allow_reroll = True self._renwu_act = "renwu_mov,2" _LOG.info("[进洞] 当前 %s @(%d,%d) — force_act 强制 renwu_mov,2 传送", world.me.map_name, world.me.x, world.me.y) self._enter("renwu_3031") @override def tick(self, conn: GatewayConnection, world: WorldMirror) -> object: """关键相位守卫: - fight/goto 恒在洞内(入口保证); 传送意外出洞 → 只允许走 renwu 链回洞 - allow_reroll=False 时框架的无怪分支自动走"持续跑图", 但其撞墙分支 _run_toward/_wander_step 里还有 4 处 `_enter(\"renwu_3031\")` 换落点 — 洞内全部覆写为"弃路径换锚点继续跑图" """ was_cave = self._in_cave(world) result = super().tick(conn, world) now_cave = self._in_cave(world) if was_cave and not now_cave: # 意外出洞(理论上不可能: 洞内不点跳转) — 高声报警 _LOG.warning("[!] 意外出洞: %s @(%d,%d) — 下 tick 走 renwu 链回洞", now_cave and world.me.map_name, world.me.x, world.me.y) # 洞内防呆: 若框架仍在洞内把相位拨回 renwu_3031(所有 reroll 入口), # 强制掰回 fight — 300ms 决策节拍下这层守卫是最后防线 if now_cave and self._phase in ("renwu_3031", "renwu_1011", "wait_map"): _LOG.warning("[防呆] 洞内相位=%s — 掰回 fight(任务跳转禁用)", self._phase) self._enter("fight") self._phase_t = time.time() + LAND_QUIET_S _ = conn return result @override def _run_toward(self, conn: GatewayConnection, world: WorldMirror, tx: int, ty: int) -> bool: """覆写: 洞内"累计弹回换落点重走链"分支 → 改为弃路径换锚点。""" me = world.me if self._in_cave(world): # 拦截父类会把相位拨回 renwu_3031 的两个入口(rej_total>=4 与 # A* 连续无路) — 先记下父类调用前的相位, 调用后掰正 before = self._phase result = super()._run_toward(conn, world, tx, ty) if self._phase in ("renwu_3031", "renwu_1011") and \ before not in ("renwu_3031", "renwu_1011"): _LOG.info("[洞内] 父类触发换落点 — 已拦截, 改为弃路径换锚点跑图") self._enter("fight") self._phase_t = time.time() # fight 相位无静置惩罚 self._path = None self._search_anchor = None self._waypoint = None return result return super()._run_toward(conn, world, tx, ty) @override def _wander_step(self, conn: GatewayConnection, world: WorldMirror, now: float) -> None: """覆写: 洞内漫游的"累计弹回换落点"分支 → 改为就地换向。""" if self._in_cave(world): before = self._phase super()._wander_step(conn, world, now) if self._phase in ("renwu_3031", "renwu_1011") and \ before not in ("renwu_3031", "renwu_1011"): _LOG.info("[洞内] 漫游触发换落点 — 已拦截, 就地继续跑图") self._enter("fight") self._phase_t = now self._path = None self._search_anchor = None return super()._wander_step(conn, world, now) @override def _current_target(self, world: WorldMirror) -> object: """非洞穴 → 永不锁定目标。""" if not self._in_cave(world): return None return super()._current_target(world) @override def _pick_target(self, world: WorldMirror) -> object: """非洞穴 → 永不选怪(tick 会走无怪分支 → 异图触发 renwu 链回洞)。""" if not self._in_cave(world): return None return super()._pick_target(world) @override def _credit_kill(self, world: WorldMirror, aid: int, name: str, signal: str) -> None: """计杀闸门: 只有洞内击杀计入 30/30。""" if not self._in_cave(world): _LOG.warning("[异图击杀不计] aid=%d %s 信号=%s @%s", aid, name, signal, world.me.map_name) _ = self._my_hits.pop(aid, None) return super()._credit_kill(world, aid, name, signal) 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("--kill", type=int, default=30) _ = ap.add_argument("--login-port", type=int, default=0) return ap.parse_args() def main() -> int: """登录 → 进游戏 → 喝药确认 → 洞穴驻留杀 30。""" a = parse_args() setup_logging() banner("[0] 启动: 区=%d 角色=%s 目标=%d 杀(仅限%s)", a.zone, a.name, a.kill, CAVE_MAP) 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) if state == "on": _LOG.info("[✓] 喝药保护已开启") else: _LOG.warning("[!] 喝药状态=%s(不阻断)", state) banner("[6] 洞穴驻留杀怪: 目标 %d @%s", a.kill, CAVE_MAP) kill = CaveStayKillBehavior( sid=sid, kill_target=a.kill, renwu_act="renwu_mov,2", allow_reroll=True, # 行为内部按图态动态覆写(洞内恒 False) ) duration = max(1200.0, a.kill * 150.0) _LOG.info("行为超时上限: %.0fs", duration) result = run(conn, world, updater, kill, duration_s=duration) banner("[7] 结果") _LOG.info("result=%s 洞内击杀=%d/%d 最终 map=%s @(%d,%d)", result.value, world.kills, a.kill, world.me.map_name, world.me.x, world.me.y) top = sorted(world.frame_counts.items(), key=lambda kv: -kv[1])[:15] _LOG.info("收帧Top15: %s", top) ok = world.kills >= a.kill and world.me.map_name == CAVE_MAP _LOG.info("[%s] %s", "✓" if ok else "✗", f"完成: {world.kills} 杀全部在 {CAVE_MAP}" if ok else "未达标") return 0 if ok else 1 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)