chuanqi/apilot/host.py

"""宿主: HTTP 登录 + 网关连接 + 创角驱动。复用 protocol_bot 已验证资产。"""
from __future__ import annotations

import json
import os
import random
import socket
import sys
import time
from typing import Any

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.join(ROOT, "protocol_bot"))

import requests  # noqa: E402
import login_996pc as L  # type: ignore  # noqa: E402
import cq2_pure_enter as P  # type: ignore  # noqa: E402
from encode6bitbuf_final import encode6, decode6  # type: ignore  # noqa: E402

from apilot.bridge import bootstrap_runtime  # noqa: E402
from apilot.stubs.env import EnvData  # noqa: E402
from apilot.stubs.netclient import NetClientStub  # noqa: E402
from apilot.stubs.networkctl import NetworkCtlStub, dispatch_frame  # noqa: E402
from apilot.stubs.scheduler import Scheduler  # noqa: E402
from apilot.wire import Frame  # noqa: E402

GW_HOST = "s{zone}-jycq-206923.dhsf.xqhuyu.com"
CHANNEL = "110585"
GAME_ID = 206923
LOGIN_PORTS = (8041, 8021, 8061, 8001)  # 端口轮询顺序(README: 8041 优先)


class _GateError(RuntimeError):
    """单端口登录链失败(超时/断连)→ 换下一端口重试。"""


