chuanqi/protocol_bot_backup/combat_bot.py

# -*- coding: utf-8 -*-
"""
combat_bot.py —— 无客户端杀怪: 进游戏 → 视野怪物表 → 寻路接近 → CS3014 近战击杀。

协议来源(游戏Lua源码):
  gamePlayerController.lua:93   CS3013 跑: recog=Y*65536+X, p2=dir(0-7), p3=lastPoint, body=mapID
  RequestLaunchCommand.lua:387  CS3014 砍: recog=skillID, body=JSON{skillID,dir,srcX,srcY,destX,destY,targetID}
  gameMapController.lua:114-130 SC10 进视野: recog=race|raceImg<<16, p1=actorID, p2=X|Y<<16, p3=type|Dir<<16, body=JSON
  mmoConstants.lua:141-148      方向: 0上 1右上 2右 3右下 4下 5左下 6左 7左上
  mmoConstants.lua:771          SKILL_INDEX_BASIC=0
用法:
  .venv/bin/python combat_bot.py --zone 14492162 --name 测C1742 [--kill N] [--minutes 5]
"""
import argparse
import json
import math
import random
import socket
import struct
import sys
import threading
import time
import zlib

import requests as rq
import urllib3
urllib3.disable_warnings()

HERE = __file__.rsplit("/", 1)[0]
sys.path.insert(0, HERE)

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

RACE_HUMAN = {0, 1, 5, 11}        # 网络玩家/英雄/宝宝/守卫 → 非怪
DIRS = [(0, -1), (1, -1), (1, 0), (1, 1), (0, 1), (-1, 1), (-1, 0), (-1, -1)]  # 0-7


def enter_game_gate_v2(zone_id, name, ticket_key, sid, game_port, uid):
    """CS20 登录体对照 LoginProxy.lua:1076 的 11 字段格式:
    *#账号/角色名/角色ID/sdata/platform/channelId/whitelist/cert/adult/isboxlogin/isBatchMsg
    旧版(enter_game_gate)多一个 /1 且 channelId 错位 → 服务器不下发 SC658(世界未初始化)。"""
    g = socket.create_connection((P.GW_HOST.format(zone=zone_id), game_port), timeout=10)
    body20 = f"*#{uid}/{name}/{ticket_key}/{sid}/2/{P.CHANNEL}/0/0/0/1/1".encode()
    head = struct.pack("<9i", 900000, 20, random.randint(1, 2**30),
                       zone_id, game_port, sid, 0, len(body20), 0)
    g.sendall(encode6(head) + body20)
    return g


def send_init_frames(bot, uid):
    """init 序列逐帧复刻 cap_unlock 真客户端(sid=159 会话 t=44.83-45.09):
    1018(t2) 1046×4(t3-6) 2020(t7,uid) 5026(t8) 5028(t9)
    5027(t10, body=当前图名!) 5260(t11) 1078(t12) 1079(t13) 5148(t14)
    5277(t15) 1103(t16) 1151(t17) 6209(t18) 6208(t19)
    关键: 5027 必须是【当前所在图名】(等待图='等待地图'), 在 5028 之后、
    5260 之前 — 旧版发 b'3' 且位置错 → 服务器图会话绑定失败 → 移动全拒。"""
    tpls = P._load_ui_templates()
    cur_map = bot.me.get("mapname") or "等待地图"
    frames = [
        (1018, P.CFG_1018.encode()),
        (1046, b""), (1046, b""), (1046, b""), (1046, b""),
        (2020, f"{uid}/3.40.4087/3.40.9112/{uid}/0/1001/110585/0/0".encode()),
        (5026, P.DEV_FINGERPRINT_5026),
        (5028, P.CHANNEL_5028),
        (5027, cur_map.encode()),
        (5260, b""), (1078, b""), (1079, b""), (5148, b""),
        (5277, tpls["5277"].encode() if tpls else b""),
        (1103, b""), (1151, b""), (6209, b""), (6208, b""),
    ]
    for i, (mid, body) in enumerate(frames):
        p1, p2 = (41878640, 98395) if mid == 1018 else (0, 0)
        bot.send(mid, 0, p1, p2, bot.sid, body, tail=2 + i)
        bot.pump(0.1)


