# 模块1加密狗绑定 class DongleBinder:def __init__(self, vid, pid, serialNone):self.vid vidself.pid pidself.serial serialdef detect(self):try:import usb.coredev usb.core.find(idVendorself.vid, idProductself.pid)if dev is None:return Falseif self.serial and self.serial ! YOUR_SERIAL:try:import usb.utilserial usb.util.get_string(dev, dev.iSerialNumber)return serial self.serialexcept:passreturn Trueexcept:return self._fallback_check()def _fallback_check(self):bind_file Path(./dongle_bind.dat)if not bind_file.exists():return Falsetry:data json.loads(bind_file.read_text())return data.get(bound, False)except:return False# 模块2游戏检测 class GameDetector:def __init__(self, process_names):self.process_names process_namesdef is_running(self):try:for name in self.process_names:result subprocess.run([tasklist, /FI, fIMAGENAME eq {name}],capture_outputTrue, textTrue, creationflagssubprocess.CREATE_NO_WINDOW)if name in result.stdout:return Truereturn Falseexcept:return Falsedef wait_for_game(self, check_interval1.0):print(⏳ 等待英雄联盟启动...)while True:if self.is_running():print(✅ 检测到游戏启动)return Truetime.sleep(check_interval)# 模块3真实LCU客户端 class LCUClient:def __init__(self):self.port Noneself.token Noneself.base_url Noneself._connected Falsedef connect(self):port, token get_lcu_credentials()if port and token:self.port portself.token tokenself.base_url fhttps://127.0.0.1:{port}self._connected Truereturn Truereturn Falsedef _request(self, method, endpoint, dataNone):if not self._connected:if not self.connect():return Noneurl f{self.base_url}{endpoint}auth (riot, self.token)headers {Content-Type: application/json}try:resp requests.request(method, url, authauth, headersheaders,jsondata, verifyFalse, timeout3)if resp.status_code 200:return resp.json()except:passreturn Nonedef get_active_player(self):return self._request(GET, /liveclientdata/activeplayer)def get_all_game_data(self):return self._request(GET, /liveclientdata/allgamedata)def get_events(self):return self._request(GET, /liveclientdata/eventdata)# 模块4真实事件检测器 dataclassclass GameEvent:游戏事件数据结构event_name: strevent_time: floatkiller: Optional[str] Nonevictim: Optional[str] Noneis_critical: bool Falseis_turret_kill: bool Falseis_first_blood: bool Falseassisters: List[str] field(default_factorylist)position: Dict[str, float] field(default_factorydict)raw_data: Dict field(default_factorydict)class EventDetector:基于真实LCU字段的事件检测def __init__(self):# 暴击缓冲区存储暴击时间戳self._crit_timestamps deque(maxlen50)# 击杀缓冲区存储击杀事件self._kill_events deque(maxlen30)# 上一次触发时间防重复self._last_trigger {critical: 0,penta: 0,low_success: 0}# 五杀状态self._penta_killer Noneself._penta_count 0self._penta_start_time 0def parse_lcu_event(self, raw_event: Dict) - Optional[GameEvent]:解析LCU事件为标准化格式try:event_name raw_event.get(EventName, )event_time raw_event.get(EventTime, time.time())# ---------- 击杀事件 ----------if event_name ChampionKill:return GameEvent(event_nameevent_name,event_timeevent_time,killerraw_event.get(KillerName),victimraw_event.get(VictimName),is_criticalraw_event.get(IsCritical, False),is_turret_killraw_event.get(IsTurretKill, False),is_first_bloodraw_event.get(IsFirstBlood, False),assistersraw_event.get(Assisters, []),positionraw_event.get(Position, {}),raw_dataraw_event)# ---------- 五杀事件 ----------if event_name PentaKill:return GameEvent(event_nameevent_name,event_timeevent_time,killerraw_event.get(KillerName),raw_dataraw_event)# ---------- 暴击伤害事件某些版本有独立暴击事件 ----------if Critical in event_name or event_name CriticalStrike:return GameEvent(event_nameCriticalStrike,event_timeevent_time,killerraw_event.get(ChampionName),is_criticalTrue,raw_dataraw_event)except Exception as e:print(f解析事件失败: {e})return Nonedef process_event(self, event: GameEvent) - List[Tuple[str, Dict]]:处理事件返回触发的事件列表返回: [(event_type, payload), ...]triggers []now time.time()# ---------- 1. 暴击检测 ----------if event.is_critical or event.event_name CriticalStrike:self._crit_timestamps.append(event.event_time or now)crit_rate self._calculate_crit_rate()if crit_rate CONFIG[CRITICAL_STRIKE_THRESHOLD]:if now - self._last_trigger[critical] CONFIG[EVENT_COOLDOWN]:self._last_trigger[critical] nowtriggers.append((critical_strike, {rate: crit_rate,killer: event.killer,victim: event.victim,count: len(self._crit_timestamps)}))# ---------- 2. 击杀记录用于五杀检测 ----------if event.event_name ChampionKill and event.killer:self._kill_events.append({time: event.event_time or now,killer: event.killer,victim: event.victim,is_critical: event.is_critical})# 检测五杀penta_result self._detect_penta()if penta_result:if now - self._last_trigger[penta] 60: # 五杀冷却60秒self._last_trigger[penta] nowtriggers.append((penta_kill, penta_result))# ---------- 3. 直接五杀事件 ----------if event.event_name PentaKill:if now - self._last_trigger[penta] 60:self._last_trigger[penta] nowtriggers.append((penta_kill, {killer: event.killer,event_time: event.event_time}))return triggersdef _calculate_crit_rate(self) - float:计算暴击速度次/秒now time.time()window CONFIG[CRITICAL_WINDOW]min_count CONFIG.get(MIN_CRIT_COUNT, 3)# 取窗口内的时间戳recent [t for t in self._crit_timestamps if now - t window]if len(recent) min_count:# 用最早和最晚的时间差计算速度time_span recent[-1] - recent[0]if time_span 0.1:return len(recent) / time_spanreturn 0.0def _detect_penta(self) - Optional[Dict]:检测五杀30秒内同一玩家5次击杀now time.time()if len(self._kill_events) 5:return None# 按击杀者分组统计30秒内的击杀killer_counts {}first_kill_time {}last_kill_time {}for kill in list(self._kill_events)[-20:]:if now - kill[time] 30:continuekiller kill[killer]killer_counts[killer] killer_counts.get(killer, 0) 1if killer not in first_kill_time:first_kill_time[killer] kill[time]last_kill_time[killer] kill[time]for killer, count in killer_counts.items():if count 5:return {killer: killer,kill_count: count,time_window: 30,first_kill: first_kill_time.get(killer),last_kill: last_kill_time.get(killer)}return Nonedef reset(self):重置状态游戏结束后调用self._crit_timestamps.clear()self._kill_events.clear()self._last_trigger {k: 0 for k in self._last_trigger}# 模块5LCU WebSocket真实订阅class LCUWebSocket:LCU WebSocket 实时事件订阅def __init__(self, event_detector: EventDetector):self.port Noneself.token Noneself.ws Noneself.connected Falseself.event_detector event_detectorself._thread Noneself._running Falseself._callbacks {critical_strike: [],penta_kill: [],game_start: [],game_end: []}def connect(self) - bool:连接WebSocketport, token get_lcu_credentials()if not port or not token:return Falseself.port portself.token tokenimport websocketws_url fwss://127.0.0.1:{port}/# 生成Basic认证import base64auth_str friot:{token}auth_b64 base64.b64encode(auth_str.encode()).decode()headers {Authorization: fBasic {auth_b64}}try:self.ws websocket.WebSocketApp(ws_url,headerheaders,on_openself._on_open,on_messageself._on_message,on_errorself._on_error,on_closeself._on_close)self._running Trueself._thread threading.Thread(targetself.ws.run_forever, daemonTrue)self._thread.start()# 等待连接timeout 5while not self.connected and timeout 0:time.sleep(0.5)timeout - 0.5return self.connectedexcept Exception as e:print(fWebSocket连接失败: {e})return Falsedef _on_open(self, ws):self.connected Trueprint(✅ LCU WebSocket 连接成功)# 订阅真实路径self._subscribe(/lol-gameflow/v1/gameflow-phase)self._subscribe(/lol-game-events/v1/game-events)self._subscribe(/lol-kill-streak/v1/kill-streaks)def _subscribe(self, endpoint: str):if self.ws:try:msg json.dumps([5, endpoint])self.ws.send(msg)except:passdef _on_message(self, ws, message):try:data json.loads(message)if len(data) 3:op, path, payload data[0], data[1], data[2]# 游戏阶段if path /lol-gameflow/v1/gameflow-phase:self._handle_game_phase(payload)# 游戏事件if path /lol-game-events/v1/game-events and payload:self._handle_game_events(payload)# 连杀信息if path /lol-kill-streak/v1/kill-streaks:self._handle_kill_streak(payload)except json.JSONDecodeError:passexcept Exception as e:print(fWebSocket消息处理错误: {e})def _handle_game_phase(self, phase: str):if phase InProgress:print( 游戏开始)self.event_detector.reset()for cb in self._callbacks[game_start]:try:cb()except:passelif phase Ended:print( 游戏结束)for cb in self._callbacks[game_end]:try:cb()except:passdef _handle_game_events(self, payload: Dict):events payload.get(Events, [])for raw_event in events:# 用事件检测器解析parsed self.event_detector.parse_lcu_event(raw_event)if parsed:triggers self.event_detector.process_event(parsed)for event_type, payload_data in triggers:if event_type in self._callbacks:for cb in self._callbacks[event_type]:try:cb(payload_data)except Exception as e:print(f回调执行错误: {e})def _handle_kill_streak(self, payload: Dict):处理连杀信息辅助五杀检测# 某些版本连杀信息包含五杀passdef _on_error(self, ws, error):if Connection not in str(error):print(fWebSocket错误: {error})def _on_close(self, ws, close_status_code, close_msg):self.connected Falseif self._running:print(⚠️ WebSocket已断开尝试重连...)threading.Thread(targetself._reconnect, daemonTrue).start()def _reconnect(self):time.sleep(3)if self._running:self.connect()def on(self, event_type: str, callback: Callable):注册事件回调if event_type in self._callbacks:self._callbacks[event_type].append(callback)def disconnect(self):self._running Falseif self.ws:self.ws.close()