315 lines
13 KiB
Python
315 lines
13 KiB
Python
import json
|
||
import os
|
||
import random
|
||
import sys
|
||
import time
|
||
import pydirectinput
|
||
|
||
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'
|
||
|
||
# 默认巡逻航点（waypoints.json 不存在或无效时使用）
|
||
DEFAULT_WAYPOINTS = [(23.8, 71.0), (27.0, 79.0), (31.0, 72.0)]
|
||
|
||
|
||
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 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 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,
|
||
):
|
||
self.last_tab_time = 0
|
||
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.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
|
||
# 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)),
|
||
)
|
||
self.death_manager = DeathManager(self.patrol_controller)
|
||
vendor_file = vendor_path or get_config_path('vendor.json')
|
||
self.logistics_manager = LogisticsManager(vendor_file)
|
||
|
||
def _should_stop(self) -> bool:
|
||
try:
|
||
return bool(self._stop_check())
|
||
except Exception:
|
||
return False
|
||
|
||
def execute_disengage_loot(self):
|
||
"""从有战斗/目标切换到完全脱战的瞬间，执行拾取 + 剥皮。"""
|
||
try:
|
||
# 拾取
|
||
pydirectinput.press(KEY_LOOT)
|
||
time.sleep(0.5)
|
||
# 剥皮
|
||
pydirectinput.press(KEY_LOOT)
|
||
time.sleep(self.skinning_wait_sec)
|
||
except Exception:
|
||
pass
|
||
|
||
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']:
|
||
pydirectinput.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']:
|
||
pydirectinput.press(cfg['mp_key'])
|
||
time.sleep(0.2)
|
||
|
||
# 每次攻击前选中目标
|
||
pydirectinput.press(KEY_LOOT)
|
||
for step in cfg['steps']:
|
||
key = step.get('key') or KEY_ATTACK
|
||
delay = float(step.get('delay', 0.5))
|
||
pydirectinput.press(key)
|
||
time.sleep(delay)
|
||
return
|
||
# 默认：模拟手动按键的节奏
|
||
if random.random() < 0.3:
|
||
pydirectinput.press(KEY_ATTACK)
|
||
|
||
def _start_eating(self):
|
||
"""开始就地吃面包恢复。"""
|
||
self._eating_started_at = time.time()
|
||
pydirectinput.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
|
||
|
||
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:
|
||
self.death_manager.run_to_corpse(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()
|
||
# 中断策略：一旦 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. 战斗/有目标：停止移动，执行攻击逻辑；仅在「跑向怪」短窗口内做卡死检测
|
||
in_combat_or_target = bool(state['combat'] or state['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()
|
||
# 跑向怪阶段的卡死检测：只在目标血量为 100% 时认为是在“刚刚开怪、接近怪物”，避免残血目标也触发跑向怪卡死逻辑
|
||
target_hp = state.get('target_hp')
|
||
if target_hp is not None and target_hp >= 100:
|
||
try:
|
||
if self.patrol_controller.stuck_handler.check_stuck(
|
||
state, self.patrol_controller.last_turn_end_time
|
||
):
|
||
print(">>> [战斗] 跑向怪时可能卡住障碍物，执行脱困")
|
||
self.patrol_controller.stop_all()
|
||
self.patrol_controller.stuck_handler.resolve_stuck()
|
||
self.patrol_controller.reset_stuck()
|
||
return
|
||
except Exception:
|
||
pass
|
||
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()
|
||
x, y = state.get('x'), state.get('y')
|
||
if x is not None and y is not None:
|
||
try:
|
||
current_pos = (float(x), float(y))
|
||
# max_ahead: 前方查看 10 个点；max_dist: 最大允许接入距离 10
|
||
self.patrol_controller.snap_to_forward_waypoint(
|
||
current_pos, max_ahead=10, max_dist=10.0, skip_current=True
|
||
)
|
||
except Exception:
|
||
pass
|
||
self.target_acquired_time = None # 脱战/无目标时清零，下次获得目标再计时
|
||
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 state['target'] and (time.time() - self.last_tab_time > 2.0):
|
||
pydirectinput.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已停止。") |