chuanqi/pipeline_bot/scripts/wear_test.py

"""wear_test.py — CS1003 穿装备最小变量实验。

假设(cap42/43 离线分析定案): bot 旧会话"冷登录链+CS20 platform=2+引导对"
是缝合特征, 服务器判会话异常 → 装备动作恒 616。
对照: 真客户端冷登录 = platform=1 + 无引导对 + init18→心跳→openmianban→1003。

实验矩阵(默认 A; --variant 切换):
  A: platform=1 + 无引导 + openmianban,7→1003   (真客户端黄金时序复刻)
  B: platform=2 + 无引导 + openmianban,7→1003   (分离 platform 变量)
  C: platform=1 + 无引导 + 裸穿(无 3199 前置)    (分离 openmianban 变量)

全程逐帧日志(含 SC-1 子帧展开), 判定: SC-1 含 615/203=成功, 616=失败。
"""
from __future__ import annotations

import argparse
import json
import logging
import sys
import time
import zlib
from pathlib import Path
from typing import Final

ROOT: Final[Path] = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from pbot.codec.bitbuf import decode6  # noqa: E402
from pbot.codec.frame import Frame, FrameSplitter  # noqa: E402
from pbot.config import Account  # noqa: E402
from pbot.game.session import apply_batch, enter_world  # noqa: E402
from pbot.login.box_client import enter_game_client, login  # noqa: E402
from pbot.login.zone_ports import fetch_zone_ports  # noqa: E402
from pbot.transport.gateway import login_gate_with_retry  # noqa: E402
from pbot.types import ZoneId  # noqa: E402
from pbot.world.mirror import WorldMirror  # noqa: E402
from pbot.world.update import WorldUpdater  # noqa: E402

_LOG = logging.getLogger("wear_test")
ZONE_40: Final[int] = 14606328
CHAR_NAME: Final[str] = "俞丁兰"
USER: Final[str] = "longwu988"
PASSWORD: Final[str] = "Longwu1988"
PORTS_FALLBACK: Final[tuple[int, ...]] = (8041, 8043, 8081)
_WOODEN_SWORD_INDEX: Final[int] = 50000  # SC201 定案: 木剑 Index=50000


def split_sub(raw: bytes) -> list[Frame]:
    frames = FrameSplitter().feed(raw)
    if frames:
        return frames
    try:
        return FrameSplitter().feed(decode6(raw))
    except Exception:  # noqa: BLE001
        return []


def describe_frame(f: Frame, direction: str) -> str:
    body_preview = f.body[:60].decode("utf-8", "replace") if f.body else ""
    line = (f"{direction}{f.msg_id:<6} tail={f.tail:<4} recog={f.recog:<11} "
            f"p1={f.p1} p2={f.p2} p3={f.p3} {body_preview!r}")
    if f.msg_id == -1 and f.body:
        try:
            subs = split_sub(zlib.decompress(f.body))
            ids = [s.msg_id for s in subs]
            line += f" 子帧={ids}"
            for s in subs:
                if s.msg_id in (200, 201, 202, 203, 52, 615, 616, 619, 622):
                    try:
                        txt = s.body.decode("utf-8", "replace")
                    except UnicodeDecodeError:
                        try:
                            txt = decode6(s.body).decode("utf-8", "replace")
                        except Exception:  # noqa: BLE001
                            txt = repr(s.body[:60])
                    line += f"\n      └ SC{s.msg_id}: {txt[:200]}"
        except zlib.error:
            pass
    return line


def pump_logged(conn, world: WorldMirror, updater: WorldUpdater,
                seconds: float, tag: str = "") -> list[Frame]:
    """收帧 + 逐帧日志 + 应用镜像。返回原始帧(含 -1)。"""
    deadline = time.time() + seconds
    collected: list[Frame] = []
    while time.time() < deadline:
        frames = conn.recv_frames(max(0.1, deadline - time.time()))
        for f in frames:
            _LOG.info("[%s] ← %s", tag, describe_frame(f, "SC"))
        collected += frames
        apply_batch(world, updater, frames)
    return collected


