796 lines
34 KiB
Python
796 lines
34 KiB
Python
import json
|
||
import os
|
||
import random
|
||
import sys
|
||
import time
|
||
import math
|
||
import ctypes
|
||
import cv2
|
||
import numpy as np
|
||
import win32con
|
||
import win32gui
|
||
from hardware_control import hw_ctrl
|
||
from game_state import parse_game_state, load_layout_config
|
||
from coordinate_patrol import CoordinatePatrol
|
||
from death_manager import DeathManager
|
||
from logistics_manager import LogisticsManager
|
||
|
||
# 定义按键常量
|
||
KEY_TAB = '3'
|
||
KEY_LOOT = '4' # 假设你在游戏里设置了互动按键为 F
|
||
KEY_ATTACK = '2' # 假设你的主攻击技能是 1
|
||
DEFAULT_FOOD_KEY = 'f1' # 面包快捷键
|
||
DEFAULT_EAT_HP_THRESHOLD = 30
|
||
DEFAULT_EAT_MAX_WAIT_SEC = 30.0
|
||
|
||
# 巡逻点配置文件
|
||
WAYPOINTS_FILE = 'waypoints.json'
|
||
WIN_TITLE = "魔兽世界"
|
||
|
||
# 默认巡逻航点（waypoints.json 不存在或无效时使用）
|
||
DEFAULT_WAYPOINTS = [(23.8, 71.0), (27.0, 79.0), (31.0, 72.0)]
|
||
|
||
try:
|
||
ctypes.windll.shcore.SetProcessDpiAwareness(1)
|
||
except Exception:
|
||
ctypes.windll.user32.SetProcessDPIAware()
|
||
|
||
|
||
def _config_base():
|
||
"""打包成 exe 时，优先从 exe 同目录读配置，否则用内嵌资源目录。"""
|
||
if getattr(sys, 'frozen', False):
|
||
exe_dir = os.path.dirname(sys.executable)
|
||
if os.path.exists(exe_dir):
|
||
return exe_dir
|
||
return getattr(sys, '_MEIPASS', exe_dir)
|
||
return os.path.dirname(os.path.abspath(__file__))
|
||
|
||
|
||
def get_config_path(filename):
|
||
"""解析配置文件路径：先 exe 同目录，再打包内嵌目录。"""
|
||
base = _config_base()
|
||
p = os.path.join(base, filename)
|
||
if os.path.exists(p):
|
||
return p
|
||
if getattr(sys, 'frozen', False):
|
||
meipass = getattr(sys, '_MEIPASS', '')
|
||
if meipass:
|
||
p2 = os.path.join(meipass, filename)
|
||
if os.path.exists(p2):
|
||
return p2
|
||
return os.path.join(base, filename)
|
||
|
||
|
||
def move_cursor_hw(x, y, settle_sec=0.02):
|
||
hw_ctrl.move_to(int(x), int(y))
|
||
if settle_sec > 0:
|
||
time.sleep(settle_sec)
|
||
|
||
|
||
def get_wow_client_rect(hwnd):
|
||
client_left, client_top, client_right, client_bottom = win32gui.GetClientRect(hwnd)
|
||
screen_left, screen_top = win32gui.ClientToScreen(hwnd, (client_left, client_top))
|
||
screen_right, screen_bottom = win32gui.ClientToScreen(hwnd, (client_right, client_bottom))
|
||
return screen_left, screen_top, screen_right, screen_bottom
|
||
|
||
|
||
def load_waypoints(path=None):
|
||
path = path or get_config_path(WAYPOINTS_FILE)
|
||
if os.path.exists(path):
|
||
with open(path, 'r', encoding='utf-8') as f:
|
||
return json.load(f)
|
||
return []
|
||
|
||
|
||
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 CursorManager:
|
||
"""通过图像识别判断鼠标图标类型"""
|
||
def __init__(self):
|
||
self.templates = {}
|
||
self.handle_cache = {} # 句柄 -> 类型缓存
|
||
self.log_path = get_config_path('cursor_recognition.log')
|
||
self._load_templates()
|
||
|
||
def _load_templates(self):
|
||
# 强制使用纯相对路径，不带盘符前缀，由 Python 自动处理 CWD
|
||
cursor_dir = get_config_path(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:
|
||
# 使用 np.fromfile 读取为字节流，再解码，完美避开 OpenCV 的 imread 编码 Bug
|
||
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 _debug_log(self, message):
|
||
try:
|
||
with open(self.log_path, 'a', encoding='utf-8') as f:
|
||
f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {message}\n")
|
||
except Exception:
|
||
pass
|
||
|
||
def _normalize_for_match(self, img):
|
||
if img is None:
|
||
return None
|
||
if len(img.shape) == 2:
|
||
gray = img
|
||
elif img.shape[2] == 4:
|
||
alpha = img[:, :, 3].astype(np.float32) / 255.0
|
||
bgr = img[:, :, :3].astype(np.float32)
|
||
bg = np.full_like(bgr, 255.0)
|
||
composed = (bgr * alpha[..., None]) + (bg * (1.0 - alpha[..., None]))
|
||
gray = cv2.cvtColor(composed.astype(np.uint8), cv2.COLOR_BGR2GRAY)
|
||
else:
|
||
gray = cv2.cvtColor(img[:, :, :3], cv2.COLOR_BGR2GRAY)
|
||
return cv2.GaussianBlur(gray, (3, 3), 0)
|
||
|
||
def _edge_map(self, gray):
|
||
if gray is None:
|
||
return None
|
||
return cv2.Canny(gray, 32, 96)
|
||
|
||
def _identify(self, hcursor):
|
||
import win32ui
|
||
try:
|
||
# 捕获 48x48 区域，留出缩放空间
|
||
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
|
||
|
||
# 多尺度匹配：尝试 0.8, 1.0, 1.2 倍缩放，解决 UI 缩放问题
|
||
best_scale = 1.0
|
||
target_gray = self._normalize_for_match(target)
|
||
target_edge = self._edge_map(target_gray)
|
||
target_has_edge = target_edge is not None and np.count_nonzero(target_edge) > 0
|
||
scales = [0.55, 0.65, 0.75, 0.85, 0.95, 1.0, 1.1, 1.2, 1.35, 1.5]
|
||
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
|
||
|
||
interp = cv2.INTER_AREA if s < 1.0 else cv2.INTER_CUBIC
|
||
res_temp = cv2.resize(temp, (s_w, s_h), interpolation=interp)
|
||
temp_gray = self._normalize_for_match(res_temp)
|
||
if temp_gray is None:
|
||
continue
|
||
|
||
gray_res = cv2.matchTemplate(target_gray, temp_gray, cv2.TM_CCOEFF_NORMED)
|
||
_, gray_score, _, _ = cv2.minMaxLoc(gray_res)
|
||
score = gray_score
|
||
|
||
temp_edge = self._edge_map(temp_gray)
|
||
if target_has_edge and temp_edge is not None and np.count_nonzero(temp_edge) > 0:
|
||
edge_res = cv2.matchTemplate(target_edge, temp_edge, cv2.TM_CCOEFF_NORMED)
|
||
_, edge_score, _, _ = cv2.minMaxLoc(edge_res)
|
||
score = max(score, edge_score)
|
||
if score > max_score:
|
||
max_score = score
|
||
best_name = name
|
||
best_scale = s
|
||
|
||
if max_score > 0.2:
|
||
self._debug_log(f"best={best_name} score={max_score:.3f} scale={best_scale:.2f}")
|
||
print(f">>>> [雷达识别] 目标: {best_name} | 最高分: {max_score:.3f}")
|
||
return best_name, max_score
|
||
except Exception as e:
|
||
return 'Other', 0.0
|
||
|
||
|
||
class AutoBotMove:
|
||
def __init__(
|
||
self,
|
||
waypoints=None,
|
||
waypoints_path=None,
|
||
vendor_path=None,
|
||
attack_loop_path=None,
|
||
skinning_wait_sec=None,
|
||
food_key=None,
|
||
eat_hp_threshold=None,
|
||
eat_max_wait_sec=None,
|
||
stop_check=None,
|
||
resurrection_waypoints_path=None,
|
||
release_spirit_key=None,
|
||
resurrect_key=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.is_running = True
|
||
self.is_moving = False
|
||
self.target_acquired_time = None # 本次获得目标的时间，用于仅在做「跑向怪」时做卡死检测
|
||
# 记录上一帧是否处于战斗/有目标，用于检测“刚刚脱战”的瞬间
|
||
self._was_in_combat_or_target = False
|
||
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()
|
||
self.food_key = (food_key or DEFAULT_FOOD_KEY).strip().lower() or DEFAULT_FOOD_KEY
|
||
self.eat_hp_threshold = int(eat_hp_threshold) if eat_hp_threshold is not None else DEFAULT_EAT_HP_THRESHOLD
|
||
self.eat_max_wait_sec = float(eat_max_wait_sec) if eat_max_wait_sec is not None else DEFAULT_EAT_MAX_WAIT_SEC
|
||
self.attack_loop_config = load_attack_loop(attack_loop_path)
|
||
self._prev_death_state = 0
|
||
self._eating_started_at = None
|
||
self.last_turn_signal_time = 0.0
|
||
self.combat_signal_retry_sec = 1.0
|
||
self.turn_error_hold_sec = 0.8
|
||
self._last_mouse_path_scale_signature = None
|
||
# stop_check: 返回 True 表示需要立即停止（用于中断阻塞中的后勤/路线导航）
|
||
self._stop_check = stop_check if callable(stop_check) else (lambda: False)
|
||
if waypoints is None:
|
||
path = waypoints_path or get_config_path(WAYPOINTS_FILE)
|
||
waypoints = load_waypoints(path) or DEFAULT_WAYPOINTS
|
||
layout = load_layout_config()
|
||
self.patrol_controller = CoordinatePatrol(
|
||
waypoints,
|
||
mount_key=str(layout.get("mount_key", "x") or "x"),
|
||
mount_hold_sec=float(layout.get("mount_hold_sec", 1.6)),
|
||
mount_retry_after_sec=float(layout.get("mount_retry_after_sec", 2.0)),
|
||
enable_mount=bool(layout.get("enable_mount", True)),
|
||
)
|
||
self.death_manager = DeathManager(
|
||
self.patrol_controller, resurrection_waypoints_path,
|
||
release_spirit_key=layout.get('release_spirit_key', '9') if release_spirit_key is None else release_spirit_key,
|
||
resurrect_key=layout.get('resurrect_key', '0') if resurrect_key is None else resurrect_key,
|
||
)
|
||
vendor_file = vendor_path or get_config_path('vendor.json')
|
||
self.logistics_manager = LogisticsManager(vendor_file)
|
||
self.logistics_manager.bag_full_hearthstone = bool(layout.get("bag_full_hearthstone", False))
|
||
self.logistics_manager.hearthstone_key = str(layout.get("hearthstone_key", "b") or "b")
|
||
|
||
def _is_effective_target(self, state) -> bool:
|
||
"""
|
||
判断当前 `state['target']` 是否是“可攻击的有效目标”。
|
||
主要用 `target` 信号 + `target_hp` 是否为合理值，过滤掉 OCR/像素误判导致的假目标。
|
||
"""
|
||
if not state.get('target'):
|
||
return False
|
||
target_hp = state.get('target_hp')
|
||
if target_hp is None:
|
||
return False
|
||
try:
|
||
target_hp = int(target_hp)
|
||
except (TypeError, ValueError):
|
||
return False
|
||
# 目标血量为 0 基本可以视为无效/已死/不可交互目标
|
||
if target_hp <= 0:
|
||
return False
|
||
return True
|
||
|
||
def _should_stop(self) -> bool:
|
||
try:
|
||
return bool(self._stop_check())
|
||
except Exception:
|
||
return False
|
||
|
||
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.8)
|
||
|
||
# 2. 最后补漏剥皮（针对脚下尸体）
|
||
hw_ctrl.press(KEY_LOOT)
|
||
time.sleep(self.skinning_wait_sec + 0.5)
|
||
except Exception:
|
||
pass
|
||
|
||
def _load_mouse_path_points(self, client_width, client_height):
|
||
path_points = []
|
||
layout = load_layout_config()
|
||
base_width = int(layout.get('mouse_path_base_window_width', 2560) or 2560)
|
||
base_height = int(layout.get('mouse_path_base_window_height', 1600) or 1600)
|
||
path_file = get_config_path("loot_path.json")
|
||
if os.path.exists(path_file):
|
||
with open(path_file, 'r', encoding='utf-8') as f:
|
||
raw_path = json.load(f)
|
||
|
||
if isinstance(raw_path, dict):
|
||
path_points = raw_path.get('points') or raw_path.get('path_points') or raw_path.get('offsets') or []
|
||
base_width = int(
|
||
raw_path.get('base_window_width')
|
||
or raw_path.get('window_width')
|
||
or raw_path.get('base_client_width')
|
||
or base_width
|
||
)
|
||
base_height = int(
|
||
raw_path.get('base_window_height')
|
||
or raw_path.get('window_height')
|
||
or raw_path.get('base_client_height')
|
||
or base_height
|
||
)
|
||
else:
|
||
path_points = raw_path
|
||
|
||
if path_points:
|
||
scale_x = client_width / max(base_width, 1)
|
||
scale_y = client_height / max(base_height, 1)
|
||
signature = (client_width, client_height, base_width, base_height)
|
||
if signature != self._last_mouse_path_scale_signature:
|
||
print(
|
||
f">>> [扫雷路径] 当前窗口 {client_width}x{client_height}，"
|
||
f"基准 {base_width}x{base_height}，缩放 x={scale_x:.3f} y={scale_y:.3f}"
|
||
)
|
||
self._last_mouse_path_scale_signature = signature
|
||
|
||
scaled_points = []
|
||
for point in path_points:
|
||
if not isinstance(point, (list, tuple)) or len(point) < 2:
|
||
continue
|
||
dx = int(round(float(point[0]) * scale_x))
|
||
dy = int(round(float(point[1]) * scale_y))
|
||
scaled_points.append((dx, dy))
|
||
if scaled_points:
|
||
return scaled_points
|
||
|
||
x_scale, y_scale = 1.8, 0.8
|
||
for r in range(50, client_height // 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)))
|
||
return path_points
|
||
|
||
def mouse_sweep_scan(
|
||
self,
|
||
target_cursor_types,
|
||
click_wait_map=None,
|
||
max_scan_sec=15.0,
|
||
score_threshold=0.7,
|
||
return_on_first_click=False,
|
||
):
|
||
if random.random() < 0.1:
|
||
return False
|
||
hwnd = win32gui.FindWindow(None, WIN_TITLE)
|
||
if not hwnd:
|
||
return False
|
||
|
||
click_wait_map = click_wait_map or {}
|
||
target_cursor_types = set(target_cursor_types or [])
|
||
|
||
try:
|
||
left, top, right, bottom = get_wow_client_rect(hwnd)
|
||
client_width = max(right - left, 1)
|
||
client_height = max(bottom - top, 1)
|
||
center_x = left + (right - left) // 2
|
||
center_y = top + (bottom - top) // 2
|
||
|
||
move_cursor_hw(left + 50, top + 50, settle_sec=0.2)
|
||
_, default_hcursor, _ = win32gui.GetCursorInfo()
|
||
|
||
path_points = self._load_mouse_path_points(client_width, client_height)
|
||
start_time = time.time()
|
||
clicked_positions = []
|
||
|
||
for dx, dy in path_points:
|
||
if time.time() - start_time > max_scan_sec:
|
||
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 clicked_positions):
|
||
continue
|
||
|
||
move_cursor_hw(target_x, target_y, settle_sec=0.02)
|
||
|
||
_, hcursor, _ = win32gui.GetCursorInfo()
|
||
if hcursor == 0 or hcursor == default_hcursor:
|
||
if self._should_stop():
|
||
break
|
||
continue
|
||
|
||
ctype_name, score = self.cursor_mgr.get_type(hcursor)
|
||
if score > score_threshold and ctype_name in target_cursor_types:
|
||
print(f">>> [扫雷] 识别成功: {ctype_name} (得分: {score:.3f}), 执行右键点击")
|
||
hw_ctrl.right_click()
|
||
clicked_positions.append((target_x, target_y))
|
||
|
||
wait_sec = float(click_wait_map.get(ctype_name, 0.3))
|
||
if wait_sec > 0:
|
||
time.sleep(wait_sec)
|
||
|
||
if return_on_first_click:
|
||
move_cursor_hw(center_x, center_y, settle_sec=0.02)
|
||
return True
|
||
|
||
wait_start = time.time()
|
||
while time.time() - wait_start < 0.8:
|
||
_, curr_h, _ = win32gui.GetCursorInfo()
|
||
check_name, _ = self.cursor_mgr.get_type(curr_h)
|
||
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} < 阈值 {score_threshold})")
|
||
|
||
if self._should_stop():
|
||
break
|
||
|
||
move_cursor_hw(center_x, center_y, settle_sec=0.02)
|
||
return bool(clicked_positions)
|
||
except Exception as e:
|
||
print(f">>> [扫雷扫描] 出错: {e}")
|
||
return False
|
||
|
||
def mouse_scan_attack_target(self):
|
||
return self.mouse_sweep_scan(
|
||
['Attack'],
|
||
click_wait_map={'Attack': 0.3},
|
||
max_scan_sec=4.0,
|
||
score_threshold=0.6,
|
||
return_on_first_click=True,
|
||
)
|
||
|
||
def mouse_sweep_loot(self):
|
||
"""支持图标识别的高精度扫雷拾取。"""
|
||
return self.mouse_sweep_scan(
|
||
['LootAll', 'Skin'],
|
||
click_wait_map={'LootAll': 1.3, 'Skin': self.skinning_wait_sec},
|
||
max_scan_sec=15.0,
|
||
score_threshold=0.7,
|
||
return_on_first_click=False,
|
||
)
|
||
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. 强制“角落校准”采样
|
||
move_cursor_hw(left + 50, top + 50, settle_sec=0.2)
|
||
_, default_hcursor, _ = win32gui.GetCursorInfo()
|
||
|
||
# 2. 获取扫瞄路径点位
|
||
path_points = []
|
||
path_file = get_config_path("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
|
||
|
||
move_cursor_hw(target_x, target_y, settle_sec=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))
|
||
|
||
# 根据图标类型决定等待时间
|
||
wait_sec = 1.3 if ctype_name == 'LootAll' else self.skinning_wait_sec
|
||
time.sleep(wait_sec)
|
||
|
||
# 等待指针恢复
|
||
wait_start = time.time()
|
||
while time.time() - wait_start < 0.8:
|
||
_, curr_h, _ = win32gui.GetCursorInfo()
|
||
check_name, _ = self.cursor_mgr.get_type(curr_h)
|
||
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)")
|
||
|
||
if self._should_stop(): break
|
||
|
||
move_cursor_hw(center_x, center_y, settle_sec=0.02)
|
||
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)
|
||
|
||
def _start_eating(self):
|
||
"""开始就地吃面包恢复。"""
|
||
self._eating_started_at = time.time()
|
||
hw_ctrl.press(self.food_key)
|
||
|
||
def _should_keep_eating(self, state) -> bool:
|
||
"""
|
||
返回 True 表示继续等待回血（暂停巡逻与找怪）；
|
||
返回 False 表示结束进食等待。
|
||
"""
|
||
if self._eating_started_at is None:
|
||
return False
|
||
hp = state.get('hp')
|
||
if hp is not None and hp >= 100:
|
||
self._eating_started_at = None
|
||
return False
|
||
if (time.time() - self._eating_started_at) >= self.eat_max_wait_sec:
|
||
self._eating_started_at = None
|
||
return False
|
||
return True
|
||
|
||
def execute_logic(self, state):
|
||
if self._should_stop():
|
||
# 在停止按钮点击后，确保马上松开移动键，并避免继续执行后勤交互。
|
||
self.patrol_controller.stop_all()
|
||
self.logistics_manager.is_returning = False
|
||
self.is_moving = False
|
||
return
|
||
|
||
combat_error_signal = int(state.get('combat_error_signal', 255) or 255)
|
||
death = state.get('death_state', 0)
|
||
if self._prev_death_state in (1, 2) and death == 0:
|
||
self.death_manager.reset_when_alive()
|
||
self._prev_death_state = death
|
||
# 1. 死亡状态：尸体(1) 记录坐标并释放灵魂；灵魂(2) 跑尸
|
||
if death == 1:
|
||
self.patrol_controller.stop_all()
|
||
self.is_moving = False
|
||
self.patrol_controller.reset_stuck()
|
||
self.death_manager.on_death(state)
|
||
return
|
||
if death == 2:
|
||
get_state = (lambda: None if self._should_stop() else parse_game_state())
|
||
self.death_manager.run_to_corpse(state, get_state)
|
||
return
|
||
|
||
# 2. 后勤检查（脱战时）：空格或耐久不足则回城
|
||
self.logistics_manager.check_logistics(state)
|
||
if self.logistics_manager.is_returning:
|
||
if self.is_moving:
|
||
self.patrol_controller.stop_all()
|
||
self.is_moving = False
|
||
self.patrol_controller.reset_stuck()
|
||
# 勾选"包满炉石回城"：按炉石后触发停止回调
|
||
if self.logistics_manager.bag_full_hearthstone:
|
||
get_state_fn = (lambda: None if self._should_stop() else parse_game_state())
|
||
self.logistics_manager.use_hearthstone_and_stop(get_state=get_state_fn)
|
||
if callable(getattr(self, '_on_hearthstone_stop', None)):
|
||
self._on_hearthstone_stop()
|
||
return
|
||
# 中断策略：一旦 GUI 停止，后续 get_state 返回 None，使 navigate_path 立即退出。
|
||
get_state = (lambda: None if self._should_stop() else parse_game_state())
|
||
self.logistics_manager.run_route1_round(get_state, self.patrol_controller)
|
||
return
|
||
|
||
# 3. 战斗/有目标：停止移动，执行攻击逻辑；仅在「跑向怪」短窗口内做卡死检测
|
||
effective_target = self._is_effective_target(state)
|
||
# 核心修改：只要还在战斗中，就不算“完全脱战”，即使当前没目标（可能正在 Tab 找下一个）
|
||
in_combat_or_target = bool(state['combat'] or effective_target)
|
||
|
||
if in_combat_or_target:
|
||
# 被动进战时立即打断进食等待，转入正常战斗流程
|
||
self._eating_started_at = None
|
||
if self.target_acquired_time is None:
|
||
self.target_acquired_time = time.time()
|
||
if self.is_moving:
|
||
self.patrol_controller.stop_all()
|
||
self.is_moving = False
|
||
self.patrol_controller.reset_stuck()
|
||
|
||
# 只有在有“活的目标”时才执行攻击和交互
|
||
if effective_target:
|
||
target_hp = state.get('target_hp', 0)
|
||
if target_hp > 0:
|
||
now = time.time()
|
||
# 识别是否切换了新目标（血量跳变）
|
||
is_new_target = (target_hp > self.last_target_hp + 5)
|
||
if is_new_target:
|
||
self._has_braked_for_target = False
|
||
|
||
# 1. 掉血刹车逻辑：第一次发现目标掉血且正在移动，按一下 S 停住
|
||
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
|
||
|
||
# 根据战斗信号执行纠偏动作
|
||
if (
|
||
combat_error_signal == 100
|
||
and (now - self.last_interaction_time) >= self.combat_signal_retry_sec
|
||
):
|
||
hw_ctrl.press(KEY_LOOT)
|
||
self.last_interaction_time = now
|
||
|
||
if (
|
||
combat_error_signal == 50
|
||
and (now - self.last_turn_signal_time) >= self.combat_signal_retry_sec
|
||
):
|
||
hw_ctrl.keyDown('s')
|
||
time.sleep(self.turn_error_hold_sec)
|
||
hw_ctrl.keyUp('s')
|
||
self.last_turn_signal_time = now
|
||
|
||
self.last_target_hp = target_hp
|
||
# 执行正常的攻击循环
|
||
self.execute_combat_logic(state)
|
||
else:
|
||
# 目标已死亡但还在战斗中，按 Tab 找下一个目标
|
||
self.last_target_hp = 0
|
||
self.last_turn_signal_time = 0.0
|
||
self.last_interaction_time = 0.0
|
||
self._has_braked_for_target = False
|
||
if state['combat']:
|
||
hw_ctrl.press(KEY_TAB)
|
||
self.last_tab_time = time.time()
|
||
|
||
self._was_in_combat_or_target = True
|
||
return
|
||
else:
|
||
# 只有当 state['combat'] 和 effective_target 均为 False 时，才执行脱战逻辑
|
||
if self._was_in_combat_or_target:
|
||
# 此时已彻底脱战，执行拾取
|
||
self.execute_disengage_loot()
|
||
|
||
# 扫尾动作执行完后，本 tick 强制结束，防止立即按下 Tab
|
||
self._was_in_combat_or_target = False
|
||
self.target_acquired_time = None
|
||
self.last_turn_signal_time = 0.0
|
||
self.last_interaction_time = 0.0
|
||
self.last_tab_time = time.time() + 1.0 # 给找怪增加 1 秒额外冷却
|
||
return
|
||
|
||
self.target_acquired_time = None
|
||
self.last_turn_signal_time = 0.0
|
||
self.last_interaction_time = 0.0
|
||
self._was_in_combat_or_target = False
|
||
|
||
# 4. 脱战低血量：就地吃面包（最多等待 30 秒或回满）
|
||
hp = state.get('hp')
|
||
if self._eating_started_at is None and hp is not None and hp < self.eat_hp_threshold:
|
||
if self.is_moving:
|
||
self.patrol_controller.stop_all()
|
||
self.is_moving = False
|
||
self._start_eating()
|
||
if self._should_keep_eating(state):
|
||
# 进食期间不巡逻、不主动找目标（不按 Tab）
|
||
return
|
||
|
||
# 4. 没战斗没目标：巡逻（卡死检测在 patrol_controller.navigate 内）
|
||
self.is_moving = True
|
||
self.patrol_controller.navigate(state)
|
||
|
||
# 5. 顺便每隔几秒按一下 Tab（主动找怪）
|
||
if not effective_target and (time.time() - self.last_tab_time > 2.0):
|
||
hw_ctrl.press(KEY_TAB)
|
||
self.last_tab_time = time.time()
|
||
|
||
# 在 main 循环中使用：从 game_state 获取 state
|
||
if __name__ == "__main__":
|
||
bot = AutoBotMove()
|
||
try:
|
||
while True:
|
||
state = parse_game_state()
|
||
if state:
|
||
death_txt = ('存活', '尸体', '灵魂')[state.get('death_state', 0)]
|
||
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"战斗:{'Y' if state['combat'] else 'N'} 目标:{'Y' if state['target'] else 'N'} | "
|
||
f"空格:{state.get('free_slots', 0)} 耐久:{state.get('durability', 0):.0%} {death_txt} | "
|
||
f"x:{state['x']} y:{state['y']} 朝向:{state['facing']:.1f}°",
|
||
end=""
|
||
)
|
||
bot.execute_logic(state)
|
||
time.sleep(0.1)
|
||
except KeyboardInterrupt:
|
||
print("\n已停止。")
|