chuanqi/pipeline_bot/scripts/run_juece13_mine30.py

"""run_juece13_mine30.py — 角色【新角色】→ 盟重传送员 → 比奇矿区 → 杀 30 怪。

v3 定案(基于 v1/v2 日志复盘 2026-09-29):
- v1 已实证: 任务三 @act_move,1 传送矿区 45 次落地成功; 矿区有怪可杀
  (56052/57083/57082 三杀在矿区); 用户拍板: 就去矿区杀 30
- v1 病根: 落地后撞墙(矿区地形与 A* 墙格不完全匹配)触发"换落点重走链"
  → 频繁任务跳转, 而非认真跑图
- v3 铁律:
  1) 进图: 盟重→传送员(342,338)→CS1010{UserID:10002,index:2}→SC643 菜单
     →连点 @jiuxian_move_1,1 → @jiuxian_move_1,1,1(比奇矿区) → @act_move,1
     (kill.py tp_dialog 相位既有实现, v1 实测可达)
  2) 落地后【绝不再点任务跳转】: allow_reroll=False + 覆写全部 renwu 入口
     → 无怪 = 跑图(CS5284 问服务器怪位 + 新颖度锚点 90s 轮换)
     → 撞墙 = 换锚点/换向继续跑, 永不 renwu
  3) 计数: 仅 比奇矿区 内击杀计入 30; 异图击杀仅记日志
  4) 出生在矿区(上轮登出点)则直接开打

用法:
    uv run python scripts/run_juece13_mine30.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.login.zone_ports import resolve_zone_gateway
from pbot.types import ZoneId
from pbot.world.mirror import WorldMirror

_LOG = logging.getLogger("juece13_mine")

MINE_MAP = "比奇矿区"
RENWU_PHASES = frozenset({"renwu_3031", "renwu_1011", "wait_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 MineStayKillBehavior(KillBehavior):
    """矿区驻留杀怪: 进图后永不点任务跳转; 无怪纯跑图; 仅矿内计杀。"""

    def _in_mine(self, world: WorldMirror) -> bool:
        """是否在比奇矿区。"""
        return world.me.map_name == MINE_MAP

    def _ban_renwu(self, why: str) -> None:
        """把被父类拨向 renwu/wait_map 的相位掰回 fight(矿区铁律)。"""
        if self._phase in RENWU_PHASES:
            _LOG.info("[矿区铁律] %s 触发相位=%s — 拦截, 掰回 fight 继续跑图",
                      why, self._phase)
            self._enter("fight")
            self._phase_t = time.time()
            self._path = None
            self._search_anchor = None
            self._waypoint = None

    @override
    def on_enter(self, conn: GatewayConnection, world: WorldMirror) -> None:
        """矿区出生 → 直接开打; 异图 → 父类链路(盟重: tp_goto 传送员三连;
        等待图: 任务链 renwu_mov,1 进比奇 → 传送员)。
        """
        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_mine(world):
            self._allow_reroll = False   # ★ 矿内: 永不允许换落点重走链
            _LOG.info("[矿区] 出生即在 %s @(%d,%d) — 静置%.1fs后开打(任务跳转已禁用)",
                      MINE_MAP, world.me.x, world.me.y, LAND_QUIET_S)
            self._enter("fight")
            self._phase_t = time.time() + LAND_QUIET_S
        else:
            _LOG.info("[进图] 出生 %s @(%d,%d) — 走父类链路: 盟重→传送员三连点→矿区",
                      world.me.map_name, world.me.x, world.me.y)
            # 父类 on_enter 的相位决策: 盟重(_is_town)→renwu_3031, 其 renwu
            # 失败后 tp_goto 传送员兜底 — 保留该链路(它最终走传送员)。
            if "土城" in world.me.map_name or world.me.map_id == "3":
                self._enter("renwu_3031")  # 立即失败 → tp_goto 兜底(v1 实证)
                self._phase_t = time.time() - 999  # 让 wait_map 立即超时走 tp
            else:
                self._enter("renwu_3031")  # 等待图: 任务链进比奇(v1 实证有效)

    @override
    def tick(self, conn: GatewayConnection, world: WorldMirror) -> object:
        """矿区守卫: 任何把相位拨向 renwu 的分支都被掰回 fight。"""
        result = super().tick(conn, world)
        if self._in_mine(world):
            self._ban_renwu("tick 守卫")
        _ = conn
        return result

    @override
    def _run_toward(self, conn: GatewayConnection, world: WorldMirror,
                    tx: int, ty: int) -> bool:
        """矿内覆写: "累计弹回换落点重走链" → 弃路径换锚点继续跑图。"""
        if self._in_mine(world):
            before = self._phase
            result = super()._run_toward(conn, world, tx, ty)
            self._ban_renwu(f"_run_toward(before={before})")
            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_mine(world):
            before = self._phase
            super()._wander_step(conn, world, now)
            self._ban_renwu(f"_wander_step(before={before})")
            return
        super()._wander_step(conn, world, now)

    @override
    def _tick_no_monster(self, conn: GatewayConnection, world: WorldMirror,
                         now: float) -> None:
        """矿内无怪: 保留 CS5284 问怪位 + 跑图; renwu 重走分支被 tick 守卫拦截。"""
        super()._tick_no_monster(conn, world, now)
        if self._in_mine(world):
            self._ban_renwu("_tick_no_monster")

    @override
    def _current_target(self, world: WorldMirror) -> object:
        """非矿区 → 永不锁定目标。"""
        if not self._in_mine(world):
            return None
        return super()._current_target(world)

    @override
    def _pick_target(self, world: WorldMirror) -> object:
        """非矿区 → 永不选怪。"""
        if not self._in_mine(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_mine(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, MINE_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)
    gw_host, conn_port, real_port = (("", a.login_port, a.login_port)
        if a.login_port else resolve_zone_gateway(a.zone))
    ports = (conn_port,)
    key32, sid, real_name, ticket_key, game_port =  login_gate_with_retry(
        ZoneId(a.zone), jwt, uid, a.name, ports, host=gw_host, real_port=real_port)
    _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, host=gw_host)
    _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, MINE_MAP)
        kill = MineStayKillBehavior(
            sid=sid,
            kill_target=a.kill,
            renwu_act="renwu_mov,1",  # 仅等待图出生时用于进比奇; 矿内被禁
            allow_reroll=True,        # 行为内部矿内恒禁 reroll
        )
        duration = max(1500.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 == MINE_MAP
        _LOG.info("[%s] %s", "✓" if ok else "✗",
                  f"完成: {world.kills} 杀全部在 {MINE_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)