chuanqi/protocol_bot/scripts/test_tp_minimal.py

"""test_tp_minimal.py — 传送最小变量实验: 字节级复刻真客户端传送链。

2026-09-29 抓包对照分析(cap_tp.jsonl)得出的 3 个差异, 本脚本全部修正:
  1. 1011 body 键序: {"UserID":..,"index":..,"Act":..}(真) ≠ bot 旧序(UserID,Act,index)
  2. 传送窗口零噪音: 不发 3199/9102/5286(真客户端窗口内只有 1010/1011×2/ACK+心跳)
  3. 时序对齐: 1010 后 2.0s 点矿区, 再 1.15s 点进入地图, 再 20ms 发 ACK
     ACK body='比奇矿区'(中文, 与真客户端一致)

成功判据: 发完 ACK 后 10s 内 world.me.map_name 变为 比奇矿区。
"""
from __future__ import annotations

import json
import logging
import sys
import time

import requests

from pbot.config import Account
from pbot.game.messages import protocol_map_name
from pbot.game.session import enter_world, apply_batch
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.bot.wander import DIRS

_LOG = logging.getLogger("tp_min")

PORTS = (8021, 8061, 8001, 8041)
TP_AID = "10002"
TP_POS = (342, 338)


def js(objects: dict) -> bytes:
    """按传入键序序列化(真客户端字节级复刻的关键)。"""
    return json.dumps(objects, separators=(",", ":")).encode()


def main() -> int:
    logging.basicConfig(level=logging.INFO,
                        format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
                        stream=sys.stdout)
    http = requests.Session()
    http.trust_env = False
    http.verify = False

    jwt, uid = login(Account(username="longwu988", password="Longwu1988"))
    _ = enter_game_client(http, jwt, ZoneId(14570166))
    _k, sid, name, ticket, gport = login_gate_with_retry(
        ZoneId(14570166), jwt, uid, "新角色", PORTS)
    stop = [False]
    conn, world, upd = enter_world(ZoneId(14570166), name, ticket, sid,
                                   gport, uid, stop=stop)
    _LOG.info("出生 %s @(%d,%d)", world.me.map_name, world.me.x, world.me.y)
    if world.me.map_name == "比奇矿区":
        _LOG.info("[i] 已在矿区, 无需传送(实验目的达成?)")
        stop[0] = True
        conn.close()
        return 0
    if world.me.map_name != "盟重土城":
        _LOG.error("[✗] 出生 %s 非盟重, 实验终止", world.me.map_name)
        stop[0] = True
        conn.close()
        return 1

    try:
        # ---- 走到传送员(真客户端: 9 步 3013 后点菜单) ----
        _LOG.info("[走] → 传送员%s", TP_POS)
        for _ in range(60):
            me = world.me
            if max(abs(TP_POS[0] - me.x), abs(TP_POS[1] - me.y)) <= 1:
                break
            dx = max(-1, min(1, TP_POS[0] - me.x))
            dy = max(-1, min(1, TP_POS[1] - me.y))
            h = next((i for i, (ox, oy) in enumerate(DIRS)
                      if ox == dx and oy == dy), 4)
            mn = protocol_map_name(me.map_id, me.map_name)
            conn.send_frame(3013, 0, h, (me.y << 16) | me.x, sid,
                            mn.encode(), pre=((me.y + dy * 2) << 16) | (me.x + dx * 2))
            time.sleep(0.55)
            apply_batch(world, upd, conn.recv_frames(0.4))
        _LOG.info("[走✓] @(%d,%d) — 静置1s(对齐真客户端节奏)",
                  world.me.x, world.me.y)
        time.sleep(1.0)
        apply_batch(world, upd, conn.recv_frames(0.8))

        # ---- [帧1] CS1010 开菜单(字节级同真客户端) ----
        conn.send_frame(1010, 0, 0, 0, sid, js({"UserID": TP_AID, "index": 2}))
        _LOG.info("[帧1] CS1010 开菜单 — 等 2.0s(真客户端含一次心跳)")
        time.sleep(2.0)
        apply_batch(world, upd, conn.recv_frames(1.5))
        _LOG.info("菜单到达: %s", world.npc_dialog is not None)

        # ---- [帧2] CS1011 点【比奇矿区】(键序: UserID,index,Act) ----
        body2 = js({"UserID": "999999999", "index": 999999999,
                    "Act": "@jiuxian_move_1,1,1"})
        conn.send_frame(1011, 0, 0, 0, sid, body2)
        _LOG.info("[帧2] CS1011 @jiuxian_move_1,1,1 (blen=%d, 键序=UserID,index,Act)",
                  len(body2))
        time.sleep(1.15)
        apply_batch(world, upd, conn.recv_frames(0.8))

        # ---- [帧3] CS1011 点【进入地图】 ----
        body3 = js({"UserID": "999999999", "index": 999999999,
                    "Act": "@act_move,1"})
        conn.send_frame(1011, 0, 0, 0, sid, body3)
        _LOG.info("[帧3] CS1011 @act_move,1 (blen=%d)", len(body3))

        # ---- [帧4/5] ACK 5012+5027('比奇矿区') — 20ms 内 ----
        time.sleep(0.02)
        conn.send_frame(5012, 0, 0, 0, sid, "比奇矿区".encode())
        conn.send_frame(5027, 0, 0, 0, sid, "比奇矿区".encode())
        _LOG.info("[帧4/5] ACK 5012+5027('比奇矿区') — 等落地 12s")

        # ---- 判定 ----
        for i in range(24):
            time.sleep(0.5)
            apply_batch(world, upd, conn.recv_frames(0.4))
            if world.me.map_name == "比奇矿区":
                _LOG.info("[✓✓✓] 传送成功! %s @(%d,%d) (耗时 %.1fs)",
                          world.me.map_name, world.me.x, world.me.y,
                          (i + 1) * 0.5)
                stop[0] = True
                conn.close()
                return 0
            if (i + 1) % 4 == 0:
                _LOG.info("  ... %.1fs 仍在 %s", (i + 1) * 0.5,
                          world.me.map_name)
        _LOG.error("[✗] 12s 未落地 — 字节级复刻仍静默, 剩余变量: 会话冷却")
        stop[0] = True
        conn.close()
        return 1
    except KeyboardInterrupt:
        stop[0] = True
        conn.close()
        return 130


if __name__ == "__main__":
    sys.exit(main())