33dc741fd9
- 新增 wyhkm.dll 硬件盒子 COM 接口封装，支持键盘鼠标控制 - 修复 game_state_config.json 中 scan_region_width 过小导致截图越界的问题 - 添加鼠标路径录制器、硬件测试脚本等工具 - 更新多项配置默认值 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
350 lines
14 KiB
Python
350 lines
14 KiB
Python
import json
|
||
import os
|
||
import random
|
||
import sys
|
||
import time
|
||
import math
|
||
import ctypes
|
||
import cv2
|
||
import numpy as np
|
||
import win32gui
|
||
import win32api
|
||
import win32con
|
||
|
||
# 开启 DPI 意识
|
||
try:
|
||
ctypes.windll.shcore.SetProcessDpiAwareness(1)
|
||
except Exception:
|
||
ctypes.windll.user32.SetProcessDPIAware()
|
||
|
||
from hardware_control import hw_ctrl
|
||
from game_state import parse_game_state
|
||
from stuck_handler import StuckHandler
|
||
|
||
# 定义按键常量
|
||
KEY_TAB = '3'
|
||
KEY_LOOT = '4' # 假设你在游戏里设置了互动按键为 F
|
||
KEY_ATTACK = '2' # 假设你的主攻击技能是 1
|
||
WIN_TITLE = "魔兽世界"
|
||
|
||
def _config_base():
|
||
if getattr(sys, 'frozen', False):
|
||
return os.path.dirname(sys.executable)
|
||
return os.path.dirname(os.path.abspath(__file__))
|
||
|
||
class CursorManager:
|
||
"""通过图像识别判断鼠标图标类型"""
|
||
def __init__(self):
|
||
self.templates = {}
|
||
self.handle_cache = {}
|
||
self._load_templates()
|
||
|
||
def _load_templates(self):
|
||
# 强制使用纯相对路径，由 Python 自动处理 CWD，完美避开中文路径编码问题
|
||
cursor_dir = os.path.join('images', 'cursor')
|
||
files = {'Point': 'Point.PNG', 'Attack': 'Attack.PNG', 'LootAll': 'LootAll.PNG', 'Skin': 'Skin.PNG'}
|
||
for name, fname in files.items():
|
||
path = os.path.join(cursor_dir, fname)
|
||
if os.path.exists(path):
|
||
try:
|
||
# 使用 numpy 读取二进制流再解码
|
||
img_array = np.fromfile(path, dtype=np.uint8)
|
||
img = cv2.imdecode(img_array, cv2.IMREAD_UNCHANGED)
|
||
if img is not None:
|
||
if img.shape[2] == 3:
|
||
img = cv2.cvtColor(img, cv2.COLOR_BGR2BGRA)
|
||
self.templates[name] = img
|
||
except Exception as e:
|
||
print(f">>> [CursorMgr] 加载 {fname} 失败: {e}")
|
||
else:
|
||
print(f">>> [CursorMgr] 找不到文件: {path}")
|
||
|
||
def get_type(self, hcursor):
|
||
if not hcursor or hcursor == 0: return 'Other', 0.0
|
||
if hcursor in self.handle_cache: return self.handle_cache[hcursor]
|
||
res = self._identify(hcursor)
|
||
self.handle_cache[hcursor] = res
|
||
return res
|
||
|
||
def _identify(self, hcursor):
|
||
import win32ui
|
||
try:
|
||
width, height = 48, 48
|
||
hdc = win32gui.GetDC(0)
|
||
dc = win32ui.CreateDCFromHandle(hdc)
|
||
memdc = dc.CreateCompatibleDC()
|
||
bitmap = win32ui.CreateBitmap()
|
||
bitmap.CreateCompatibleBitmap(dc, width, height)
|
||
memdc.SelectObject(bitmap)
|
||
win32gui.DrawIconEx(memdc.GetSafeHdc(), 0, 0, hcursor, width, height, 0, 0, win32con.DI_NORMAL)
|
||
bits = bitmap.GetBitmapBits(True)
|
||
target = np.frombuffer(bits, dtype='uint8').reshape(height, width, 4)
|
||
memdc.DeleteDC()
|
||
win32gui.ReleaseDC(0, hdc)
|
||
|
||
best_name = 'Other'
|
||
max_score = 0.0
|
||
|
||
scales = [0.8, 1.0, 1.2]
|
||
for name, temp in self.templates.items():
|
||
t_h, t_w = temp.shape[:2]
|
||
for s in scales:
|
||
s_w, s_h = int(t_w * s), int(t_h * s)
|
||
if s_w > width or s_h > height: continue
|
||
res_temp = cv2.resize(temp, (s_w, s_h))
|
||
res = cv2.matchTemplate(target, res_temp, cv2.TM_CCOEFF_NORMED)
|
||
_, score, _, _ = cv2.minMaxLoc(res)
|
||
if score > max_score:
|
||
max_score = score
|
||
best_name = name
|
||
|
||
if max_score > 0.2:
|
||
print(f">>>> [雷达识别] 目标: {best_name} | 最高分: {max_score:.3f}")
|
||
return best_name, max_score
|
||
except Exception:
|
||
return 'Other', 0.0
|
||
|
||
def load_attack_loop(path):
|
||
"""从 JSON 加载攻击循环配置。支持 hp_below_percent/hp_key、mp_below_percent/mp_key（0=不启用）"""
|
||
if not path or not os.path.exists(path):
|
||
return None
|
||
try:
|
||
with open(path, 'r', encoding='utf-8') as f:
|
||
cfg = json.load(f)
|
||
steps = cfg.get('steps') or []
|
||
trigger = float(cfg.get('trigger_chance', 0.3))
|
||
hp_below = int(cfg.get('hp_below_percent', 0)) or 0
|
||
mp_below = int(cfg.get('mp_below_percent', 0)) or 0
|
||
return {
|
||
'trigger_chance': trigger,
|
||
'steps': steps,
|
||
'hp_below_percent': hp_below,
|
||
'hp_key': (cfg.get('hp_key') or '').strip() or None,
|
||
'mp_below_percent': mp_below,
|
||
'mp_key': (cfg.get('mp_key') or '').strip() or None,
|
||
}
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
class AutoBot:
|
||
def __init__(self, waypoints=None, attack_loop_path=None, skinning_wait_sec=None, enable_mouse_loot=True):
|
||
self.last_tab_time = 0
|
||
self.last_interaction_time = 0
|
||
self.last_target_hp = 0
|
||
self._has_braked_for_target = False
|
||
self._was_in_combat_or_target = False
|
||
self.is_running = True
|
||
self.attack_loop_config = load_attack_loop(attack_loop_path)
|
||
self.tab_no_target_count = 0 # 连续按 Tab 仍无目标的次数，满 5 次则转向
|
||
self.stuck_handler = StuckHandler()
|
||
self.target_acquired_time = None # 本次获得目标的时间（目前仅用于调试/保留接口）
|
||
self.last_turn_end_time = 0 # 最近一次结束 A/D 转向的时间，供卡死检测排除
|
||
self.skinning_wait_sec = float(skinning_wait_sec) if skinning_wait_sec is not None else 1.5
|
||
self.enable_mouse_loot = enable_mouse_loot
|
||
self.cursor_mgr = CursorManager()
|
||
|
||
def execute_disengage_loot(self):
|
||
"""从有战斗/目标切换到完全脱战的瞬间，执行拾取 + 剥皮。"""
|
||
try:
|
||
# 1. 拟人化鼠标扫雷补漏拾取
|
||
if self.enable_mouse_loot:
|
||
self.mouse_sweep_loot()
|
||
else:
|
||
hw_ctrl.press(KEY_LOOT)
|
||
time.sleep(0.5)
|
||
# 2. 剥皮动作
|
||
hw_ctrl.press(KEY_LOOT)
|
||
time.sleep(self.skinning_wait_sec + 0.5)
|
||
except Exception:
|
||
pass
|
||
|
||
def mouse_sweep_loot(self):
|
||
"""支持图标识别的高精度扫雷拾取。"""
|
||
if random.random() < 0.1: return False
|
||
hwnd = win32gui.FindWindow(None, WIN_TITLE)
|
||
if not hwnd: return False
|
||
|
||
try:
|
||
rect = win32gui.GetWindowRect(hwnd)
|
||
left, top, right, bottom = rect
|
||
center_x = left + (right - left) // 2
|
||
center_y = top + (bottom - top) // 2
|
||
|
||
# 1. 强制“角落校准”采样
|
||
win32api.SetCursorPos((left + 50, top + 50))
|
||
time.sleep(0.2)
|
||
_, default_hcursor, _ = win32gui.GetCursorInfo()
|
||
|
||
# 2. 获取扫瞄路径点位
|
||
path_points = []
|
||
path_file = "loot_path.json"
|
||
if os.path.exists(path_file):
|
||
with open(path_file, 'r') as f:
|
||
path_points = json.load(f)
|
||
else:
|
||
# 降级方案：生成拟人化半椭圆
|
||
x_scale, y_scale = 1.8, 0.8
|
||
for r in range(50, (bottom-top)//2, 40):
|
||
angles = range(180, 360, 5) if (r//40)%2==0 else range(360, 180, -5)
|
||
for a in angles:
|
||
rad = math.radians(a)
|
||
path_points.append((int(r * math.cos(rad) * x_scale), int(r * math.sin(rad) * y_scale)))
|
||
|
||
# 3. 开始沿路径扫瞄
|
||
start_time = time.time()
|
||
looted_positions = []
|
||
|
||
for dx, dy in path_points:
|
||
if time.time() - start_time > 15.0: break
|
||
|
||
target_x = center_x + dx + random.randint(-5, 5)
|
||
target_y = center_y + dy + random.randint(-5, 5)
|
||
|
||
if not (left+10 < target_x < right-10 and top+10 < target_y < bottom-10): continue
|
||
if any(math.dist((target_x, target_y), pos) < 30 for pos in looted_positions): continue
|
||
|
||
win32api.SetCursorPos((target_x, target_y))
|
||
time.sleep(0.02)
|
||
|
||
_, hcursor, _ = win32gui.GetCursorInfo()
|
||
if hcursor != 0 and hcursor != default_hcursor:
|
||
# 识别图标
|
||
ctype_name, score = self.cursor_mgr.get_type(hcursor)
|
||
if score > 0.7 and ctype_name in ['LootAll', 'Skin']:
|
||
print(f">>> [扫雷] 识别成功: {ctype_name} (得分: {score:.3f}), 执行右键点击")
|
||
hw_ctrl.right_click()
|
||
looted_positions.append((target_x, target_y))
|
||
|
||
# 根据类型等待
|
||
ws = 1.3 if ctype_name == 'LootAll' else self.skinning_wait_sec
|
||
time.sleep(ws)
|
||
|
||
# 等待指针恢复
|
||
w_start = time.time()
|
||
while time.time() - w_start < 0.8:
|
||
_, ch, _ = win32gui.GetCursorInfo()
|
||
check_name, _ = self.cursor_mgr.get_type(ch)
|
||
if check_name == 'Point': break
|
||
time.sleep(0.1)
|
||
time.sleep(random.uniform(0.1, 0.2))
|
||
elif score > 0.4:
|
||
print(f">>> [扫雷] 疑似图标: {ctype_name} (得分: {score:.3f} < 门槛 0.7)")
|
||
|
||
win32api.SetCursorPos((center_x, center_y))
|
||
return True
|
||
except Exception as e:
|
||
print(f">>> [扫雷拾取] 出错: {e}")
|
||
return False
|
||
|
||
def execute_combat_logic(self, state):
|
||
if self.attack_loop_config:
|
||
cfg = self.attack_loop_config
|
||
if random.random() >= cfg['trigger_chance']:
|
||
return
|
||
hp = state.get('hp')
|
||
if hp is not None and cfg.get('hp_below_percent') and cfg.get('hp_key') and hp < cfg['hp_below_percent']:
|
||
hw_ctrl.press(cfg['hp_key'])
|
||
time.sleep(0.2)
|
||
mp = state.get('mp')
|
||
if mp is not None and cfg.get('mp_below_percent') and cfg.get('mp_key') and mp < cfg['mp_below_percent']:
|
||
hw_ctrl.press(cfg['mp_key'])
|
||
time.sleep(0.2)
|
||
|
||
# 每次攻击前选中目标
|
||
# hw_ctrl.press(KEY_LOOT)
|
||
for step in cfg['steps']:
|
||
key = step.get('key') or KEY_ATTACK
|
||
delay = float(step.get('delay', 0.5))
|
||
hw_ctrl.press(key)
|
||
time.sleep(delay)
|
||
return
|
||
# 默认：模拟手动按键的节奏
|
||
if random.random() < 0.3:
|
||
hw_ctrl.press(KEY_ATTACK)
|
||
time.sleep(0.5)
|
||
|
||
def execute_logic(self, state):
|
||
current_time = time.time()
|
||
target_hp = state.get('target_hp', 0)
|
||
|
||
effective_target = bool(state['target'] and target_hp > 0)
|
||
in_combat_or_target = bool(state['combat'] or effective_target)
|
||
|
||
if in_combat_or_target:
|
||
self.target_acquired_time = (self.target_acquired_time or current_time)
|
||
self.tab_no_target_count = 0
|
||
|
||
if effective_target:
|
||
is_new_target = (target_hp > self.last_target_hp + 5)
|
||
if is_new_target:
|
||
self._has_braked_for_target = False
|
||
|
||
# 1. 刹车逻辑
|
||
if not self._has_braked_for_target and 0 < target_hp < self.last_target_hp:
|
||
hw_ctrl.keyDown('s')
|
||
time.sleep(0.05)
|
||
hw_ctrl.keyUp('s')
|
||
self._has_braked_for_target = True
|
||
|
||
# 2. 交互逻辑
|
||
cooldown = 2.0 if not state['combat'] else 6.0
|
||
if is_new_target or (current_time - self.last_interaction_time > cooldown):
|
||
hw_ctrl.press(KEY_LOOT)
|
||
self.last_interaction_time = current_time
|
||
self.last_target_hp = target_hp
|
||
if state['combat']:
|
||
self.execute_combat_logic(state)
|
||
|
||
self._was_in_combat_or_target = True
|
||
return
|
||
else:
|
||
if self._was_in_combat_or_target:
|
||
self.execute_disengage_loot()
|
||
self._was_in_combat_or_target = False
|
||
self.last_tab_time = current_time + 1.0
|
||
self.last_target_hp = 0
|
||
self._has_braked_for_target = False
|
||
return
|
||
|
||
self._was_in_combat_or_target = False
|
||
self.last_target_hp = 0
|
||
self._has_braked_for_target = False
|
||
|
||
self.tab_no_target_count = min(self.tab_no_target_count, 5)
|
||
if self.tab_no_target_count >= 5:
|
||
turn_key = random.choice(["a", "d"])
|
||
hw_ctrl.keyDown(turn_key)
|
||
time.sleep(random.uniform(0.3, 0.6))
|
||
hw_ctrl.keyUp(turn_key)
|
||
self.tab_no_target_count = 0
|
||
self.last_tab_time = current_time
|
||
self.last_turn_end_time = current_time
|
||
elif current_time - self.last_tab_time > random.uniform(0.8, 1.5):
|
||
hw_ctrl.press(KEY_TAB)
|
||
self.last_tab_time = current_time
|
||
self.tab_no_target_count += 1
|
||
|
||
# 在 main 循环中使用：从 game_state 获取 state
|
||
if __name__ == "__main__":
|
||
bot = AutoBot()
|
||
try:
|
||
while True:
|
||
state = parse_game_state()
|
||
if state:
|
||
target_hp = state.get('target_hp')
|
||
if target_hp is not None:
|
||
hp_part = f"血量:{state['hp']}% 目标血:{target_hp}% | 法力:{state['mp']}% | "
|
||
else:
|
||
hp_part = f"血量:{state['hp']}% | 法力:{state['mp']}% | "
|
||
print(
|
||
f"\r[状态] {hp_part}"
|
||
f"战斗:{'YES' if state['combat'] else 'NO '} | "
|
||
f"目标:{'YES' if state['target'] else 'NO '}",
|
||
end=""
|
||
)
|
||
bot.execute_logic(state)
|
||
time.sleep(0.1)
|
||
except KeyboardInterrupt:
|
||
print("\n已停止。")
|