def wear_and_check(conn, world: WorldMirror, updater: WorldUpdater, sid: int,
                   item_name: str, tail_hint: str, makeindex: int = 0) -> str:
    """发 CS1003 并监听 3s, 返回 SUCCESS / FAIL_616 / NO_RESP。

    pre=makeindex — cap_hoe_success2 pcap 9字段头定案: 真客户端把
    MakeIndex 放在头部 pre 字段(CS20 的 pre=900000 同位), recog=1 是
    常量标志。pre=0 → 服务器 GetItem 查无 → 恒 616(此前全部失败的根因)。
    """
    used = conn.send_frame(1003, 1, 0, 0, sid, item_name.encode("utf-8"),
                           pre=makeindex)
    _LOG.info("[%s] → CS1003 %r tail=%d pre=%d(recog=1 p1=0 p2=0 p3=%d)",
              tail_hint, item_name, used, makeindex, sid)
    frames = pump_logged(conn, world, updater, 3.0, tag=f"{tail_hint}:resp")
    result = "NO_RESP"
    for f in frames:
        if f.msg_id == 616:
            result = "FAIL_616"
            break
        if f.msg_id == -1 and f.body:
            try:
                subs = split_sub(zlib.decompress(f.body))
            except zlib.error:
                continue
            ids = [s.msg_id for s in subs]
            if 203 in ids or 615 in ids:
                result = "SUCCESS"
                break
            if 616 in ids:
                result = "FAIL_616(batch)"
                break
    _LOG.info("[%s] 判定: %s", tail_hint, result)
    return result


def fetch_bag(conn, world: WorldMirror, updater: WorldUpdater, sid: int) -> list:
    """CS81 拉背包 → SC-1 子帧 SC201 JSON。"""
    _ = conn.send_frame(81, 0, 0, 0, sid, b"")
    _LOG.info("[bag] → CS81 拉背包")
    deadline = time.time() + 3.0
    while time.time() < deadline:
        frames = conn.recv_frames(max(0.1, deadline - time.time()))
        for f in frames:
            _LOG.info("[bag] ← %s", describe_frame(f, "SC"))
        apply_batch(world, updater, frames)
        for f in frames:
            if f.msg_id == -1 and f.body:
                try:
                    subs = split_sub(zlib.decompress(f.body))
                except zlib.error:
                    continue
                for s in subs:
                    if s.msg_id == 201:
                        # SC201 body 为 6bit 文本态(JSON), 双层解码
                        for cand in (s.body,):
                            try:
                                return json.loads(cand.decode("utf-8"))
                            except (UnicodeDecodeError, json.JSONDecodeError):
                                pass
                        try:
                            return json.loads(decode6(s.body).decode("utf-8", "replace"))
                        except json.JSONDecodeError:
                            _LOG.warning("[bag] SC201 解码失败")
    return []


