chuanqi/protocol_bot/main.py

"""主入口: 登录 → 选区 → 创角/复用 → 进游戏 → 开喝药 → 等待图走动 10 秒。

用法:
    uv run python main.py -u <账号> -p <密码> [--zone <区ID>] [--name <角色名>]
                          [--wander-seconds 10] [--login-port 0]

区 ID=0 时自动选服务器列表最新已开区; 登录端口 0 时按 8021/8061/8001/8041 轮询
(backup 实测: 38区=8041, 52区=8001, 端口随区变化)。
"""

import argparse
import base64
import json
import logging
import os
import re
import sys
import time

import requests

from pbot.bot.automove import AutoMoveBehavior
from pbot.bot.behavior import TickResult
from pbot.bot.kill import KillBehavior
from pbot.bot.runner import pump, run
from pbot.bot.unlock import UnlockBehavior
from pbot.bot.wander import WanderBehavior
from pbot.config import Account, RoleSpec
from pbot.errors import PbotError
from pbot.game.heal import enable_auto_heal
from pbot.game.session import enter_world
from pbot.login.box_client import enter_game_client, login
from pbot.transport.gateway import (
    GatewayConnection,
    create_role_needed,
    delete_role,
    login_gate_with_retry,
)
from pbot.types import ZoneId
from pbot.world.mirror import WorldMirror
from pbot.world.update import WorldUpdater, as_dict, to_bool, to_float, to_int, typed_json

SERVERLIST_URL = "https://list-box.dhsf.xqhuyu.com/box_serverlist/206923.txt"
DEFAULT_PORTS: tuple[int, ...] = (8021, 8061, 8001, 8041)


def setup_logging() -> None:
    """根日志: 时间+级别+模块, UTF-8 输出。"""
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
        stream=sys.stdout,
    )


def pick_latest_zone() -> int:
    """服务器列表里 state=1 且已开服的最新区(backup 实测逻辑)。"""
    r = requests.get(SERVERLIST_URL, timeout=20)
    m = re.search(r"\{.*\}", r.text, re.DOTALL)
    if not m:
        raise PbotError(detail="区列表拉取失败")
    data = typed_json(json.loads(m.group(0)))  # pyright: ignore[reportAny]
    best: tuple[float, str, int] | None = None

    def walk(o: object) -> None:
        nonlocal best
        if isinstance(o, dict):
            d = as_dict(typed_json(o))  # pyright: ignore[reportUnknownArgumentType]
            sid = to_int(d.get("serverid"))
            state = to_int(d.get("state"))
            ot = to_float(d.get("open_time"))
            if sid > 0 and state == 1 and 0 < ot <= time.time():
                name = str(d.get("list_name") or d.get("server_name") or "")
                if best is None or ot > best[0]:
                    best = (ot, name, sid)
            for v in d.values():
                walk(v)
        elif isinstance(o, list):
            items: list[object] = list(o)  # pyright: ignore[reportUnknownArgumentType]
            for v in items:
                walk(v)

    walk(data)
    result = best
    if result is None:
        raise PbotError(detail="无已开新区")
    logging.getLogger("main").info("自动选区: %s (id=%d, 开服=%s)",
                                   result[1], result[2], result[0])
    return result[2]