class Host:
    def __init__(self, zone_id: int, user: str, password: str,
                 state_dir: str = ".apilot_state") -> None:
        self.zone = zone_id
        self.user, self.password = user, password
        self.state_dir = state_dir
        self.nc = NetClientStub()
        self.ctl = NetworkCtlStub(self.nc)
        self.sched = Scheduler()
        self.env = EnvData({})
        self.lua: Any = None
        self.jwt = ""
        self.uid = ""
        self._login_port = 0
        self._tail = 0

    def build_2001_body(self, uid: str, jwt: str) -> bytes:
        return (f"{uid}/".encode() + encode6(jwt.encode()) + b"/" +
                encode6(CHANNEL.encode()) + b"/1/0")

    def build_101_body(self, name: str, uid: str, job: int = 0, sex: int = 0) -> bytes:
        return json.dumps({"server_name": self.zone, "chr": name, "secsdk": "0",
                           "job": job, "sex": sex, "account": uid,
                           "sdkid": CHANNEL}, ensure_ascii=False).encode()

    def http_login(self) -> requests.Session:
        s = requests.Session()
        j, _ = L.do_login(s, self.user, self.password)
        if j.get("code") != 200:
            raise SystemExit(f"[!] 登录失败: {j.get('msg')}")
        self.jwt = j["data"]["token"]
        self.uid = str((j["data"].get("userInfo") or {}).get("id") or "")
        self.env.data.update({"uid": self.uid, "token": self.jwt,
                              "new_boxtoken": self.jwt, "isBoxLogin": 1})
        return s

    def boot_lua(self) -> None:
        self.lua = bootstrap_runtime(self.nc, self.ctl, self.sched, self.env, self.state_dir)

    def connect_login_gate(self, port: int = 8041) -> None:
        self._login_port = port
        self.nc.connect(GW_HOST.format(zone=self.zone), port)

    def pump(self, seconds: float, on_frame=None) -> list:
        """收帧→分发→调度, 驱动 seconds 墙钟。"""
        end = time.time() + seconds
        got: list = []
        while time.time() < end:
            for f in self.nc.tick(0.2):
                if self.lua is not None:
                    dispatch_frame(self.lua, self.ctl, f)
                got.append(f)
                if on_frame:
                    on_frame(f)
            self.sched.tick(0.05)
        return got

    def tick(self, dt: float = 0.05) -> None:
        for f in self.nc.tick(dt):
            if self.lua is not None:
                dispatch_frame(self.lua, self.ctl, f)
        self.sched.tick(dt)

    def _send(self, mid: int, recog: int, p1: int, p2: int, p3: int,
              tail: int, body: bytes) -> None:
        """登录链统一发包: body 字节由 host 构造,帧头走 pack_frame(wire.py)唯一真源。"""
        self.nc.send_frame(Frame(recog=recog, msg_id=mid, p1=p1, p2=p2,
                                  p3=p3, tail=tail, body=body))

    def _wait_for(self, mid: int, timeout: float):
        """收帧直到命中 mid 或超时。返回 (帧, 已收列表)。"""
        deadline = time.time() + timeout
        got: list = []
        while time.time() < deadline:
            remain = max(0.05, deadline - time.time())
            for f in self.pump(min(remain, 0.2)):
                got.append(f)
                if f.msg_id == mid:
                    return f, got
        return None, got

    def _recv_with_pre(self, target_mid: int, timeout: float):
        """等帧时同时拿回 m[0]("pre" 字段,apilot Frame 抽象丢弃)。
        仅 SC530 用 — cq2_pure_enter.login_gate:138 sid=m[0]。"""
        import struct
        wire = self.nc.wire
        if wire is None:
            return None, None
        deadline = time.time() + timeout
        while time.time() < deadline:
            remain = max(0.05, deadline - time.time())
            wire.sock.settimeout(min(0.2, remain))
            try:
                d = wire.sock.recv(65536)
                if not d:
                    break
            except (socket.timeout, ConnectionResetError, OSError):
                d = b""
            # 半包处理: 每轮把 wire._buf 前置到 data 再扫描, 扫剩的写回 _buf。
            data = wire._buf + d
            wire._buf = b""
            while len(data) >= 48:
                try:
                    m = struct.unpack("<9i", decode6(data[:48]))
                except Exception:
                    break
                blen = m[7]
                if len(data) < 48 + blen:  # 帧体不完整, 整帧留待下轮
                    break
                if m[1] == target_mid:
                    wire._buf = data[48 + blen:]
                    return Frame(recog=m[2], msg_id=m[1], p1=m[3], p2=m[4],
                                 p3=m[5], tail=m[8],
                                 body=data[48:48 + blen]), m[0]
                # 未命中帧 → 丢弃并继续扫下一帧(一次收到多帧的场景)
                data = data[48 + blen:]
            wire._buf = wire._buf + data
        return None, None

    def run_create_role(self, name: str | None = None, timeout: float = 30,
                        ports: tuple[int, ...] = LOGIN_PORTS) -> dict:
        """登录链逐帧驱动(对照 cq2_pure_enter.login_gate 唯一真源,逐帧注释)。

        内部: 逐端口 connect(8041/8021/8061/8001 轮询)→ 2001 链 → 101 → 521。
        返回 {"role_name", "ticket", "sid", "key32", "port"}。

        2001(tail=0,recog=random) → 等 529 取 key32(decode6 split /) →
        104(body=encode6(key32),tail=1) → 等 530 取 sid(m[0]/pre 字段,
        cq2_pure_enter:138) →
        nc.set_sdata(sid) → 100(uid/sid/zone/channel/gameid/key32,tail=2) →
        等 520 角色 JSON(decode6 后 json.loads) →
        角色列表含 name→复用;否则 101(tail=3) → 等 521(decode6 后 split /)
        /522(换名重试 ≤5 次)。
        """
        last_err: Exception | None = None
        for port in ports:
            if not self.nc.is_connected():
                try:
                    self.connect_login_gate(port)
                except OSError as e:
                    last_err = e
                    continue
            try:
                return self._create_role_once(name, timeout, port)
            except _GateError as e:
                last_err = e
                self.nc.disconnect()
                continue
        raise RuntimeError(f"[!] 所有候选端口失败: {last_err}")

    def _delete_role(self, role_uid: str, timeout: float) -> bool:
        """CS102 删角色(LoginProxy:RequestDeleteRole, body=UserID, p3=0, tail 严格递增)。
        返回是否收到 523(成功)。"""
        # 522 后立即发 102 服务器静默丢弃(实测); 停顿 ~2s 才被受理。
        time.sleep(1.5)
        self.pump(0.5)
        self._tail += 1
        self._send(102, 0, 0, 0, 0, self._tail, role_uid.encode())
        deadline = time.time() + timeout
        while time.time() < deadline:
            remain = max(0.05, deadline - time.time())
            for f in self.pump(min(remain, 0.2)):
                if f.msg_id == 523:
                    return True
                if f.msg_id == 524:
                    return False
        return False

    def _create_role_once(self, name: str | None, timeout: float,
                          port: int) -> dict:
        # CS2001 ─ 登录服首帧,带 uid/encode6(jwt)/encode6(channel)/1/0
        self._send(2001, random.randint(1, 2**30), self.zone, port, 0, 0,
                   self.build_2001_body(self.uid, self.jwt))
        # 等 SC529 → 会话密钥 key32(body 形如 "key32/...")
        f529, _ = self._wait_for(529, timeout)
        if f529 is None:
            raise _GateError("[!] 登录服未下发 529 key")
        key32 = decode6(f529.body).split(b"/")[0]
        # CS104 ─ 回传 key32 协商
        self._send(104, 0, 0, 0, 0, 1, encode6(key32))
        # 等 SC530 → sid 真实位置 m[0](首 int,"pre" 字段),不是 Frame.recog(m[2])。
        # cq2_pure_enter.login_gate:138 sid=m[0] 唯一真源;apilot Frame 抽象丢弃 m[0],
        # 这里 _recv_with_pre 直读 raw 字节补回。未命中超时即报错。
        _, sid = self._recv_with_pre(530, timeout)
        if sid is None:
            raise _GateError("[!] 登录服未下发 530 sid")
        self.nc.set_sdata(sid)
        # CS100 ─ 拉角色列表(body 形如 "uid/sid/zone/channel/gameid/"+key32)
        body100 = (f"{self.uid}/{sid}/{self.zone}/{CHANNEL}/{GAME_ID}/"
                   .encode() + key32)
        self._send(100, random.randint(1, 2**30), self.zone, port, sid, 2, body100)
        # 等 SC520 → 角色列表 JSON(cq2_pure_enter 兼容: encode6 包裹时先 decode6)
        roles: dict | None = None
        deadline = time.time() + timeout
        while time.time() < deadline:
            remain = max(0.05, deadline - time.time())
            for f in self.pump(min(remain, 0.2)):
                if f.msg_id == 520 and f.body:
                    try:
                        b = f.body
                        if all(0x3C <= c <= 0x7B for c in b[:200]):
                            b = decode6(b)
                        roles = json.loads(b)
                    except Exception:
                        roles = None
                    if roles is not None:
                        break
            if roles is not None:
                break
        # 选/创角
        if name is None:
            name = "萌新" + str(random.randint(1000, 9999))
        chr_info = next((c for c in (roles or {}).get("list", [])
                          if c.get("chr") == name), None)
        self._tail = 2
        attempts = 0
        while chr_info is None and attempts < 6:
            self._tail += 1
            self._send(101, 0, 0, 0, sid, self._tail,
                       self.build_101_body(name, self.uid))
            ticket: str | None = None
            got_522 = False
            sub_deadline = time.time() + timeout
            while time.time() < sub_deadline:
                remain = max(0.05, sub_deadline - time.time())
                for f in self.pump(min(remain, 0.2)):
                    if f.msg_id == 521 and f.body:
                        try:
                            ticket = decode6(f.body).decode("utf-8", "replace")
                        except Exception:
                            ticket = f.body.decode("utf-8", "replace")
                    if ticket is None and f.msg_id == 522:
                        got_522 = True
                if ticket or got_522:
                    break
            if ticket:
                chr_info = {"chr": name, "UserID": ticket.split("/", 1)[1]}
                break
            if got_522:
                attempts += 1
                # 槽满(客户端 LoginProxy:648 i>2 break → 每区 2 角色上限):
                # 删一个非目标角色腾槽(CS102 tail 严格递增, p3=0), 再原名重试;
                # 删无可删才换名。523/524 等待在 _delete_role 内完成。
                lst = (roles or {}).get("list") or []
                # 服务器只允许删 last=1(当前选中)角色 — 删其它静默无响应(实测)。
                # LoginProxy:916 删除后 DelRoleByID(GetSelectedRoleID()) 同一语义。
                victim = next((c for c in lst
                               if c.get("chr") != name and c.get("last") == 1),
                              None)
                if victim is not None and self._delete_role(victim["UserID"],
                                                            timeout):
                    lst.remove(victim)  # 防下轮重复删同一已删角色(重复删静默)
                else:
                    name = "萌新" + str(random.randint(1000, 9999))
                continue
            break
        if chr_info is None:
            lst = (roles or {}).get("list") or []
            if lst:
                chr_info = lst[-1]
        if chr_info is None:
            raise _GateError("[!] 无角色且创角失败")
        ticket = chr_info["UserID"]
        return {"role_name": chr_info["chr"], "ticket": ticket,
                "sid": sid, "key32": key32, "port": port}


def run_zone(zone_id: int, user: str, password: str, name: str | None = None,
             port: int | None = None) -> dict:
    """一站式: HTTP 登录 → enter_game → 网关连接(端口轮询) → 创角 → 返回票据。"""
    h = Host(zone_id, user, password)
    session = h.http_login()
    j2 = P.enter_game(session, h.jwt, zone_id)
    if j2.get("code") != 200:
        raise SystemExit(f"[!] enter_game 失败: {j2.get('msg') or j2}")
    ports = (port,) if port else LOGIN_PORTS
    info = h.run_create_role(name, ports=ports)
    return {"uid": h.uid, "jwt": h.jwt, "zone": h.zone, **info}