def main() -> int:
    ap = argparse.ArgumentParser()
    _ = ap.add_argument("--variant", choices=["A", "B", "C"], default="A")
    _ = ap.add_argument("--item", default="木剑")
    _ = ap.add_argument("--takeoff-mk", type=int, default=0,
                        help="卸载分支用的 makeindex(装备栏 SC200 未接入前的手动通道)")
    _ = ap.add_argument("--zone", type=int, default=ZONE_40)
    _ = ap.add_argument("--name", default=CHAR_NAME)
    _ = ap.add_argument("-u", "--user", default=USER)
    _ = ap.add_argument("-p", "--password", default=PASSWORD)
    a = ap.parse_args()
    logging.basicConfig(level=logging.INFO,
                        format="%(asctime)s %(levelname)s %(message)s",
                        stream=sys.stdout)

    _LOG.info("=== 实验变体 %s: item=%s zone=%d char=%s ===",
              a.variant, a.item, a.zone, a.name)

    jwt, uid = login(Account(username=a.user, password=a.password))
    _LOG.info("[1] uid=%d", uid)
    _ = enter_game_client(_http_session(), jwt, ZoneId(a.zone))
    _LOG.info("[2] 选区上报 OK")

    ports = fetch_zone_ports().get(a.zone)
    port_tuple = (ports["port"],) if ports and ports["port"] else PORTS_FALLBACK
    _LOG.info("[3] 登录端口: %s", port_tuple)

    key32, sid, real, ticket, gport = login_gate_with_retry(
        ZoneId(a.zone), jwt, uid, a.name, port_tuple)
    _LOG.info("[4] sid=%d 角色=%s 票据OK 游戏端口=%d", sid, real, gport)

    platform = 1 if a.variant in ("A", "C") else 2
    stop: list[bool] = [False]
    conn, world, updater = enter_world(
        ZoneId(a.zone), real, ticket, sid, gport, uid,
        stop=stop, platform=platform, with_guide=False)
    _LOG.info("[5] 进世界 OK platform=%d 出生 map=%s @(%d,%d)",
              platform, world.me.map_name, world.me.x, world.me.y)

    try:
        bag = fetch_bag(conn, world, updater, sid)
        if bag:
            brief = [{"mk": it.get("makeindex"), "Index": it.get("Index"),
                      "where": it.get("where")} for it in bag]
            _LOG.info("[6] 背包 %d 件: %s", len(bag), json.dumps(brief, ensure_ascii=False))
        else:
            _LOG.warning("[6] 背包拉取失败(空)")

        if a.variant in ("A", "B"):
            _LOG.info("[6] 节奏: +2.2s → 3199(自动tail) → +2.5s → 1003")
            pump_logged(conn, world, updater, 2.2, tag="gap1")

            used = conn.send_frame(
                3199, 0, 0, 0, sid,
                b'{"act":"@openmianban,7","subid":1,"index":104}')
            _LOG.info("[7] → CS3199 @openmianban,7 tail=%d", used)
            got = pump_logged(conn, world, updater, 2.5, tag="panel")
            sc1109 = [f for f in got if f.msg_id == 1109]
            _LOG.info("[7.1] SC1109(面板确认) %s",
                      "收到✓ 会话正常" if sc1109 else "未收(会话异常)")
        else:
            _LOG.info("[6] 变体C: 裸穿, 仅静置")
            pump_logged(conn, world, updater, 3.0, tag="gap1")

        mk = 0
        for it in bag:
            if it.get("Index") == _WOODEN_SWORD_INDEX:
                mk = int(it.get("makeindex", 0))
        _LOG.info("[6.5] %s 在包中=%s makeindex=%d",
                  a.item, "是" if mk else "否(已穿上或不在)", mk)

        if mk:
            result = wear_and_check(conn, world, updater, sid, a.item,
                                    tail_hint=f"variant{a.variant}", makeindex=mk)
            _LOG.info("=== 最终结果: %s ===", result)
            if result.startswith("SUCCESS"):
                _LOG.info("[8] 等 2s 观察装备栏/属性推送")
                _ = pump_logged(conn, world, updater, 2.0, tag="after")
                _LOG.info("[9] 3s 后卸下(验证 CS1004 同构: pre=makeindex)")
                time.sleep(3.0)
                used = conn.send_frame(1004, 1, 0, 0, sid,
                                       a.item.encode("utf-8"), pre=mk)
                _LOG.info("[9] → CS1004 %r tail=%d pre=%d", a.item, used, mk)
                frames = pump_logged(conn, world, updater, 3.0, tag="takeoff")
                off = "NO_RESP"
                for f in frames:
                    if f.msg_id == -1 and f.body:
                        try:
                            subs = split_sub(zlib.decompress(f.body))
                        except zlib.error:
                            continue
                        ids = [s.msg_id for s in subs]
                        if 619 in ids or 622 in ids:
                            off = "TAKEOFF_SUCCESS"
                            break
                        if 620 in ids:
                            off = "TAKEOFF_FAIL_620"
                            break
                    if f.msg_id in (619, 622):
                        off = "TAKEOFF_SUCCESS"
                        break
                    if f.msg_id == 620:
                        off = "TAKEOFF_FAIL_620"
                        break
                _LOG.info("[9] 卸下判定: %s", off)
        else:
            _LOG.info("=== 跳过穿戴(物品不在包) — 直接验证卸载 ===")
            used = conn.send_frame(1004, 1, 0, 0, sid,
                                   a.item.encode("utf-8"), pre=a.takeoff_mk)
            _LOG.info("[9] → CS1004 %r tail=%d pre=%d", a.item, used, a.takeoff_mk)
            frames = pump_logged(conn, world, updater, 3.0, tag="takeoff0")
            for f in frames:
                if f.msg_id == -1 and f.body:
                    try:
                        ids = [s.msg_id for s in split_sub(zlib.decompress(f.body))]
                    except zlib.error:
                        continue
                    if 619 in ids or 620 in ids or 622 in ids:
                        _LOG.info("[9] 卸载应答子帧: %s", ids)
                        break
                elif f.msg_id in (619, 620, 622):
                    _LOG.info("[9] 卸载应答: SC%d", f.msg_id)
                    break
    finally:
        stop[0] = True
        time.sleep(0.3)
        conn.close()
    return 0




def _http_session():
    import requests
    s = requests.Session()
    s.trust_env = False
    s.verify = False
    return s


if __name__ == "__main__":
    raise SystemExit(main())