def parse_args() -> dict[str, str | int | float]:
    """命令行参数(强类型化 dict)。"""
    ap = argparse.ArgumentParser()
    _ = ap.add_argument("-u", "--user", required=True)
    _ = ap.add_argument("-p", "--password", required=True)
    _ = ap.add_argument("--zone", type=int, default=0, help="0=自动选最新已开区")
    _ = ap.add_argument("--name", default="", help="角色名(空=随机萌新XXXX)")
    _ = ap.add_argument("--login-port", type=int, default=0, help="0=自动轮询")
    _ = ap.add_argument("--wander-seconds", type=float, default=10.0)
    _ = ap.add_argument("--kill", type=int, default=0,
                        help="杀怪数达标线(>0 时进入打怪流程: 传比奇→杀N怪)")
    _ = ap.add_argument("--run", action="store_true",
                        help="走动用 CS3013 跑(跳2格/0.55s)替代 CS3011 走")
    _ = ap.add_argument("--reset", action="store_true", help="先删角重创(新手任务重置)")
    _ = ap.add_argument("--jwt", default="",
                        help="外部注入 JWT(pwd 型 token); 跳过盒子登录与 EnterGameClient")
    _ = ap.add_argument("--wait-duration", action="store_true",
                        help="走动前等待 game_duration 首跳(≤5min, §八.11 打分假设)")
    _ = ap.add_argument("--idle", action="store_true",
                        help="创角+喝药后原地不动(只心跳+收帧, 不走动不退出)")
    _ = ap.add_argument("--goto", default="",
                        help="自动寻路目标 x,y(当前图内; 复刻客户端自动寻路 5013 协议)")
    a: object = ap.parse_args()
    user = str(a.user)  # pyright: ignore[reportAny]
    password = str(a.password)  # pyright: ignore[reportAny]
    zone_v = to_int(a.zone)  # pyright: ignore[reportAny]
    name = str(a.name)  # pyright: ignore[reportAny]
    login_port = to_int(a.login_port)  # pyright: ignore[reportAny]
    wander_seconds = to_float(a.wander_seconds)  # pyright: ignore[reportAny]
    kill_target = to_int(a.kill)  # pyright: ignore[reportAny]
    ext_jwt = str(a.jwt)  # pyright: ignore[reportAny]
    return {
        "user": user, "password": password, "zone": zone_v, "name": name,
        "login_port": login_port, "wander_seconds": wander_seconds,
        "kill": kill_target,
        "reset": bool(a.reset),  # pyright: ignore[reportAny]
        "jwt": ext_jwt,
        "wait_duration": bool(a.wait_duration),  # pyright: ignore[reportAny]
        "run": bool(a.run),  # pyright: ignore[reportAny]
        "idle": bool(a.idle),  # pyright: ignore[reportAny]
        "goto": str(a.goto),  # pyright: ignore[reportAny]
    }


def _enter_game(http: requests.Session, zone: ZoneId, jwt: str, uid: int,
                role_name: str, ports: tuple[int, ...], stop: list[bool],
                ext_mode: bool = False, wait_duration: bool = False,
            ) -> tuple[GatewayConnection, WorldMirror, WorldUpdater, int]:
    """阶段 2-4: 选区上报 → 登录服票据 → 游戏服就绪。"""
    log = logging.getLogger("main")

    if ext_mode:
        log.info("======== [2/5] 跳过 EnterGameClient (pwd 型 token 不走 PC 网关) ========")
    else:
        log.info("======== [2/5] EnterGameClient zone=%d ========", zone)
        for attempt in range(4):
            try:
                _ = enter_game_client(http, jwt, zone)
                break
            except Exception:
                if attempt == 3:
                    raise
                time.sleep(12)

    log.info("======== [3/5] 登录服选角/创角 角色=%s ========", role_name)
    key32, sid, real_name, ticket_key, game_port = login_gate_with_retry(
        zone, jwt, uid, role_name, ports)
    log.info("[票据] key32=%s sid=%d 角色=%s ticket=%s 游戏端口=%d",
             key32.decode(errors="replace"), sid, real_name, ticket_key, game_port)

    log.info("======== [4/5] 进游戏(角色=%s) ========", real_name)
    conn, world, updater = enter_world(zone, real_name, ticket_key, sid,
                                       game_port, uid, stop=stop)

    # 真客户端在角色属性初始化后立即上报 game_login(PlayerPropertyProxy:620
    # → DataTrackProxy:OnEnterGame → SDK) — bot 补上这条腿, 否则服务器视
    # 本会话为"无壳层幽灵会话"(§八.4 假设), 走动帧全拒。
    if ext_mode:
        try:
            from pbot.login.sdk_report import SdkReporter
            sdk = SdkReporter()
            if sdk.up_device_login(box_game_id=206923, channel_id="110585"):
                sdk.send_game_login(
                    uid=uid, zone_id=zone, zone_name=str(zone),
                    role_id=ticket_key.split("/", 1)[-1], role_name=real_name,
                    role_level=1, job_name="战士", job_id=0)
                globals()["_sdk_reporter"] = sdk
                if wait_duration:
                    log.info("======== [4.5/5] 等待 game_duration 首跳(≤5min) ========")
                    _ = sdk.wait_first_duration(timeout=360.0)
        except Exception as e:  # noqa: BLE001 - 上报失败不阻断走动实验
            log.warning("[SDK] 上报链失败(实验继续): %s", e)
    return conn, world, updater, sid


