# -*- 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()