def log(m):
    print(f"[{time.strftime('%H:%M:%S')}] {m}", flush=True)


def lo16(v):
    return v & 0xFFFF


def hi16(v):
    return (v >> 16) & 0xFFFF


def pack_point(x, y):
    return (y & 0xFFFF) * 65536 + (x & 0xFFFF)


def proto_map_name(me):
    """走路帧 body 的协议图名 = 服务器下发的 MapId(SC634/SC51 JSON)。
    中文名→协议ID 反查兜底: 比奇='0' 盟重='3' 等待图='等待地图'(中文名即协议名)。
    发错图名服务器直接拒帧(run5-11 站桩根因: me['map'] 残留出生图'3')。"""
    m = str(me.get("map") or me.get("mapname") or "0")
    return {"比奇大城": "0", "盟重土城": "3"}.get(m, m)


def dir_to(x, y, tx, ty):
    dx, dy = max(-1, min(1, tx - x)), max(-1, min(1, ty - y))
    for i, (ox, oy) in enumerate(DIRS):
        if ox == dx and oy == dy:
            return i
    return 2


class CombatBot:
    def __init__(self, sock, sid, name):
        self.g, self.sid, self.name = sock, sid, name
        self.me = {"id": 0, "x": 0, "y": 0, "dir": 4, "map": "0", "mapname": ""}
        self.monsters = {}          # actorID -> {x,y,race,name,die}
        self.npcs = {}              # actorID -> {cfg_id,name,index,x,y}
        self.frame_log = {}         # msgId -> count (观测用)
        self.kills = 0
        self.exp0 = None
        self._tail = 100
        self._buf = b""
        self.alive = True
        self.lock = threading.Lock()

    # ---------- 发送 ----------
    def send(self, mid, recog, p1, p2, p3, body=b"", tail=None):
        self._tail += 1
        t = self._tail if tail is None else tail
        with self.lock:
            self.g.sendall(encode6(struct.pack(
                "<9i", 0, mid, recog, p1, p2, p3, 0, len(body), t)) + body)

    def send_run(self, x, y, last_x, last_y, d):
        """CS3013 跑2格 — 实测验证布局(combat_bot 早期等待图22步全通同源):
        Lua: LuaSendMsg(msgType, point, 0, dir, lastPoint, mapID)
        → recog=pack_point(x,y), p1=0, p2=dir, p3=pack_point(last), body=MapId。"""
        for v in (x, y, last_x, last_y):
            if not 0 <= v <= 65535:
                return
        self.send(3013, pack_point(x, y), 0, d, pack_point(last_x, last_y),
                  proto_map_name(self.me).encode())

    def send_walk(self, x, y, last_x, last_y, d):
        """CS3011 走1格 — 布局同 send_run(走/跑共用 handleActionBegin 路径)。"""
        for v in (x, y, last_x, last_y):
            if not 0 <= v <= 65535:
                return
        self.send(3011, pack_point(x, y), 0, d, pack_point(last_x, last_y),
                  proto_map_name(self.me).encode())

    def send_attack(self, target_id, tx, ty):
        # cap3 实证 targetID 为字符串("57659"), 且 body 键序与真实客户端一致
        d = dir_to(self.me["x"], self.me["y"], tx, ty)
        payload = {"skillID": 0, "dir": d, "srcX": self.me["x"], "srcY": self.me["y"],
                   "destX": tx, "destY": ty, "targetID": str(target_id)}
        body = json.dumps(payload, separators=(",", ":")).encode()
        # 关键诊断: 打印首次攻击,确认攻击确实发出
        if not getattr(self, "_first_atk_logged", False):
            self._first_atk_logged = True
            log(f"    [首次攻击] me=({self.me['x']},{self.me['y']}) → "
                f"target={target_id} @{tx},{ty} dir={d}")
        self.send(3014, 0, 0, 0, 0, body)

    # ---------- 收帧/状态 ----------
    def wait_msg(self, mid, timeout):
        deadline = time.time() + timeout
        while time.time() < deadline and self.alive:
            self.pump(0.3)
            if mid in self.frame_log:
                return True
        return mid in self.frame_log

    def pump(self, seconds=0.2):
        end = time.time() + seconds
        events = []
        while time.time() < end:
            self.g.settimeout(max(0.05, end - time.time()))
            try:
                d = self.g.recv(65536)
                if not d:
                    self.alive = False
                    return events
            except (socket.timeout, OSError):
                break
            events += self._parse(d)
        return events

    def _parse(self, d: bytes):
        """帧切分 + 按已知消息号更新状态。返回可见事件列表。"""
        ev = []
        buf = self._buf + d
        pos = 0
        while len(buf) - pos >= 48:
            try:
                h = struct.unpack("<9i", decode6(buf[pos:pos + 48]))
            except Exception:
                pos += 1
                continue
            blen = h[7]
            if blen < 0 or blen > 200000 or len(buf) - pos < 48 + blen:
                break
            mid, recog, p1, p2, p3, tail = h[1], h[2], h[3], h[4], h[5], h[8]
            body = buf[pos + 48: pos + 48 + blen]
            pos += 48 + blen
            self.frame_log[mid] = self.frame_log.get(mid, 0) + 1
            if mid == -1 and body[:2] == b"x\x9c":
                # SC-1 批量帧: body=zlib(子帧流), 子帧同顶层格式 → 递归
                try:
                    ev += self._parse_sub(zlib.decompress(body))
                except zlib.error:
                    pass
                continue
            ev += self._dispatch(mid, recog, p1, p2, p3, body)
        self._buf = buf[pos:]
        return ev

    def _parse_sub(self, raw: bytes):
        """批量帧内的子帧流拆包(不回写 _buf)。"""
        ev = []
        pos = 0
        while len(raw) - pos >= 48:
            try:
                h = struct.unpack("<9i", decode6(raw[pos:pos + 48]))
            except Exception:
                break
            blen = h[7]
            if blen < 0 or blen > 200000 or len(raw) - pos < 48 + blen:
                break
            self.frame_log[h[1]] = self.frame_log.get(h[1], 0) + 1
            ev += self._dispatch(h[1], h[2], h[3], h[4], h[5],
                                 raw[pos + 48: pos + 48 + blen])
            pos += 48 + blen
        return ev

    def _jbody(self, body):
        """zlib → encode6 → JSON 逐层剥。实测: SC11/13 body=encode6(JSON) 无 zlib;
        部分帧 zlib(encode6(JSON)); 部分帧直接 JSON。"""
        if not body:
            return None
        for blob in CombatBot._body_variants(body):
            if not blob:
                continue
            try:
                return json.loads(blob)
            except Exception:
                continue
        return None

    @staticmethod
    def _body_variants(body: bytes):
        """产出所有可能的解码变体: 原文 / encode6解 / zlib解 / zlib(encode6解)。"""
        out = [body]
        out.append(CombatBot._try_decode6(body))
        if body[:2] == b"x\x9c":
            try:
                dec = zlib.decompress(body)
                out.append(dec)
                out.append(CombatBot._try_decode6(dec))
            except zlib.error:
                pass
        return out

    @staticmethod
    def _try_decode6(raw: bytes):
        """可打印区(0x3C..0x7B)开头才尝试 decode6(长度不必 8 倍数, 内部按 bit 流)。"""
        if not raw or not all(0x3C <= c <= 0x7B for c in raw[:16]):
            return b""
        try:
            return decode6(raw)
        except Exception:
            return b""

    def _dispatch(self, mid, recog, p1, p2, p3, body):
        ev = []
        # 旁路钩子: 外部脚本监听全部已拆包子帧(文本类解码由钩子自理)
        hook = getattr(self, "raw_frame_hook", None)
        if hook:
            try:
                hook(mid, recog, p1, p2, p3, body)
            except Exception:
                pass
        # SC50 创建主玩家 / SC10 进视野(对照 converActorData: X=lo16(p2) Y=hi16(p2),
        # race=lo16(recog), type=lo16(p3), nActorId=p1)
        if mid in (50, 10):
            x, y = lo16(p2), hi16(p2)
            typ = lo16(p3)
            if mid == 50:
                self.me["id"] = p1
                j = self._jbody(body)
                if j and j.get("info"):
                    self.me["uid"] = str(j["info"][0])   # 完整UserID字符串(SC11/13自识别用)
                if p2 != 0:                   # SC50 头 p2=0 时坐标以 SC51 为准
                    self.me["x"], self.me["y"] = x, y
                ev.append(("me_born", self.me["x"], self.me["y"]))
            elif typ == 50:                    # ACTOR_MONSTER(mmoConstants:76)
                j = self._jbody(body) or {}
                cs = j.get("cs", 0)
                dead = (cs & 0b100) or (cs & 0b10)   # die/skeleton 位
                info = j.get("info")
                self.monsters[p1] = {
                    "x": x, "y": y, "race": lo16(recog),
                    "name": info[2] if info else str(p1),
                    "die": bool(dead), "t": time.time()}
                ev.append(("mon_in", p1, x, y))
            else:                              # NPC/玩家: info=[actorID字符串, 名称]
                j = self._jbody(body) or {}
                info = j.get("info")
                if info and len(info) >= 2:
                    self.npcs[p1] = {"aid": str(info[0]), "name": info[1],
                                     "index": p1, "x": x, "y": y}
                    ev.append(("npc_in", str(info[0]), info[1], p1, x, y))
        elif mid == 643:                               # SC643 NPC对话树
            j = self._jbody(body)
            if j:
                ev.append(("npc_talk", j))
        elif mid == 20:                               # SC20 怪物出生(复活)
            try:
                aid = int(body.decode("utf-8", "replace").strip())
            except ValueError:
                aid = 0
            if aid in self.monsters:
                self.monsters[aid]["die"] = False
                self.monsters[aid]["t"] = time.time()
        elif mid == 30:                               # SC30 出视野
            try:
                aid = int(body.decode("utf-8", "replace").strip())
            except ValueError:
                aid = 0
            if aid in self.monsters:
                del self.monsters[aid]
                ev.append(("mon_out", aid))
        elif mid == 32:                               # SC32 死亡倒下
            try:
                aid = int(body.decode("utf-8", "replace").strip())
            except ValueError:
                aid = 0
            if aid in self.monsters:
                self.monsters[aid]["die"] = True
                self.kills += 1
                ev.append(("mon_die", aid))
        elif mid == 28:                               # SC28 移动失败拉回(p1=X,p2=Y, p3=dir)
            self.me["x"], self.me["y"] = p1, p2
            self.me["dir"] = p3 & 0xFF
            ev.append(("pull_back", p1, p2))
        elif mid in (11, 13):                      # SC11 走/SC13 跑 广播(recog=dir,p1=X,p2=Y)
            if recog and 0 <= (recog & 0xFF) < 8:  # 疑似自己移动的确认(简化: 无aid时跳过)
                pass
            ev.append(("actor_move", lo16(p1) if p1 > 65535 else p1,
                       hi16(p1) if p1 > 65535 else 0, mid))
        elif mid == 51:                               # SC51 地图初始化(p1=X,p2=Y 整数)
            j = self._jbody(body) or {}
            mapname = j.get("MapName") or str(j.get("MapId", "?"))
            self.me["map"] = j.get("MapId", mapname)
            self.me["mapname"] = mapname
            if mid == 51:
                # p1/p2 bit11 是标志位非坐标(README 定案: 2048 类值即此)
                self.me["x"], self.me["y"] = p1 & 0x7FF, p2 & 0x7FF
            ev.append(("map", mapname, p1, p2))
        elif mid == 634:                              # SC634 场景切换(等待图→比奇)
            ev.append(("map_change",))
        elif mid == 44:                               # SC44 经验
            j = self._jbody(body)
            if j:
                ev.append(("exp", j))
        return ev

    # ---------- 行为 ----------
    def nearest_monster(self):
        cands = [(math.dist((m["x"], m["y"]), (self.me["x"], self.me["y"])), aid)
                 for aid, m in self.monsters.items() if not m["die"]]
        return min(cands)[1] if cands else None

    def step_toward(self, tx, ty):
        """一格一步: 候选方向按接近度排序, 走路(CS3011)。"""
        x, y = self.me["x"], self.me["y"]
        dx = max(-1, min(1, tx - x))
        dy = max(-1, min(1, ty - y))
        if dx == 0 and dy == 0:
            return True
        occupied = {(m["x"], m["y"]) for m in self.monsters.values() if not m["die"]}
        cands = []
        for i, (ox, oy) in enumerate(DIRS):
            nx, ny = x + ox, y + oy
            if (nx, ny) in occupied:
                continue
            cands.append((math.dist((nx, ny), (tx, ty)), i, nx, ny))
        if not cands:
            return False
        cands.sort()
        _, d, nx, ny = cands[0]
        self.send_walk(nx, ny, x, y, d)
        self.me["x"], self.me["y"], self.me["dir"] = nx, ny, d
        return False

    def fight(self, kill_target=None, minutes=5):
        log(f"战斗循环启动: 目标数={kill_target or '∞'} 时长={minutes}min "
            f"玩家=({self.me['x']},{self.me['y']}) map={self.me.get('mapname', self.me['map'])}")
        deadline = time.time() + minutes * 60
        target = None
        no_target_t0 = None
        last_map_log = ""
        while time.time() < deadline and self.alive:
            self.pump(0.3)
            if kill_target and self.kills >= kill_target:
                log(f"已达击杀数 {self.kills}, 停止")
                break
            mname = self.me.get("mapname", "")
            if mname != last_map_log:
                last_map_log = mname
                log(f"当前地图: {mname} 玩家@({self.me['x']},{self.me['y']}) 怪物数={len(self.monsters)}")
                self.monsters.clear()     # 换图后视野重建, 防旧坐标污染
                target = None
            m = None
            if target is not None and target in self.monsters:
                m = self.monsters[target]
                if m["die"]:
                    target, m = None, None
            if m is None:
                target = self.nearest_monster()
                m = self.monsters.get(target)
                if target is not None:
                    log(f"选定目标 #{target} {m['name']} @({m['x']},{m['y']})")
            if m is None:
                if no_target_t0 is None:
                    no_target_t0 = time.time()
                    log(f"视野内无怪物({mname}), 开始探索走动...")
                for _ in range(5):
                    self.pump(0.05)
                    d = random.randrange(8)
                    nx = self.me["x"] + DIRS[d][0]
                    ny = self.me["y"] + DIRS[d][1]
                    self.send_walk(nx, ny, self.me["x"], self.me["y"], d)
                    self.me["x"], self.me["y"], self.me["dir"] = nx, ny, d
                    time.sleep(0.35)
                continue
            no_target_t0 = None
            dist = max(abs(m["x"] - self.me["x"]), abs(m["y"] - self.me["y"]))
            if dist <= 1:
                self.send_attack(target, m["x"], m["y"])
                time.sleep(0.65)     # HitTime=650ms(见SC2 gameoptions)
            else:
                self.step_toward(m["x"], m["y"])
                time.sleep(0.35)
        log(f"循环结束: kills={self.kills} 收帧统计={sorted(self.frame_log.items(), key=lambda kv: -kv[1])[:12]}")
        return self.kills


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("-u", "--user", default="longwu988")
    ap.add_argument("-p", "--password", default="Longwu1988")
    ap.add_argument("--zone", type=int, required=True)
    ap.add_argument("--name", default="测C1742")
    ap.add_argument("--port", type=int, default=0)
    ap.add_argument("--kill", type=int, default=0, help="击杀N只后停, 0=不限")
    ap.add_argument("--minutes", type=float, default=5)
    args = ap.parse_args()

    log("[1/5] HTTP登录...")
    j = None
    for attempt in range(5):
        try:
            s = rq.Session()
            j, _ = L.do_login(s, args.user, args.password)
            break
        except Exception as e:
            log(f"      登录失败(第{attempt+1}次): {str(e)[:70]}, 15s后重试")
            time.sleep(15)
    if not j or j.get("code") != 200:
        raise SystemExit(f"[!] 登录失败: {(j or {}).get('msg', '网络异常')}")
    jwt = j["data"]["token"]
    uid = str((j["data"].get("userInfo") or {}).get("id") or P.UID)

    log("[2/5] 选区...")
    j2 = None
    for attempt in range(5):
        try:
            j2 = P.enter_game(s, jwt, args.zone)
            break
        except Exception as e:
            log(f"      enter_game 失败(第{attempt+1}次): {str(e)[:60]}, 15s后重试")
            time.sleep(15)
    if not j2 or j2.get("code") != 200:
        raise SystemExit(f"[!] enter_game 失败: {(j2 or {}).get('msg', '网络异常')}")

    log("[3/5] 登录服握手(复用角色)...")
    ports = [args.port] if args.port else [8041, 8021, 8061, 8001]
    ok = None
    for pt in ports:
        try:
            key32, sid, rname, tkey, game_port = P.login_gate(
                args.zone, jwt, args.name, pt, uid)
            if key32:
                ok = (key32, sid, rname, tkey, game_port, pt)
                break
        except SystemExit as e:
            log(f"      端口{pt}: {e}")
        except Exception as e:
            log(f"      端口{pt}: {str(e)[:60]}")
    if not ok:
        raise SystemExit("[!] 所有端口失败")
    key32, sid, rname, tkey, game_port, pt = ok
    log(f"      角色={rname} 游戏端口={game_port}")

    log("[4/5] 连游戏服+进世界...")
    g = enter_game_gate_v2(args.zone, rname, tkey, sid, game_port, uid)
    bot = CombatBot(g, sid, rname)
    P.heartbeat_loop(g, sid, bot.lock)      # CS20后立即心跳, 防超时踢
    ok = False
    t0 = time.time()
    while time.time() - t0 < 10:
        bot.pump(0.3)
        if 658 in bot.frame_log:
            ok = True
            break
    log(f"      SC658 {'✓' if ok else '✗ 超时'} ({time.time()-t0:.1f}s)")
    send_init_frames(bot, uid)   # 不用 P.send_init_sequence: 其收尾 recv 清空会吃掉 SC50

    for _ in range(20):                     # 收出生/地图/视野帧
        evs = bot.pump(0.5)
        for e in evs:
            log(f"      事件: {e}")
        if bot.me["id"] and bot.me["map"] not in ("0", ""):
            break
        if not bot.alive:
            break
    log(f"      收帧统计: {sorted(bot.frame_log.items(), key=lambda kv: -kv[1])[:15]}")
    if not bot.me["id"]:
        g.close()
        raise SystemExit("[!] 未收到 SC50 主玩家创建, 中止")
    log(f"      主玩家 #{bot.me['id']} @({bot.me['x']},{bot.me['y']}) map={bot.me['map']}")

    log("[5/5] 开喝药保护 + 心跳...")
    state = P.enable_auto_heal(g, sid, want_on=True)
    log(f"      喝药={state}")
    P.heartbeat_loop(g, sid, bot.lock)

    try:
        kills = bot.fight(kill_target=args.kill or None, minutes=args.minutes)
        log(f"[完成] 击杀={kills}")
    finally:
        g.close()


if __name__ == "__main__":
    main()