def _jwt_uid(jwt: str) -> int:
    """从 JWT payload 的 sub 提取 uid(pwd 型 token 的 sub 即 CS2001 uid)。"""
    try:
        payload_b64 = jwt.split(".")[1]
        payload = as_dict(typed_json(json.loads(  # pyright: ignore[reportAny]
            base64.urlsafe_b64decode(payload_b64 + "=="))))
        return to_int(payload.get("sub"))
    except (IndexError, ValueError):
        raise PbotError(detail="--jwt 无效: payload 不是合法 JSON(base64)") from None


def main() -> int:
    """全链路: 每阶段失败即退出并输出原因日志。"""
    cfg = parse_args()
    user = str(cfg["user"])
    password = str(cfg["password"])
    zone_arg = to_int(cfg["zone"])
    name_arg = str(cfg["name"])
    login_port = to_int(cfg["login_port"])
    wander_seconds = to_float(cfg["wander_seconds"])
    kill_target = to_int(cfg.get("kill"))
    do_reset = to_bool(cfg.get("reset"))
    ext_jwt = str(cfg.get("jwt") or "")
    wait_duration_flag = bool(cfg.get("wait_duration"))
    use_run = bool(cfg.get("run"))
    idle_mode = bool(cfg.get("idle"))
    goto_raw = str(cfg.get("goto") or "")
    goto: tuple[int, int] | None = None
    if "," in goto_raw:
        xs, ys = goto_raw.split(",", 1)
        if xs.strip().lstrip("+-").isdigit() and ys.strip().lstrip("+-").isdigit():
            goto = (int(xs), int(ys))
        else:
            raise PbotError(detail=f"--goto 无效: {goto_raw!r} (格式 x,y)")
    elif goto_raw:
        raise PbotError(detail=f"--goto 无效: {goto_raw!r} (格式 x,y)")

    setup_logging()
    log = logging.getLogger("main")
    account = Account(username=user, password=password)
    zone = ZoneId(pick_latest_zone() if zone_arg == 0 else zone_arg)
    if name_arg:
        role_name = create_role_needed(RoleSpec(name=name_arg))
    else:
        # 用户没传 → 随机纯中文名(无数字)
        import random as _rnd
        cn_pool = ("玖扬测试", "新角色", "复古玩家", "江湖游侠", "风云再起",
                   "独行侠客", "剑指天下", "无名英雄", "传奇新篇", "傲世狂刀",
                   "凌风踏雪", "落日长河", "踏雪无痕", "一剑封喉", "笑傲江湖")
        role_name = _rnd.choice(cn_pool)  # noqa: S311
        log.info("[随机纯中文名] 选用: %s", role_name)
    ports = (login_port,) if login_port else DEFAULT_PORTS

    ext_mode = bool(ext_jwt)
    if ext_mode:
        # 优先实时提取盒子最新 token(消除签发时间差), 失败再用显式 --jwt
        try:
            from pbot.login.sdk_report import fetch_fresh_box_token
            jwt, uid = fetch_fresh_box_token()
            log.info("======== [1/5] 盒子实时 token 注入 uid=%d ========", uid)
        except (OSError, ValueError, KeyError) as e:
            jwt = ext_jwt
            uid = _jwt_uid(jwt)
            log.info("======== [1/5] 外部 JWT 注入 uid=%d (盒子token不可用: %s) ========",
                     uid, e)
    else:
        # 1. 盒子登录
        log.info("======== [1/5] 盒子登录 %s ========", account.username)
        jwt, uid = login(account)

    if do_reset:
        log.info("======== [1.5/5] 删角重创 %s ========", role_name)
        deleted = delete_role(zone, jwt, uid, role_name, ports)
        log.info("[删角] %s", "成功" if deleted else "失败(继续)")
        time.sleep(10)  # 服务器落库延迟(实测: 立即重创会拿不到票据)

    http = requests.Session()
    http.trust_env = False
    http.verify = False

    # 心跳: enter_world 内部在 CS20 后立即启动(backup 实测: 不跑心跳不下发出生帧)
    stop: list[bool] = [False]
    conn, world, updater, sid = _enter_game(http, zone, jwt, uid, role_name,
                                            ports, stop, ext_mode=ext_mode,
                                            wait_duration=wait_duration_flag)

    try:
        # 5. 喝药保护
        log.info("======== [5/5] 喝药保护 ========")
        state = enable_auto_heal(conn, sid, want_on=True, world=world)
        if state == "on":
            log.info("[✓] 喝药保护已开启(服务端 but_1 确认)")
        elif state == "off":
            log.warning("[!] 喝药保护为关(切换未生效?)")
        else:
            log.warning("[!] 未捕获到开关状态面板(zlib 面板未到达)")

        # 6. 打怪(--kill>0) / 自动寻路(--goto x,y) / 解锁+wander / idle(原地不动)
        if goto is not None:
            log.info("======== 自动寻路: 目标(%d,%d) 当前 %s @(%d,%d) ========",
                     goto[0], goto[1], world.me.map_name, world.me.x, world.me.y)
            auto_move = AutoMoveBehavior(sid=sid, dest_x=goto[0], dest_y=goto[1])
            goto_duration = float(os.environ.get("PBOT_GOTO_DURATION_S", "0")) or 900.0
            goto_result = run(conn, world, updater, auto_move,
                              duration_s=goto_duration)
            log.info("[寻路结束] result=%s 位置=(%d,%d) 地图=%s",
                     goto_result.value, world.me.x, world.me.y, world.me.map_name)
        elif idle_mode:
            # idle: 创角+喝药后原地不动, 只心跳+收帧, 不发任何行动帧
            log.info("======== idle: 原地不动(创角+喝药已完成) ========")
            log.info("  当前 %s @(%d,%d) actor=%s",
                     world.me.map_name, world.me.x, world.me.y, world.me.actor_key)
            idle_until = float(os.environ.get("PBOT_IDLE_DURATION_S", "0")) or 1e18
            log.info("  idle 持续 %.0fs (0=永久); Ctrl-C 或 pkill -f 'uv run python main.py' 退出",
                     min(idle_until, 86400 * 365))
            idle_deadline = time.time() + idle_until
            while time.time() < idle_deadline:
                _ = pump(conn, world, updater, 0.5)
        elif kill_target > 0:
            renwu_act = os.environ.get("PBOT_RENWU_ACT", "act_move,1")
            no_reroll = os.environ.get("PBOT_NO_REROLL") == "1"
            wp_raw = os.environ.get("PBOT_WAYPOINT", "")
            wp: tuple[int, int] | None = None
            if "," in wp_raw:
                xs, ys = wp_raw.split(",", 1)
                if xs.strip().lstrip("+-").isdigit() \
                        and ys.strip().lstrip("+-").isdigit():
                    wp = (int(xs), int(ys))
            la_raw = os.environ.get("PBOT_LAND_AT", "")
            la: tuple[int, int] | None = None
            if "," in la_raw:
                xs, ys = la_raw.split(",", 1)
                if xs.strip().lstrip("+-").isdigit() \
                        and ys.strip().lstrip("+-").isdigit():
                    la = (int(xs), int(ys))
            # PBOT_BOUND=x,y,r 限制 _search_anchor 范围(用户视野内)
            bd_raw = os.environ.get("PBOT_BOUND", "")
            bound: tuple[tuple[int, int], int] | None = None
            if "," in bd_raw:
                parts = bd_raw.split(",")
                if len(parts) == 3 \
                        and all(p.strip().lstrip("+-").isdigit() for p in parts):
                    bound = ((int(parts[0]), int(parts[1])), int(parts[2]))
            hdr = ("======== 打怪: 目标 %d 只 @%s, land_at=%s, waypoint=%s, "
                   "bound=%s, 当前 %s @(%d,%d) ========")
            log.info(hdr, kill_target, renwu_act, la, wp, bound,
                     world.me.map_name, world.me.x, world.me.y)
            kill = KillBehavior(sid=sid, kill_target=kill_target,
                                renwu_act=renwu_act,
                                allow_reroll=not no_reroll, waypoint=wp,
                                land_at=la, bound=bound,
                                force_act=os.environ.get("PBOT_FORCE_ACT") == "1",
                                land_only=os.environ.get("PBOT_LAND_ONLY") == "1",
                                roam=os.environ.get("PBOT_ROAM") == "1")
            kill_duration_s = float(
                os.environ.get("PBOT_KILL_DURATION_S", "0")) or \
                max(600.0, kill_target * 120.0)
            kill_result = run(conn, world, updater, kill,
                              duration_s=kill_duration_s)
            log.info("[打怪结束] result=%s kills=%d 位置=(%d,%d) 地图=%s",
                     kill_result.value, world.kills, world.me.x, world.me.y,
                     world.me.map_name)
            if kill_result is TickResult.DONE and world.kills >= kill_target:
                # 达标后保持原地: 仅心跳(线程已有) + 收帧, 不发任何行动帧
                idle_s = float(os.environ.get("PBOT_KILL_IDLE_S", "0")) or 1e18
                log.info("======== 达标待机(原地不动, 仅心跳) %.0fs; Ctrl-C 或 pkill 退出 ========",
                         min(idle_s, 86400 * 365))
                idle_until = time.time() + idle_s
                while time.time() < idle_until:
                    _ = pump(conn, world, updater, 0.5)
                log.info("[待机结束] 最终位置=(%d,%d) 地图=%s",
                         world.me.x, world.me.y, world.me.map_name)
        else:
            # 解锁(滞留带/异图 → renwu_mov/NPC链推进到等待图可走区)
            log.info("======== 解锁: 当前 %s @(%d,%d) ========",
                     world.me.map_name, world.me.x, world.me.y)
            unlock = UnlockBehavior(sid=sid)
            unlock_result = run(conn, world, updater, unlock,
                                duration_s=max(60.0, wander_seconds * 6))
            log.info("[解锁] result=%s 位置=(%d,%d) 地图=%s",
                     unlock_result.value, world.me.x, world.me.y, world.me.map_name)

            # 7. 等待图走动/跑动
            mode = "跑动" if use_run else "走动"
            log.info("======== 等待图%s %.0f 秒 @(%d,%d) map=%s ========",
                     mode, wander_seconds, world.me.x, world.me.y, world.me.map_name)
            behavior = WanderBehavior(sid=sid, use_run=use_run, random_roam=True)
            result = run(conn, world, updater, behavior, duration_s=wander_seconds)
            log.info("[结束] result=%s pos=(%d,%d) 地图=%s",
                     result.value, world.me.x, world.me.y, world.me.map_name)
        top = sorted(world.frame_counts.items(), key=lambda kv: -kv[1])[:10]
        log.info("收帧Top10: %s", top)
        return 0
    finally:
        stop[0] = True
        conn.close()


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