chuanqi/protocol_bot/scripts/diag_5012_monster_probe.py

#!/usr/bin/env python3
"""diag_5012_monster_probe.py — CS5012 小地图怪物请求探针.

目的: 验证本服 CS5012(body=协议图名)→SC5012 响应格式(全图怪坐标 JSON).
背景: 本服不下发怪物 SC10, 静止怪对协议端不可见(SC11 只广播移动怪) —
客户端 MiniMapProxy:RequestMonsters 是合法主动查询通道.

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

import argparse
import logging
import sys
import time

sys.path.insert(0, "scripts")
from run_juece13_mine_farthest import (  # noqa: E402
    MINE_MAP, MineEnterBehavior, setup_logging, banner,
)
from run_juece13_wugong_autofight import AutoFightBehavior  # noqa: E402

import requests

from pbot.bot.runner import pump, run
from pbot.config import Account
from pbot.errors import PbotError
from pbot.game.messages import protocol_map_name
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

_LOG = logging.getLogger("probe5012")


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("--seconds", type=float, default=60.0)
    a = ap.parse_args()
    setup_logging()
    banner("[0] 5012 怪物雷达探针: 区=%d 角色=%s", a.zone, a.name)

    http = requests.Session()
    http.trust_env = False
    http.verify = False
    jwt, uid = login(Account(username=a.user, password=a.password))
    _ = enter_game_client(http, jwt, ZoneId(a.zone))
    key32, sid, real_name, ticket_key, game_port = login_gate_with_retry(
        ZoneId(a.zone), jwt, uid, a.name, (8021, 8061, 8001, 8041))
    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) uid=%s", world.me.map_name,
              world.me.x, world.me.y, world.me.uid)
    try:
        if world.me.map_name != MINE_MAP:
            banner("[1] 传送比奇矿区")
            bh = MineEnterBehavior(sid=sid, kill_target=0)
            r = run(conn, world, updater, bh, duration_s=300.0)
            if world.me.map_name != MINE_MAP:
                _LOG.error("[✗] 传送失败 result=%s", r.value)
                return 1
            time.sleep(3.0)
            _ = pump(conn, world, updater, 2.0)
        mn = protocol_map_name(world.me.map_id, world.me.map_name)
        _LOG.info("[发送] CS5012 body=%r", mn)
        _ = conn.send_frame(5012, 0, 0, 0, 0, mn.encode())

        deadline = time.time() + a.seconds
        n5012 = 0
        while time.time() < deadline:
            frames = pump(conn, world, updater, 0.5)
            for f in frames:
                if f.msg_id == 5012:
                    n5012 += 1
                    _LOG.info("[SC5012] recog=%d p1=%d p2=%d body=%r",
                              f.recog, f.p1, f.p2, f.body[:400])
        _LOG.info("[结束] 5012 响应 %d 帧", n5012)
        return 0 if n5012 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)