# -*- coding: utf-8 -*- """ quest_bot3.py —— 推送驱动微缩客户端(真实客户端双向通信模式重构版)。 2026-09-22 cap2.jsonl 抓包破译的关键规则(本文件的设计依据): 1. 服务器是流程的主人: SC801 传送帧(recog=X,p1=Y,p2=dir,p3=256,body={"UserID":票据}) → 客户端立即 CS5012+CS5027(body=新地图名) 加载资源 → 落地后 CS1 ACK SC51 地图初始化(p1=X,p2=Y) → 更新自身坐标/地图名 SC10/50 视野 → NPC/怪物表(新手接待NPC的 UserID 每角色不同: 矿工10032/斗士10033, 必须动态发现) 2. CS1 落地确认: recog=新X, p1=新Y, p2=dir, p3=32171624, tail=943, body=encode6("+GOOD/<自增毫秒"), 计数器初值 52549000000 量级, 每步 +547。 (每步都发也无害 — 真实客户端走路时每步 CS1) 3. 走路: CS3013(跑2格)/CS3011(走1格): p1=dir(0-7), p2=起点(Y<<16|X), body=地图名。 实测服务器按 dir 推算落点, body 地图名可以为过期缓存(斗士全程带"等待地图"走比奇网格)。 4. 任务规则(用户实测): 点击任务指引(@renwu_mov,1) → 服务器直接传送角色到出生地 → 客户端自己跑到新手接待NPC → 点NPC(CS1010, index=34) → 传送新手村(比奇) → 跟随指引跑到任务点 → @startguaji 挂机 → SC32×N → @stopguaji。 5. 防呆: @renwu_mov 后必须在 CS5012/5027 里回发服务器指定的地图名, 否则流程卡死。 6. 移动全链路(2026-09-23 Lua真源定案, 解密源码/mod_fgcq/stab/scripts): BehaviorMove(行为树,每帧) → checkNextMovePos(A*路径下一格,可合1-3格) → player:SetMoveTo+SetAction(WALK/RUN) → 动画到点 SetConfirmed(false) → gamePlayerController:handleActionBegin 发帧: LuaSendMsg(3011/3013, point=Y<<16|X, 0, dir, lastPoint, mapID) → 服务器 SC11 自回显/SC1 确认 → SetConfirmed(true) → 下一格; 超时-5s强制确认。服务器开关 gameoptions.RunOne 开启时走一格也用 RUN(3013/3012)。 等待图滞留带(y=0出生行, 0139.map 中 y0-2 全墙)内一切移动帧被拒(SC28 recog=22/24), 唯一出口=新手接待NPC对话(CS1010 idx=35实证)传比奇; @renwu_mov 传回等待图落点(23,25)属可走区, 但移动仍需 SC20102(自动寻路目标) → CS5013(START,x,y) 注册寻路会话(AutoMoveBeginCommand), 未注册则 SC28 全拒。 架构: 单线程事件循环(recv 泵 + 定时器) + 显式状态机 — 与真实客户端 "recv_loop 常驻线程 + Lua 消息订阅"同构, 但无线程竞争。 状态机: ENTER → [等待图: 找NPC→走近→CS1010] → TELEPORT → [比奇: 走向任务点] → GUAJI(挂机杀怪) → 杀满 N → STOP → 退出 用法: .venv/bin/python quest_bot3.py --zone 14492165 --name 斗士9906 [--kill 2] [--dry-run] 等待图走动: .venv/bin/python quest_bot3.py --mode wait [--zone 0=自动选最新区] [--minutes 2] """ import argparse import json import os import random import re import socket import struct import sys import time 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 from combat_bot import (CombatBot, send_init_frames, DIRS, log, pack_point, # noqa: E402 lo16, hi16, proto_map_name) from mir_map import load as map_load # noqa: E402 STEP_MS = 547 # 实测步长 # 杀怪区: cap3 抓包实证战斗发生在 (287-288, 621-623) — 传送落点(292,618)旁 4 格。 # (288,622)在 0.map 文件中是墙格(服务器网格与文件有1-2格偏差) → 取邻格(287,622)。 # 旧值(292,580)是斗士 run 终点, 实测无怪(run9 站桩7分钟)。 KILL_AREA = (287, 622) WANDER_RANGE = 50 # 游走找怪半径(Lua checkAimlessMove 同款) QUEST_POINT = (292, 618) # 比奇任务点(传送落点) WAIT_MAP = "等待地图" # 协议地图名(SC51 MapName/MapId) → .map 文件。 # 抓包实证: 进盟重土城加载 '3', 点NPC传送比奇加载 '0' → 比奇='0'(0.map 700x700), # 盟重='3'(3.map 1000x800)。(旧会话总结中两者写反, 已按坐标可走性+抓包双重纠正) MAP_FILES = {WAIT_MAP: "0139", "比奇大城": "0", "盟重土城": "3", "0": "0", "3": "3"} _MAP_OBJS = {} def _map_cached(fname): if fname not in _MAP_OBJS: try: _MAP_OBJS[fname] = map_load(fname) except Exception: _MAP_OBJS[fname] = None return _MAP_OBJS[fname] def map_file_for(mname, x, y): """按当前位置在哪张图可走来选 .map 文件(名字不可信: 比奇协议id='0'撞盟重文件名)。""" if mname in MAP_FILES: return MAP_FILES[mname] for cand in ("3", "0", "0139"): m = _map_cached(cand) if m and m.in_bounds(x, y) and m.walkable(x, y): return cand return None def _walkable_at(x, y): """按当前图文件判断可走(服务器网格与文件有1-2格偏差, 仅作游走候选粗筛)。""" for cand in ("0", "3", "0139"): m = _map_cached(cand) if m and m.in_bounds(x, y): return m.walkable(x, y) return False class PushBot(CombatBot): """推送驱动的微缩客户端。 与旧 WalkBot 的区别: - SC801(传送) → 自动加载地图资源(CS5012/5027) + CS1 落地 ACK(对齐真实客户端) - SC51 → 强制校正坐标+地图名(过期的客户端地图缓存不再污染走路 body) - NPC/UserID 动态发现(SC10), 不硬编码 - 单循环状态机, 所有推进都由 pump() 里收到的推送触发 """ def __init__(self, sock, sid, name): super().__init__(sock, sid, name) self.good_ms = 52549000000 + random.randint(0, 10000) # CS1 计数器(实测区间) self.state = "ENTER" # ENTER/WALK_NPC/TELEPORT/WALK_QUEST/GUAJI/DONE self.pending_map_ack = None # SC801 到达时立即回发的目标图名(CS5012/5027 body) self.teleports = 0 self.npc_10032 = None # 新手接待 NPC(actordata) self.hp = self.max_hp = 0 # SC53/SC31 状态跟踪(引擎必处理帧) self.missions = {} # SC55 任务状态 {type: flag} flag:0未接/1进行/2可交 self.mission_bodies = {} # SC55 原样任务体 {type: json} — CS3031 回构造用 self._last_step_t = 0.0 self._step_pending_ack = None # (dx, dy, due_t) self._landing_ack_due = None # (x, y, dir, due_t) 传送落地ACK self._conf_pending = False # SC1 行为确认门控(mIsConfirmed 同款) self.mon_kills = 0 # SC32 计数(与父类 kills 分开, 供挂机停止判定) self.dead = False # SC32 body=自己uid → 死亡态(等SC27复活) self.on_event = None # debug hook # ---------- 覆写: 推送处理 ---------- def _dispatch(self, mid, recog, p1, p2, p3, body): ev = [] if mid == 1: # SC1 = MSG_SC_ACTION_CONFIREM_SUCCESS(主玩家行为确认, netMessageDef:94) # Lua: handle_MSG_SC_ACTION_CONFIRM → SetConfirmed(true) — 走路/攻击 # 的每一步都要等它, 没确认前真实客户端不发下一个动作。 if getattr(self, "_conf_pending", False): self._conf_pending = False self.confirmed_at = (self.me["x"], self.me["y"], time.time()) ev.append(("action_confirmed",)) if mid == 10: # SC10 进视野: (race, dir) 统计 + 保留 body 样本供离线分析。 # 实测新手怪 race=1049(39只集中出现), type=lo16(p3) 实为 255/256(dir 域)。 self.sc10_types = getattr(self, "sc10_types", {}) t = lo16(recog) self.sc10_types[t] = self.sc10_types.get(t, 0) + 1 if lo16(recog) not in (0, 1, 5, 11) and len(getattr(self, "sc10_bodies", [])) < 10: j2 = self._jbody(body) if j2: self.sc10_bodies = getattr(self, "sc10_bodies", []) + [(recog, p1, p2, p3, j2)] if mid == 800: # 传送帧真身 = SC800(cap3 实证): body=JSON{"UserID":"<自己长uid>"}, p3=256, # 坐标在随后的 SC51 里。SC801 是"别人飞进视野"广播, 勿混。 j = self._jbody(body) or {} uid = str(j.get("UserID", "")) myuid = self.me.get("uid") or "" if uid and uid == myuid: self.teleports += 1 dest = self.pending_map_ack or self._guess_map_name(self.me["x"], self.me["y"]) self._landing_ack_due = (self.me["x"], self.me["y"], self.me["dir"], time.time() + 0.5) ev.append(("teleport", dest)) if mid == 801: # 传送帧(引擎:MSG_SC_NET_PLAYER_FLY_IN)。注意: 自己出生也会来一帧 FLY_IN, # 只有位移>5格才算真传送。抓包时序: SC801→30ms CS5012/5027(body=目标图名)。 x, y, d = recog, p1, p2 & 0xFF if y & 0x800: y &= 0x7FF # 出生帧 Y bit11 编码位 moved = max(abs(x - self.me["x"]), abs(y - self.me["y"])) self.me["x"], self.me["y"], self.me["dir"] = x, y, d if moved > 5: self.teleports += 1 dest = self.pending_map_ack or self._guess_map_name(x, y) self._send_map_load(dest) self._landing_ack_due = (x, y, d, time.time() + 0.5) ev.append(("teleport", x, y, dest)) else: ev.append(("fly_in_self", x, y)) elif mid == 51: # SC51 场景创建 — body=JSON{MapName(显示名), MapId(协议ID)} # 布局(cap_unlock t=45.00 批量子帧实证): recog=X, p1=Y(2048=bit11标志, # 非坐标) — 与 SC634/SC11/SC28/SC1 同族。旧按 p1=X,p2=Y 解析 → # 出生坐标整体读错(如(23,25)读成(25,0)) → 从错位起点发移动帧 → # 服务器 SC28 拉回真位置 → 表现为"角色永远走不动"(全案根因)。 j = self._jbody(body) or {} newname = j.get("MapName") or str(j.get("MapId", "?")) old = self.me.get("mapname") self.me["mapname"] = newname if j.get("MapId") is not None: self.me["map"] = str(j["MapId"]) nx = recog & 0xFFFF if recog > 0 else self.me["x"] ny = p1 & 0xFFFF if p1 > 0 else self.me["y"] self.me["x"], self.me["y"] = nx, ny # 坐标跳变即传送/出生: lastPoint 重置(CS3011 p3 连续性校验) self.me["last_x"], self.me["last_y"] = nx, ny if newname != old: self.monsters.clear() self.npcs.clear() # 换图即主动回 CS5012/5027(协议图名) — cap3 实证客户端必发, # 不发则服务器不下发该图视野推送(SC10 一帧不来, diag 实证) self._send_map_load(newname) ev.append(("map", newname, nx, ny)) elif mid == 634: # SC634 场景切换(跨图传送) — Lua真源 MapProxy:handle_MSG_SC_MAP_CHANGE: # mapX=param1, mapY=param2(标准布局)。body=encode6(JSON){MapName,MapId,...}。 # 旧实现不解析 → me['map'] 残留出生图'3', 比奇走路 body 全发'3'被拒 # (run5-11 站桩根因); 坐标误用 recog/p1 也修(p1=X, p2=Y&0x7FF)。 j = self._jbody(body) or {} old = self.me.get("mapname") if j.get("MapName"): self.me["mapname"] = j["MapName"] if j.get("MapId") is not None: self.me["map"] = str(j["MapId"]) # 布局(SPY实捕 2026-09-23): recog=X, p1=Y(p2是bit11标志2048) — # 与 SC11/SC28/SC801 同族。旧按 p1=X,p2=Y 解析 → 坐标错位, # 后续走路链起点错 → SC28 拉回滞留带(等待图走不动的根因之一) nx = recog & 0xFFFF if recog > 0 else self.me["x"] ny = p1 & 0x7FF if (p1 & 0x800) else (p1 & 0xFFFF) self.me["x"], self.me["y"] = nx, ny self.me["last_x"], self.me["last_y"] = nx, ny self._last_sc634_t = time.time() # run17 修复: 防后续 SC50 立即覆盖 if self.me["mapname"] != old: self.monsters.clear() self.npcs.clear() self._send_map_load(self.me["mapname"]) log(f" [SC634] map={self.me['mapname']!r} MapId={self.me['map']!r} " f"@({nx},{ny})") ev.append(("map", self.me["mapname"], nx, ny)) elif mid == 55: # 引擎:MSG_SC_MISSION_CHANGE — 任务状态推送(引擎必处理帧) # flag: 0未接取 1进行中 2完成未提交; 无head=移除。 # 虚拟环境原则: 服务器推的任务体原样缓存, CS3031 提交时回构造它 # (sniffer 截断无法存模板, 真源只能是服务器自己)。 j = self._jbody(body) or {} if j.get("head"): old = self.missions.get(j.get("type")) self.missions[j["type"]] = j.get("flag", 1) self.mission_bodies[j["type"]] = j # 原样保留 if old != j.get("flag", 1): ev.append(("mission", j.get("type"), j.get("flag", 1))) elif j.get("type") in self.missions: del self.missions[j["type"]] ev.append(("mission_done", j["type"])) elif mid in (11, 13): # SC11/13 = 全体 actor 移动广播 — Lua真源 netActorController: # handle_MSG_SC_ACTOR_WALK/RUN: p={x=param1, y=param2}(目标格), # body.UserID 查表→SetMoveTo(p) 且 SetConfirmed(true)。 # 即: 对主玩家, SC11=服务器对 CS3011 的成功确认(目标格坐标)。 j = self._jbody(body) or {} uid = str(j.get("UserID", "")) myuid = self.me.get("uid") or "" myid = str(self.me.get("id") or "") sid_short = self.me.get("sid_short") or "" if uid and (uid == myuid or uid == myid or (sid_short and uid == sid_short)): # 布局(2026-09-23 抓包定案): recog=X, p1=Y, p2=dir — # 旧按 p1=X,p2=Y 解析错位, 自回显永远匹配不上。 x2 = recog & 0xFFFF if 0 <= recog <= 0xFFFF else recog & 0x7FF y2 = p1 & 0x7FF if (p1 & 0x800) else p1 self.me["x"], self.me["y"] = x2, y2 self.me["last_x"], self.me["last_y"] = x2, y2 self.confirmed_at = (x2, y2, time.time()) self._conf_pending = False # 移动已被服务器确认 ev.append(("self_move", x2, y2)) elif uid.isdigit() and int(uid) in self.monsters: y2 = p2 & 0x7FF if (p2 & 0x800) else p2 self.monsters[int(uid)]["x"] = p1 self.monsters[int(uid)]["y"] = y2 self.monsters[int(uid)]["t"] = time.time() elif uid.isdigit() and int(uid) in self.npcs: y2 = p2 & 0x7FF if (p2 & 0x800) else p2 self.npcs[int(uid)]["x"] = p1 self.npcs[int(uid)]["y"] = y2 elif mid == 14: # SC14 攻击广播 — cap3 实证: {"UserID":"<攻击者ID>","X":..,"Y":..}。 # 三用: ①怪打人→怪雷达补录; ②自ID学习(攻击者坐标==我刚攻击的位置 # →该 UserID 是自己的短会话ID, SC11 自回显匹配用); ③攻击确认。 j = self._jbody(body) or {} uid = str(j.get("UserID", "")) x, y = int(j.get("X", 0)), int(j.get("Y", 0)) y2 = y & 0x7FF if (y & 0x800) else y # 自ID学习: 我发出攻击后 ~2s 内, 广播坐标==我攻击时的位置 → 是我 atk = getattr(self, "_my_attack_echo", None) if atk and time.time() - atk[2] < 2.0 and abs(x - atk[0]) <= 1 and abs(y2 - atk[1]) <= 1: if self.me.get("sid_short") != uid: self.me["sid_short"] = uid ev.append(("self_id_learned", uid)) atk = None self._my_attack_echo = None elif uid == getattr(self.me, "sid_short", None): self.me["x"], self.me["y"] = x, y2 ev.append(("self_move_atk", x, y2)) elif uid.isdigit() and int(uid) < 1000000 and y2 > 50: aid = int(uid) if aid not in self.monsters: self.monsters[aid] = {"x": x, "y": y2, "race": 1049, "name": uid, "die": False, "t": time.time()} ev.append(("radar_mon", aid, x, y2)) else: self.monsters[aid]["x"], self.monsters[aid]["y"] = x, y2 self.monsters[aid]["t"] = time.time() elif mid in (10, 50): # SC10/SC50 进视野 — 分类对照 gameMapController.converActorData 真源: # recog = race | raceImg<<16, p2 = X | Y<<16, p3 = type | Dir<<16 # 实测 p3 常为 0xFF/0x100(type 位与 Lua 矛盾) → 怪物判定改用 race 段 # (>1000=怪, 实测新手怪 1049) + body cs 状态位(bit2=躺尸 bit1=骨头)。 x, y = lo16(p2), hi16(p2) j = self._jbody(body) or {} cs = j.get("cs", 0) or 0 die = bool(cs & 0b100) or bool(cs & 0b10) info = j.get("info") if mid == 50: self.me["id"] = p1 if info: self.me["uid"] = str(info[0]) # 关键修复(run17/18/19): SC634 后立即来的 SC50 位置经常错的(服务器 # 传送动画完成前就广播新图SC50,实际玩家还没落到p1/p2指定点)。SC28才是 # 真源 — 后到的SC28会把bot拉回(621,6)等真实坐标,导致 bot 用错误位置 # 走路/攻击被 SC28 反复拒绝。 # 处理: SC634 后 5s 内不更新 me['x'],me['y'],保留 SC634 设的位置。 # run17 修复: SC634 后 5s 内不更新 me['x'],me['y'],保留 SC634 设的位置 if p2 != 0: last_sc634 = getattr(self, "_last_sc634_t", 0) since = time.time() - last_sc634 if last_sc634 else 999 if not (last_sc634 and since < 5.0): self.me["x"], self.me["y"] = x, y ev.append(("me_born", self.me["x"], self.me["y"])) elif lo16(recog) > 1000: self.monsters[p1] = { "x": x, "y": y, "race": lo16(recog), "name": (info[2] if info and len(info) > 2 else str(p1)), "die": die, "t": time.time()} ev.append(("mon_in", p1, x, y)) elif info and len(info) >= 2: # TypeIndex = nActorId = 帧头 param1 (Lua 真源 gameMapController # converActorData: jsonData.nActorId = msgHdr.param1 → # actorUtils.SetActorProperty → actor:SetTypeIndex)。 # p3 lo16 是 jsonData.type(SetSubType), 实测常 0xFF — 旧代码误用 # 导致 CS1010 index 错误, 服务器静默不传送(2026-09-23 zone19 实证)。 type_index = p1 self.npcs[p1] = {"aid": str(info[0]), "name": info[1], "index": type_index, "x": x, "y": y} ev.append(("npc_in", str(info[0]), info[1], p1, type_index, x, y)) elif mid == 53: # 引擎:MSG_SC_PLAYER_HP_MP — 实测定案(diag): body=encode6( # "长uid&HP&maxHP&MP&maxMP|power"), UserID 是长uid(非短actorID) self.hp_samples = getattr(self, "hp_samples", []) if len(self.hp_samples) < 20: self.hp_samples.append(body[:80]) myuid = self.me.get("uid") or "" myid = str(self.me.get("id") or "") for v in CombatBot._body_variants(body): try: s = v.decode("utf-8") parts = s.split("|")[0].split("&") if len(parts) >= 3 and parts[0] in (myuid, myid): self.hp, self.max_hp = int(parts[1]), int(parts[2]) ev.append(("hp", self.hp, self.max_hp)) break except Exception: continue elif mid == 32: # SC32 死亡倒下 — Lua真源 handle_DeathFallDown: body=死亡actorID字符串, # 主玩家/怪共用此帧! 主玩家死亡→DIE状态, 等SC27复活/SC28回城拉回。 raw = body.decode("utf-8", "replace").strip() myuid = self.me.get("uid") or "" myid = str(self.me.get("id") or "") if raw and (raw == myuid or raw == myid): self.dead = True ev.append(("me_die",)) try: aid = int(raw) except ValueError: j = self._jbody(body) if isinstance(j, dict): try: aid = int(j.get("UserID", 0) or j.get("Id", 0) or 0) except (TypeError, ValueError): aid = 0 elif isinstance(j, int): aid = j else: aid = 0 if aid and aid in self.monsters: self.monsters[aid]["die"] = True # 击杀归属: SC32 是全区广播(不分谁杀)。只计"我近期攻击过且未死" # 的目标 — run6 实证: 视野内他人杀怪会让裸计数虚高(报2杀实际1杀)。 atk = getattr(self, "_my_hits", {}) last_hit_t = atk.get(aid, 0) if last_hit_t and time.time() - last_hit_t < 8.0: self.mon_kills += 1 ev.append(("my_kill", aid)) else: ev.append(("mon_die_other", aid)) ev.append(("mon_die", aid)) elif mid == 44: # SC44 经验(Lua: MSG_SC_PLAYER_EXP) — body=JSON{curr,changed,isNG} # (PlayerPropertyProxy:handle_MSG_SC_PLAYER_EXP, Run 16 实测 curr/变更字段名) j = self._jbody(body) or {} if isinstance(j, dict): new_exp = j.get("curr") or j.get("CurExp") or j.get("exp") changed = j.get("changed") or j.get("Changed") or 0 if new_exp is not None: old_exp = getattr(self, "_exp_total", None) self._exp_total = int(new_exp) delta = int(changed) if changed else 0 if old_exp is not None and delta > 0: log(f" [EXP] +{delta} → total={new_exp}") elif old_exp is None: log(f" [EXP] 当前={new_exp} (无变更)") ev.append(("exp", new_exp, delta)) else: log(f" [SC44] body={j}") elif mid == 200: # SC200 背包加物品(Lua: MSG_SC_BAG_ADD_ITEM) — 怪物掉落 j = self._jbody(body) or {} if isinstance(j, dict): name = j.get("Name") or j.get("name") or "?" count = j.get("Count") or j.get("count") or 1 stdmode = j.get("StdMode") or j.get("stdMode") or 0 log(f" [掉落] 背包+{name} ×{count} StdMode={stdmode}") self._drops = getattr(self, "_drops", 0) + 1 ev.append(("bag_add", name, count)) else: # 尝试直接用body字符串 raw = body.decode("utf-8", "replace").strip() if raw: log(f" [掉落] SC200 body={raw[:80]}") self._drops = getattr(self, "_drops", 0) + 1 elif mid == 610: # SC610 地图物品批量显示(Lua: MSG_SC_MAP_ITEM_DROP_BATCH) — 怪物掉落物在地上 j = self._jbody(body) if isinstance(j, dict): items = j.get("items") or j.get("Items") or j.get("itemList") or [] if items: for it in items[:5]: if isinstance(it, dict): log(f" [地面物品] {it.get('name') or it.get('Name') or '?'} @" f"({it.get('x') or it.get('X')},{it.get('y') or it.get('Y')})") ev.append(("ground_drop",)) elif mid == 643: # SC643 NPC对话(Lua真源 NPCProxy.handle_MSG_SC_NPC_TALK): # body=JSON{sMsg(SUI富文本), Name, UserID, index}。 # sMsg 含 , 点击=CS1011{UserID,Act,index}。 # content=="CloseNpcLayer" 表示对话被服务器关闭。 j = self._jbody(body) if isinstance(j, dict) and j.get("sMsg"): ev.append(("npc_talk", j)) elif mid == 27: # SC27 复活(Lua: MSG_SC_NET_PLAYER_REVIVE) — 清死亡态 self.dead = False ev.append(("me_revive",)) elif mid == 28: # SC28 拉回(Lua真源 MOVE_FAIL: p1/p2=服务器接受坐标, 客户端照收)。 # 两类异常帧(run27/29 漂移链条定案): # a) X/Y 反序: (291,620)→报(618,1), 反转(1,618)才在杀怪区 # b) (0,0) 清位帧: 传送后对旧走路帧的无效拉回, 坐标无意义 # 处理: (0,0)忽略(等SC11自回显/SC51校正); 其余按"离旧位置更近"的 # X/Y 解读采信, 防坐标污染。 if p1 == 0 and p2 == 0: ev.append(("pull_back_zero_ignored",)) else: # SC28 本义是"上一步被拒, 拉回最近接受格" → 距离必近(≤4)。 # 跳变大的帧是传送前后服务器按旧位置结算的过期拒绝 # (09:03 实证: 传回等待图(23,25)后0秒发步, 被按传送前(25,0)拒) # → 就近过滤丢弃, 真实位置由 SC11 自回显/SC51 维护。 old = (self.me["x"], self.me["y"]) y2 = p2 & 0x7FF if (p2 & 0x800) else p2 near = [c for c in ((p1, y2), (y2, p1)) if max(abs(c[0] - old[0]), abs(c[1] - old[1])) <= 4] if not near: self._sc28_drop = getattr(self, "_sc28_drop", 0) + 1 if self._sc28_drop <= 5: log(f" [SC28] 丢弃脏帧 raw=({p1},{y2}) old={old}") ev.append(("pull_back_bogus_dropped",)) else: best = near[0] self.me["x"], self.me["y"] = best self.me["dir"] = p3 & 0xFF # 服务器权威坐标校正: lastPoint 同步重置(后续 CS3011 p3 从这里续链) self.me["last_x"], self.me["last_y"] = best if getattr(self, "_sc28_log_n", 0) < 12: self._sc28_log_n = getattr(self, "_sc28_log_n", 0) + 1 log(f" [SC28] 拉回 old={old} → server={best} " f"(raw p1={p1} p2={y2})") ev.append(("pull_back", best[0], best[1])) else: ev += super()._dispatch(mid, recog, p1, p2, p3, body) return ev # ---------- 真实客户端的反馈帧 ---------- def _send_map_load(self, mapname): """传送/进图后回发 CS5012+CS5027(body=图名) — 引擎要求, 不发则无该图视野推送。 抓包实证 body 用协议名(比奇='0', 盟重='3', 等待图=中文名), 后跟一帧 body='cTh'。""" proto = proto_map_name({"mapname": mapname}) for mid in (5012, 5027): self.send(mid, 0, 0, 0, self.sid, proto.encode()) self.send(5012, 0, 0, 0, 32611848, b"cTh") self.me["mapname"] = mapname self.monsters.clear() self.npcs.clear() def _guess_map_name(self, x, y): """无预期目的地时按落点坐标猜协议图名(等待图y<50, 否则比奇)。""" if y < 50 and x < 50: return WAIT_MAP for fname, mname in (("0", "比奇大城"), ("3", "盟重土城")): m = _map_cached(fname) if m and m.in_bounds(x, y) and m.walkable(x, y): return mname return "0" def _send_good(self, x, y, d): """CS1: recog=新X p1=新Y p2=dir p3=32171624 tail=943 body=+GOOD/毫秒。""" self.good_ms += STEP_MS self.send(1, x & 0xFFFF, y & 0xFFFF, d, 32171624, f"+GOOD/{self.good_ms}".encode(), tail=943) # ---------- 客户端驱动走路(与真实客户端同构) ---------- def step_move(self, dx, dy, run=True): """CS3011 走1格 — run5/6/8 三轮实证通过的布局(等待图走位+比奇杀怪): recog=pack_point(起点), p1=0, p2=dir, p3=pack_point(上一点), body=MapId run13 反证: recog=目标 的"读法"连等待图都被拒。 成功确认 = 收到自己 UserID 的 SC11(p1=X目标, p2=Y目标); 超时由 walk 节奏控制(0.547s/步), 拒绝则 SC28 拉回并重置 last。""" d = next((i for i, (ox, oy) in enumerate(DIRS) if ox == dx and oy == dy), None) if d is None: return False x, y = self.me["x"], self.me["y"] nx, ny = x + dx, y + dy if not (0 <= nx <= 65535 and 0 <= ny <= 65535): return False lx, ly = self.me.get("last_x", x), self.me.get("last_y", y) body = proto_map_name(self.me).encode() self.send(3011, pack_point(x, y), 0, d, pack_point(lx, ly), body) if getattr(self, "_walk_log_n", 0) < 12: self._walk_log_n = getattr(self, "_walk_log_n", 0) + 1 log(f" [WALK] 3011 recog=({x},{y}) p2=dir{d} p3=({lx},{ly}) body={body!r}") self.me["last_x"], self.me["last_y"] = x, y self.me["x"], self.me["y"], self.me["dir"] = nx, ny, d self._step_pending_ack = (dx, dy, time.time() + STEP_MS / 1000.0) self._last_step_t = time.time() return True def step_toward(self, tx, ty): """战斗走位也必须登记 CS1 步末确认 — run28/31 实证: 战斗循环只 pump 不发 CS1 时, 服务器拒绝所有后续走路帧(SC28 拉回风暴, 角色原地站桩)。""" x0, y0 = self.me["x"], self.me["y"] arrived = super().step_toward(tx, ty) if (self.me["x"], self.me["y"]) != (x0, y0): self._step_pending_ack = (0, 0, time.time() + STEP_MS / 1000.0) self._last_step_t = time.time() return arrived def tick_step_ack(self): """步末 CS1 + 传送落地 CS1(延后1.5s, 对齐真实客户端时序)。 注意: CS1 是我们→服务器的 +GOOD 计数帧; 服务器→我们的行为确认也是 1 号帧 (SC1), 两者同号不同向。SC1 未到超 2s 视为确认超时(Lua 客户端 -5s 兜底同款)。""" la = self._landing_ack_due if la and time.time() >= la[3]: self._send_good(la[0], la[1], la[2]) self._landing_ack_due = None ca = self._step_pending_ack if ca and time.time() >= ca[2]: self._send_good(self.me["x"], self.me["y"], self.me["dir"]) self._step_pending_ack = None def walk_step_toward(self, tx, ty, run=True): """单步走向目标: 主策略=贪心直线(抓包实证真实客户端就是这么走的, 全程直线 (32,0)→(292,580), 由服务器 SC28 校正); 卡住才用 A*(文件网格可能与服务器 不一致 — 等待图服务器网格≠0139.map, 用户实证客户端有1k分辨率缩放)。 返回 (moved, arrived)。""" x, y = self.me["x"], self.me["y"] if abs(tx - x) <= 1 and abs(ty - y) <= 1: return False, True dx = max(-1, min(1, tx - x)) dy = max(-1, min(1, ty - y)) if self._stuck_steps() >= 4: nxt = self._astar_next(x, y, tx, ty) if nxt: dx = max(-1, min(1, nxt[0] - x)) dy = max(-1, min(1, nxt[1] - y)) else: dx, dy = self._sidestep(x, y, dx, dy) if dx and dy: m = self._current_map(x, y) if m and not (m.walkable(x + dx, y) or m.walkable(x, y + dy)): dx, dy = self._sidestep(x, y, dx, dy) moved = self.step_move(dx, dy, run=run) return moved, False def _stuck_steps(self): """最近连续无位移步数(SC28 拉回/走帧被拒都会体现为位置不动)。""" hist = getattr(self, "_pos_hist", None) if hist is None: hist = self._pos_hist = [] hist.append((self.me["x"], self.me["y"])) if len(hist) > 6: self._pos_hist = hist[-6:] if len(self._pos_hist) < 3: return 0 return 3 if len(set(self._pos_hist[-3:])) == 1 else 0 def _current_map(self, x, y): fname = map_file_for(self.me.get("mapname") or "", x, y) return _map_cached(fname) if fname else None def _astar_next(self, x, y, tx, ty): m = self._current_map(x, y) if not m or not m.walkable(x, y): return None path = m.find_path(x, y, tx, ty) return path[1] if path and len(path) > 1 else None def _sidestep(self, x, y, dx, dy): """贪心/A* 都失败时的垂直绕行方向。""" m = self._current_map(x, y) for cand in ((0, -1), (0, 1), (-1, 0), (1, 0)): if m is None or m.walkable(x + cand[0], y + cand[1]): return cand return dx, dy # ---------- NPC 发现 ---------- def find_guide_npc(self): """从视野表找新手接待NPC(名称含 接待/新手/指引)。""" for n in self.npcs.values(): if any(k in n["name"] for k in ("接待", "新手", "指引")): return n return list(self.npcs.values())[0] if self.npcs else None def find_spawn_npc(self): """从视野表找【点我刷怪】NPC(用户实测: 比奇无怪时点它可刷出大量怪物)。""" for n in self.npcs.values(): if "刷怪" in n["name"]: return n return None # ---------- NPC 对话(协议真源: NPCProxy.lua) ---------- # CS1010 {UserID:cfgID, index:typeIndex} 打开对话 → SC643 {sMsg,UserID,index} # sMsg 是 SUI 富文本(LexicalHelper), 按钮 link=@命令 → 点击 = # CS1011 {UserID, Act:"@命令", index:SC643回传的index} (ExecuteWithJsonData) def npc_talk_open(self, npc, timeout=5): """点 NPC 打开对话, 返回 SC643 对话体(dict)或 None。""" body = json.dumps({"UserID": npc["aid"], "index": npc["index"]}, separators=(",", ":")).encode() self.send(1010, 0, 0, 0, self.sid, body) deadline = time.time() + timeout while time.time() < deadline and self.alive: for e in self.pump(0.3): if e[0] == "npc_talk": return e[1] return None def npc_click_act(self, talk, act, timeout=5): """点击对话里的按钮(Act), 返回随后的 SC643(可能有新对话)或 None。""" payload = {"UserID": talk.get("UserID"), "Act": act, "index": talk.get("index")} self.send(1011, 0, 0, 0, self.sid, json.dumps(payload, separators=(",", ":")).encode()) deadline = time.time() + timeout while time.time() < deadline and self.alive: for e in self.pump(0.3): if e[0] == "npc_talk": return e[1] return None @staticmethod def parse_sui_acts(smsg): """从 SC643 sMsg(SUI 富文本)提取按钮/文本的 link 命令。 格式(LexicalHelper 文档注释): / """ if not smsg: return [] return re.findall(r"link=@([^\|<>]+)", smsg) # ============================================================================ # 等待图走动模式(--mode wait) — 2026-09-23 cap_gold 双向抓包 + Lua/C++ 源码定案 # # 抓包关键更正(旧 wait_wander6 "三帧组"理论证伪): # cap_gold 是 socket 级双向 hook(dir 字段全标 CS 系标记瑕疵): # - tail=817 的 m=1/m=28 帧是【服务器→客户端】方向: # SC1: recog=X, p1=Y, p2=dir, body=encode6("+GOOD/") 行为确认(落点) # SC28×2: recog=X, p1=Y 段起止点申报 # (Lua 真源 gamePlayerController.lua:304 handle_MSG_SC_NET_PLAYER_MOVE_FAIL: # param1/param2=服务器接受坐标; :58 注册 SC1=MSG_SC_ACTION_CONFIREM_SUCCESS) # - 客户端走一步【只发一帧】CS3011: # recog=0, p1=dir, p2=pack_point(起点), p3=sid, tail=主链递增, # body=图名UTF-8; 起点=上一步 SC1 确认落点(链式咬合) # - 心跳 CS60000 主链 tail, p3=sid; 服务器回显 60000(tail=0,p3=负数) # - 全程无 CS5013(AutoMoveBegin 寻路会话理论证伪, 客户端从未发送) # # 解锁仪式(每段成功走动的前置, cap_gold t=4985/5312/5692 三段同构): # CS3031(任务体) → CS1011 @renwu_mov,1 → 服务器把角色从滞留带(y=0)传送 # 到可走区权威落点(~(23,27), SC634/SC51 下发) → CS5012/5027(body=等待地图) # ×2组+cTh×2 → CS3013 首步(p2=权威落点) → SC1 确认 → CS3011 链(550ms/步) # 未做此仪式时服务器侧角色仍在滞留带 → SC28 拉回 y=0 全拒(run1 实证) # ============================================================================ SERVERLIST_URL = "https://list-box.dhsf.xqhuyu.com/box_serverlist/206923.txt" def pick_latest_zone(): """服务器列表里 state=1 且已开服的最新区(open_time 最大且<=now)。""" r = rq.get(SERVERLIST_URL, timeout=20) m = re.search(r"\{.*\}", r.text, re.S) if not m: raise SystemExit("[!] 区列表拉取失败") data = json.loads(m.group(0)) best = None def walk(o): nonlocal best if isinstance(o, dict): sid_, st, ot = o.get("serverid"), o.get("state"), o.get("open_time") if sid_ and st == 1 and ot and ot <= time.time(): nm = o.get("list_name") or o.get("server_name") or "" if best is None or ot > best[0]: best = (ot, nm, sid_) for v in o.values(): walk(v) elif isinstance(o, list): for v in o: walk(v) walk(data) if not best: raise SystemExit("[!] 区列表解析失败: 无 state=1 已开区") log(f"[新区] {best[1]} serverid={best[2]} " f"开服={time.strftime('%Y-%m-%d %H:%M', time.localtime(best[0]))}") return best[2] class TrueWalkBot(PushBot): """等待图走动: 3011+CS1 每步, 位置以 SC1(tail=817,+GOOD回显)确认为真源。""" def __init__(self, *a, **kw): super().__init__(*a, **kw) self.good_ms = 98996961125 + random.randint(0, 5000) self.move_confirmed = 0 self.move_rejected = 0 self.last_confirmed_at = None self.chain_pos = None # 走链起点(上一确认落点) def _send_good(self, x, y, d): """客户端 CS1: recog=落点X, p1=落点Y, p2=dir, p3=32368232, body=encode6('+GOOD/'), tail=主链递增(黄金抓包 t=5692.99 tail=167 实证)。""" self.good_ms += random.randint(750, 1000) body = encode6(f"+GOOD/{self.good_ms}".encode()) self.send(1, x & 0xFFFF, y & 0xFFFF, d, 32368232, body) def step_walk(self, d): """单步走动: 仅发 CS3011, 起点=当前链位(=server权威位置)。 cap_wait 验证: 客户端每步单发 3011(p2=起点pack, p3=sid, body=等待地图12B), 服务器回 SC1(tail=817, +GOOD回显, recog=落点X, p1=落点Y) 或 SC28(拉回, param1=X,param2=Y)。 """ if self.chain_pos is None: self.chain_pos = (self.me["x"], self.me["y"]) x, y = self.chain_pos dx, dy = DIRS[d] nx, ny = x + dx, y + dy self.send(3011, 0, d, pack_point(x, y), self.sid, WAIT_MAP.encode()) self.chain_pos = (nx, ny) log(f" >> CS3011 起点=({x},{y})→dir{d}→({nx},{ny})") def _dispatch(self, mid, recog, p1, p2, p3, body): if mid == 28: # SC28 MOVE_FAIL (Lua gamePlayerController.lua:316): param1=X, param2=Y, param3=dir self.move_rejected += 1 x, y = p1, p2 log(f" << SC28 #{self.move_rejected} pos=({x},{y}) dir={p3} 链={self.chain_pos}") self.chain_pos = (x, y) self.me["x"], self.me["y"] = x, y return [] ev = super()._dispatch(mid, recog, p1, p2, p3, body) if mid == 1: # SC1 行为确认 (Lua 316: netMessageDef:94 SC_ACTION_CONFIREM_SUCCESS): # 客户端发: recog=落点X, p1=落点Y, p2=dir, p3=32368232, body=encode6("+GOOD/") # 服务器回 SC1 也用 1 号帧, body=服务器确认的 +GOOD 回显, recog/p1/p2 同布局 self.move_confirmed += 1 self.me["x"], self.me["y"] = recog & 0xFFFF, p1 & 0xFFFF self.me["last_x"], self.me["last_y"] = self.me["x"], self.me["y"] self.chain_pos = (self.me["x"], self.me["y"]) self.last_confirmed_at = (self.me["x"], self.me["y"], time.time()) log(f" << SC{1 if False else 1} #{self.move_confirmed} ★确认 → " f"({self.me['x']},{self.me['y']})") return ev def enter_wait_world(s, jwt, uid, zone, name, port=0): """等待图模式专用进世界: TrueWalkBot + 心跳走 bot.send 主链 tail。""" for a in range(4): try: P.enter_game(s, jwt, zone) break except Exception: if a == 3: raise time.sleep(12) got = None for pt in ((port,) if port else (8021, 8061, 8001, 8041)): try: r = P.login_gate(zone, jwt, name, pt, uid) if r[0]: got = r break except SystemExit as e: log(f" 端口{pt}: {e}") except Exception as e: log(f" 端口{pt}: {str(e)[:60]}") if not got: raise SystemExit("[!] 登录服失败") key32, sid, rname, tkey, gp = got log(f"[连接] 登录端口OK 角色={rname} sid={sid} 游戏端口={gp}") g = P.enter_game_gate(zone, rname, tkey, sid, gp, uid) bot = TrueWalkBot(g, sid, rname) # 心跳经 bot.send: tail 与主链共用递增(黄金抓包实证: 主链 tail 全序连续, # 独立计数器 → 服务器判会话非法 → SC28 拒绝一切移动) P.heartbeat_loop(g, sid, sender=bot.send) t0 = time.time() while time.time() - t0 < 10: bot.pump(0.3) if 658 in bot.frame_log: break if 658 not in bot.frame_log: g.close() raise SystemExit("[!] SC658 超时, 世界未初始化") log("[初始化] SC658 到达, 发送17帧初始化") send_init_frames(bot, uid) for _ in range(24): bot.pump(0.5) if bot.me["id"] and bot.me.get("mapname"): break log(f"进世界: map={bot.me.get('mapname')} @{bot.me['x']},{bot.me['y']} " f"uid={bot.me.get('uid')}") return g, bot def run_wait_wander(s, jwt, uid, zone, name, port, minutes): """等待图走动(cap_wait 验证过的最小形态): 进世界后直接走链, 每步单发 CS3011, 服务器通过 SC1(确认, +GOOD回显) / SC28(MOVE_FAIL, 拉回) 告诉客户端真实位置。链位永远以服务端为准。 注意: 角色必须在可走区(0139.map 中 y>=3); 若角色服务端位置在滞留带 y0-2, 需先用 quest 流程 NPC 传比奇再 renwu_mov 回到可走区。""" g, bot = enter_wait_world(s, jwt, uid, zone, name, port) try: P.enable_auto_heal(g, bot.sid, want_on=True) log("喝药保护: on") # 等首批视野帧(SC10/50/51 稳定 server 位置) for _ in range(20): bot.pump(0.3) if bot.me.get("mapname") and bot.me["x"] + bot.me["y"] > 0: break # 把 SC1/SC28 收到的位置同步进 chain_pos if bot.me.get("mapname"): bot.chain_pos = (bot.me["x"], bot.me["y"]) log("=" * 64) log(f"[走链] server起始位=({bot.me['x']},{bot.me['y']}) " f"map={bot.me.get('mapname')}") log("=" * 64) t_end = time.time() + minutes * 60 last_step = 0.0 rep = time.time() heading = 5 while time.time() < t_end and bot.alive: bot.pump(0.05) now = time.time() if now - last_step >= 0.55: if bot.move_rejected > bot.move_confirmed: heading = random.choice([h for h in range(8) if h != heading]) log(f"[换向] 拒绝({bot.move_rejected})>确认({bot.move_confirmed}) → dir={heading}") bot.step_walk(heading) last_step = now if now - rep >= 10: rep = now fl = sorted(bot.frame_log.items(), key=lambda kv: -kv[1]) log(f"[STAT] pos=({bot.me['x']},{bot.me['y']}) " f"确认={bot.move_confirmed} 拒绝={bot.move_rejected} " f"SC1={bot.frame_log.get(1, 0)} SC28={bot.frame_log.get(28, 0)} " f"SC10={bot.frame_log.get(10, 0)} SC11={bot.frame_log.get(11, 0)} " f"SC50={bot.frame_log.get(50, 0)} Top8={fl[:8]}") log("=" * 64) log(f"[结束] pos=({bot.me['x']},{bot.me['y']}) " f"确认={bot.move_confirmed} 拒绝={bot.move_rejected}") fl = sorted(bot.frame_log.items(), key=lambda kv: -kv[1]) log(f"收帧Top12: {fl[:12]}") return bot.move_confirmed > 0 finally: g.close() def npc_talk_to_biqi(bot): """滞留带 → 点NPC → 传比奇(quest_bot3 实证链, 保留供比奇流程复用)。""" t0 = time.time() while time.time() - t0 < 8 and not bot.npcs: bot.pump(0.4) npc = bot.find_guide_npc() if not npc: return bot.me.get("mapname") == "比奇大城" aid = npc["aid"] for idx in [npc["index"], 35, 34]: bot.pending_map_ack = "比奇大城" bot.send(1010, 0, 0, 0, bot.sid, json.dumps( {"UserID": aid, "index": idx}, separators=(",", ":")).encode()) bot.send(5012, 0, 0, 0, bot.sid, b"0") bot.send(5027, 0, 0, 0, bot.sid, b"0") t1 = time.time() while time.time() - t1 < 8 and bot.alive: evs = bot.pump(0.3) if bot.me.get("mapname") == "比奇大城": bot.pending_map_ack = None return True talk = next((e[1] for e in evs if e[0] == "npc_talk"), None) if talk: acts = bot.parse_sui_acts(str(talk.get("sMsg", ""))) act = next((a for a in acts if a not in ("exit", "close")), None) if act: bot.npc_click_act(talk, act) bot.pending_map_ack = None if bot.me.get("mapname") == "比奇大城": return True return bot.me.get("mapname") == "比奇大城" def enter_world(s, jwt, uid, zone, name, port=0): """登录链 + 进世界(17帧初始化), 返回 (sock, bot)。失败重试对齐 combat_bot。""" for a in range(4): try: P.enter_game(s, jwt, zone) break except Exception: if a == 3: raise time.sleep(12) got = None for pt in ((port,) if port else (8021, 8061, 8001, 8041)): try: r = P.login_gate(zone, jwt, name, pt, uid) if r[0]: got = r break except SystemExit as e: log(f" 端口{pt}: {e}") except Exception as e: log(f" 端口{pt}: {str(e)[:60]}") if not got: raise SystemExit("[!] 登录服失败") key32, sid, rname, tkey, gp = got g = P.enter_game_gate(zone, rname, tkey, sid, gp, uid) bot = PushBot(g, sid, rname) P.heartbeat_loop(g, sid, bot.lock) # CS60000 常驻心跳(断=踢) t0 = time.time() while time.time() - t0 < 10: bot.pump(0.3) if 658 in bot.frame_log: break if 658 not in bot.frame_log: g.close() raise SystemExit("[!] SC658 超时, 世界未初始化") send_init_frames(bot, uid) for _ in range(24): bot.pump(0.5) if bot.me["id"] and bot.me.get("mapname"): break log(f"进世界: map={bot.me.get('mapname')} @{bot.me['x']},{bot.me['y']} uid={bot.me.get('uid')}") return g, bot def run_fsm(bot, g, kill_target, minutes): """主循环: 状态机由服务器推送驱动。""" deadline = time.time() + minutes * 60 act_3199 = lambda act: bot.send(3199, 0, 0, 0, bot.sid, json.dumps( {"act": act, "index": 104, "subid": 2}, separators=(",", ":")).encode()) # 喝药保护(真实客户端进图即开) state = P.enable_auto_heal(g, bot.sid, want_on=True) log(f"喝药保护: {state}") # ---- 阶段1: 任意出生图 — 对齐 cap3 真实客户端序列 ---- # cap3 实证: CS3031(带完整任务体,富文本在 docs/cs3031_template.txt) # → CS1011(@renwu_mov,1) → CS5012/5027(等待地图) # → SC800(传送,UserID=自己) → CS1010(NPC,index=35) → 比奇 mname = bot.me.get("mapname") or "" log(f"[1] 出生于 {mname} @{bot.me['x']},{bot.me['y']} — CS3031+@renwu_mov,1") # CS3031 任务提交体: 优先用服务器 SC55 推来的实时任务体(虚拟环境真源), # 否则退回抓包模板(docs/cs3031_template.txt, sniffer 截断 200/562B) if 1 in bot.mission_bodies: body3031 = json.dumps(bot.mission_bodies[1], separators=(",", ":")).encode() src = "SC55实时" else: tpl_path = os.path.join(HERE, "docs", "cs3031_template.txt") try: body3031 = open(tpl_path, "rb").read() except OSError: body3031 = json.dumps( {"offsetY": 0, "flag": 1, "type": 1, "body": {"content": "", "color": "#ffff00"}, "head": {"content": "", "color": "#ffff00"}}, separators=(",", ":")).encode() src = "模板" bot.send(3031, 0, 0, 0, bot.sid, body3031) log(f" CS3031 body 来源={src} len={len(body3031)}") bot.pump(0.5) bot.tick_step_ack() bot.send(1011, 0, 0, 0, bot.sid, json.dumps( {"Act": "@renwu_mov,1", "UserID": "999999999", "index": 999999999}, separators=(",", ":")).encode()) # cap3: 点击后客户端立即预判加载等待图(盟重→指引→等待地图的第一跳) if mname != WAIT_MAP: bot.pending_map_ack = WAIT_MAP bot.send(5012, 0, 0, 0, bot.sid, WAIT_MAP.encode()) bot.send(5027, 0, 0, 0, bot.sid, WAIT_MAP.encode()) bot.send(5012, 0, 0, 0, 32611848, b"cTh") bot.pump(0.5) bot.tick_step_ack() # 等待: SC801 传送(→等待图) 或 已在等待图无传送 t0 = time.time() while time.time() - t0 < 12 and bot.alive: bot.pump(0.3) bot.tick_step_ack() if bot.teleports: break if bot.me.get("mapname") == WAIT_MAP and time.time() - t0 > 3: break log(f" renwu_mov: teleports={bot.teleports} → {bot.me.get('mapname')} @{bot.me['x']},{bot.me['y']}") bot.pending_map_ack = None mname = bot.me.get("mapname") or "" if mname == WAIT_MAP: # 等待图: 客户端自己走向新手接待NPC(服务器对距离宽容, 走不到也能点) log("[2] 等待图 — 找新手接待NPC ...") t0 = time.time() while time.time() - t0 < 8 and not bot.npcs: bot.pump(0.4) npc = bot.find_guide_npc() if npc: log(f" NPC {npc['name']} aid={npc['aid']} @({npc['x']},{npc['y']})") t_end = time.time() + 30 while time.time() < t_end and bot.alive: bot.pump(0.1) bot.tick_step_ack() moved, arrived = bot.walk_step_toward(npc["x"], npc["y"], run=False) if arrived: break if bot.me.get("mapname") != WAIT_MAP: break time.sleep(max(0.0, STEP_MS / 1000.0 - (time.time() - bot._last_step_t))) npc = bot.find_guide_npc() aid = npc["aid"] if npc else None type_index = npc["index"] if npc else 35 log(f" [NPC] 新手指引 aid={aid} type_index={type_index} @({npc['x'] if npc else '?'},{npc['y'] if npc else '?'})") # 用 NPC 实际的 type_index (从 p3 解析, p3 = type | Dir<<16, lo16 = typeIndex) # cap3 实证: 矿工 index=35 是当时 typeIndex 值; 萌新2814/不同角色可能不同 # index 候选: 实测不同区/角色 NPC 的 typeIndex 不同(cap3=35, 逐鹿20区 # p3解析=192), 逐个尝试直到触发传送; 每次点击后若回 SC643 对话则点按钮。 idx_cands = [type_index] + [i for i in (35, 34) if i != type_index] base_tp = bot.teleports teleported = False for idx_try in idx_cands: bot.pending_map_ack = "比奇大城" bot.send(1010, 0, 0, 0, bot.sid, json.dumps( {"UserID": aid, "index": idx_try}, separators=(",", ":")).encode()) bot.send(5012, 0, 0, 0, bot.sid, b"0") bot.send(5027, 0, 0, 0, bot.sid, b"0") bot.send(5012, 0, 0, 0, 32611848, b"cTh") log(f"[3] CS1010 已发(aid={aid},index={idx_try})+预判比奇ACK — 等传送 ...") t0 = time.time() clicks = 0 while time.time() - t0 < 12 and bot.alive: evs = bot.pump(0.3) bot.tick_step_ack() if bot.teleports > base_tp or (bot.me.get("mapname") == "比奇大城" and time.time() - t0 > 1.0): teleported = True break talk = next((e[1] for e in evs if e[0] == "npc_talk"), None) if talk is None: continue smsg = str(talk.get("sMsg", "")) acts = bot.parse_sui_acts(smsg) log(f" [NPC对话] UserID={talk.get('UserID')} index={talk.get('index')}") log(f" [NPC对话] acts={acts}") log(f" [NPC对话] sMsg={smsg[:500]}") if clicks >= 3: break act = next((a for a in acts if a not in ("exit", "close", "guanbi")), None) if act: bot.npc_click_act(talk, act) clicks += 1 log(f" [NPC对话] 已点 @{act} (第{clicks}次)") bot.pending_map_ack = None if teleported or bot.me.get("mapname") == "比奇大城": break log(f" index={idx_try} 未传送, 换下一候选 ...") log(f" 传送后: {bot.me.get('mapname')} @{bot.me['x']},{bot.me['y']} 怪视野={len(bot.monsters)}") # ---- 比奇: 落点(293,619)疑为安全区 → 走到杀怪区(292,580)再挂机(斗士抓包同款) ---- mname = bot.me.get("mapname") or "" tx, ty = KILL_AREA if abs(tx - bot.me["x"]) > 6 or abs(ty - bot.me["y"]) > 6: log(f"[4] {mname} @{bot.me['x']},{bot.me['y']} — 走向杀怪区 {KILL_AREA}") t_end = time.time() + 240 while time.time() < t_end and bot.alive: bot.pump(0.1) bot.tick_step_ack() if bot.me.get("mapname") == WAIT_MAP: log(" [!] 被传回等待图, 中止走路") break _, arrived = bot.walk_step_toward(tx, ty, run=True) if arrived: break time.sleep(max(0.0, STEP_MS / 1000.0 - (time.time() - bot._last_step_t))) log(f" 走路结束 pos=({bot.me['x']},{bot.me['y']}) 怪视野={len(bot.monsters)}") # ---- 挂机杀怪 ---- # 机制结论(Lua源码+抓包): @startguaji/CS9102 只是开关声明, 打怪本体是 # 客户端 AI(AutoController BT链: 选怪→走位→CS3014)。bot 复刻这段 AI: # 视野选怪(race>1000) → step_toward 走近 → CS3014 普攻, SC32 计杀。 log("[5] 挂机开启(SUI声明+客户端AI) ...") act_3199("@startguaji") bot.send(5286, 0, 0, 0, bot.sid) bot.send(9102, 1, 0, 0, bot.sid) t0 = time.time() rep = 0.0 target = None wander_dst = None no_mon_since = None next_attack_t = 0.0 spawn_npc_used = 0 # 点【点我刷怪】NPC 次数(防死循环) while time.time() - t0 < 420 and bot.alive and time.time() < deadline: bot.pump(0.15) # 高频收帧: 怪在动, 虚拟环境必须实时 bot.tick_step_ack() # CS1 步末确认(run31 实证: 不发=走路全拒) k = bot.mon_kills if getattr(bot, "dead", False): # 主玩家死亡(run20 实证: 1级号被围殴致死, SC32 body=自己uid)。 # 真实客户端行为: 躺尸→回城复活(SC28 拉回复活点)。bot: 等复活, # 复活后清怪表重走杀怪区。 if target is not None: log(f" [!] 死亡(被 {target} 击杀) — 等待复活...") target = None bot.monsters.clear() time.sleep(1.0) continue # --- 客户端 AI: AutoFindActorBehavior → AutoFindSkillBehavior --- m = None if target is not None and target in bot.monsters: m = bot.monsters[target] if m["die"]: target, m = None, None if m is None: # 目标过滤: 曼哈顿距离≤15, y>50(比奇坐标系, 排除等待图残留伪目标) cands = [(abs(mm["x"] - bot.me["x"]) + abs(mm["y"] - bot.me["y"]), aid) for aid, mm in bot.monsters.items() if not mm["die"] and mm["y"] > 50 and abs(mm["x"] - bot.me["x"]) + abs(mm["y"] - bot.me["y"]) <= 15] if cands: target = min(cands)[1] m = bot.monsters[target] wander_dst = None no_mon_since = None log(f" 选定目标 #{target} {m['name']} @({m['x']},{m['y']})") if m is not None: dist = max(abs(m["x"] - bot.me["x"]), abs(m["y"] - bot.me["y"])) # 关键修复(run17): 把攻击半径放大到 dist<=3 (Lua 客户端物理攻击多数<=3格内有效); # 走路仅在 dist=4-5 之间触发,且严格在 KILL_AREA 范围内。 # 这样绝大多数情况下 (怪物固定在 619-622 区域) 都不需要走路,避免 SC28 风暴 if dist <= 3 and time.time() >= next_attack_t: bot.send_attack(target, m["x"], m["y"]) bot._my_hits = {**getattr(bot, "_my_hits", {}), target: time.time()} # 击杀归属登记 bot._my_attack_echo = (bot.me["x"], bot.me["y"], time.time()) next_attack_t = time.time() + 0.65 # HitTime=650ms elif 3 < dist <= 6 and time.time() >= next_attack_t - 0.30: tx, ty = m["x"], m["y"] myx, myy = bot.me["x"], bot.me["y"] in_safe_zone = (270 <= tx <= 310 and 605 <= ty <= 640 and 270 <= myx <= 310 and 605 <= myy <= 640) if in_safe_zone and _walkable_at(tx, ty): bot.step_toward(tx, ty) next_attack_t = time.time() + 0.35 else: log(f" [放弃目标] #{target} 移出安全区 @{tx},{ty} (我@{myx},{myy})") target = None next_attack_t = time.time() + 0.10 elif dist > 6: log(f" [放弃目标] #{target} 太远 dist={dist}") target = None next_attack_t = time.time() + 0.10 else: # 无怪 → 优先点【点我刷怪】NPC(用户实测可刷出大量怪), 次选游走找怪。 # NPC 对话要求距离≤8格(NPCLayerMediator.CheckCloseNpcTalkLayer), # 超距先走近再对话。 live_near = [mm for mm in bot.monsters.values() if not mm["die"] and mm["y"] > 50] if not live_near and spawn_npc_used < 3: snpc = bot.find_spawn_npc() if snpc is None: bot.pump(1.0) # 视野推送可能未到, 再收 1s snpc = bot.find_spawn_npc() if snpc is not None: dist_n = max(abs(snpc["x"] - bot.me["x"]), abs(snpc["y"] - bot.me["y"])) if dist_n > 5: bot.step_toward(snpc["x"], snpc["y"]) next_attack_t = time.time() + 0.35 else: spawn_npc_used += 1 log(f" [刷怪] 点NPC «{snpc['name']}» aid={snpc['aid']} " f"第{spawn_npc_used}次 @({snpc['x']},{snpc['y']})") talk = bot.npc_talk_open(snpc) if talk: acts = bot.parse_sui_acts(talk.get("sMsg", "")) log(f" [刷怪] SC643 对话 acts={acts[:5]} " f"sMsg前120={str(talk.get('sMsg'))[:120]}") # 点第一个非 exit/link 类命令(通常是"刷怪"本尊) act = next((a for a in acts if a not in ("exit", "renwu_mov")), None) if act: bot.npc_click_act(talk, act) log(f" [刷怪] 已点 Act=@{act} — 等怪进视野 ...") bot.pump(2.0) no_mon_since = None wander_dst = None else: log(" [刷怪] NPC 无对话回应") continue # 无怪游走找怪 — Lua真源 AssistProxy:checkAimlessMove 同款思路: # 随机可走点走过去刷新视野。本地化: 小范围(±12格, 新手杀怪区就 ~10格), # 且需稳定无怪 3s 才触发(传送落地后视野推送需要 1-2s, run38 实证 # 过早游走会背向怪群跑)。 # 关键修复(run15): 取消游走。视野无怪时仅原地等怪, 怪自然刷新。 # 原因: 游走半径±12会把bot带进(2,619)西城墙或东侧墙区,触发SC28拉回 if wander_dst is None and no_mon_since is None: no_mon_since = time.time() # 极端兜底: 60s 还无怪才尝试附近小范围(±3)一步试探 if wander_dst is None and no_mon_since and time.time() - no_mon_since > 60.0: for _ in range(5): nx = bot.me["x"] + random.randint(-3, 3) ny = bot.me["y"] + random.randint(-3, 3) if _walkable_at(nx, ny): wander_dst = (nx, ny) log(f" 游走找怪 → ({nx},{ny})") break no_mon_since = time.time() # 重置,避免连续触发 else: no_mon_since = None if wander_dst is not None: _, arrived = bot.walk_step_toward(*wander_dst, run=True) if arrived: wander_dst = None no_mon_since = None next_attack_t = time.time() + 0.35 if time.time() - rep > 10: rep = time.time() near = [(aid, mm["x"], mm["y"]) for aid, mm in bot.monsters.items() if mm["y"] > 50][:6] drift = getattr(bot, "drift_log", []) log(f"[FIGHT] pos=({bot.me['x']},{bot.me['y']}) map={bot.me.get('mapname')} kills={k} " f"SC32={bot.frame_log.get(32, 0)} SC14={bot.frame_log.get(14, 0)} " f"HP={bot.hp}/{bot.max_hp} 比奇怪={near}" + (f" 漂移={drift[-3:]}" if drift else "")) if kill_target and k >= kill_target: act_3199("@stopguaji") bot.send(5283, 0, 0, 0, bot.sid) bot.send(9102, 2, 0, 0, bot.sid) log(f"★ 击杀达标: {k}") bot.state = "DONE" break fl = sorted(bot.frame_log.items(), key=lambda kv: -kv[1]) # 关键帧(经验/掉落/背包/战斗)单独打印 key_frames = [(m, c) for m, c in fl if m in (44, 200, 202, 203, 610, 611, 1001, 870, 871, 53, 14, 32)] log(f"结束: kills={getattr(bot, 'mon_kills', 0)} map={bot.me.get('mapname')} " f"@({bot.me['x']},{bot.me['y']}) EXP={getattr(bot, '_exp_total', '?')} " f"掉落数={getattr(bot, '_drops', 0)}") log(f" 关键帧(经验/掉落/战斗): {key_frames}") log(f" Top10收帧: {fl[:10]}") log(f" 所有收帧: {fl}") return getattr(bot, "mon_kills", 0) def main(): ap = argparse.ArgumentParser() ap.add_argument("-u", "--user", default="longwu988") ap.add_argument("-p", "--password", default="Longwu1988") ap.add_argument("--mode", choices=("quest", "wait"), default="quest", help="quest=新手任务杀怪, wait=等待图走动") ap.add_argument("--zone", type=int, default=14492165, help="0=自动选服务器列表最新已开区") ap.add_argument("--name", default=None) ap.add_argument("--port", type=int, default=0) ap.add_argument("--kill", type=int, default=2) ap.add_argument("--minutes", type=float, default=2) args = ap.parse_args() zone = pick_latest_zone() if args.zone == 0 else args.zone name = args.name or ("萌新" + str(random.randint(1000, 9999))) j = None for a in range(5): try: s = rq.Session() j, _ = L.do_login(s, args.user, args.password) if j.get("code") == 200: break except Exception as e: log(f"登录重试{a + 1}: {str(e)[:60]}") time.sleep(12) if not j or j.get("code") != 200: raise SystemExit(f"[!] 登录失败: {(j or {}).get('msg', '网络异常')}") jwt = j["data"]["token"] uid = str(j["data"]["userInfo"]["id"]) if args.mode == "wait": ok = run_wait_wander(s, jwt, uid, zone, name, args.port, args.minutes) log(f"[完成] 等待图走动 {'成功' if ok else '未获服务器确认'} 区={zone} 角色={name}") return args.minutes = min(args.minutes, 2) # quest 模式快速测试上限(用户约定) g, bot = enter_world(s, jwt, uid, zone, args.name if args.name else name, args.port) try: kills = run_fsm(bot, g, args.kill or None, args.minutes) log(f"[完成] kills={kills}") finally: g.close() if __name__ == "__main__": main()