# 模块6环形录屏 class RingBufferRecorder:def __init__(self, fps30, buffer_sec30, output_dir./recordings):self.fps fpsself.buffer_sec buffer_secself.output_dir Path(output_dir)self.output_dir.mkdir(parentsTrue, exist_okTrue)self.frame_buffer deque(maxlenfps * buffer_sec * 2)self.recording Falseself.thread Noneself._frame_count 0self._last_save_time 0def start(self):if self.recording:returnself.recording Trueself.thread threading.Thread(targetself._record_loop, daemonTrue)self.thread.start()print( 录屏已启动环形缓冲)def stop(self):self.recording Falseif self.thread:self.thread.join(timeout3)print( 录屏已停止)def _record_loop(self):try:import mssimport numpy as npimport cv2with mss.mss() as sct:monitor sct.monitors[1]frame_interval 1.0 / self.fpswhile self.recording:start_time time.time()img sct.grab(monitor)frame np.array(img)frame cv2.cvtColor(frame, cv2.COLOR_BGRA2BGR)self.frame_buffer.append(frame)self._frame_count 1elapsed time.time() - start_timesleep_time frame_interval - elapsedif sleep_time 0:time.sleep(sleep_time)except ImportError:self._record_loop_fallback()except Exception as e:print(f录屏错误: {e})self._record_loop_fallback()def _record_loop_fallback(self):降级录屏方案try:from PIL import ImageGrabimport numpy as npwhile self.recording:start_time time.time()img ImageGrab.grab()frame np.array(img)frame frame[:, :, ::-1] # RGB to BGRself.frame_buffer.append(frame)self._frame_count 1elapsed time.time() - start_timesleep_time 1.0 / self.fps - elapsedif sleep_time 0:time.sleep(sleep_time)except Exception as e:print(f降级录屏失败: {e})# 如果录屏完全失败生成测试帧while self.recording:import numpy as npframe np.random.randint(0, 255, (1080, 1920, 3), dtypenp.uint8)self.frame_buffer.append(frame)time.sleep(1.0 / self.fps)def save_clip(self, event_name: str, seconds_before: int 30, seconds_after: int 30) - List[str]:保存事件视频片段if len(self.frame_buffer) 0:return []# 防止频繁保存now time.time()if now - self._last_save_time 1:return []self._last_save_time nowbuffer_list list(self.frame_buffer)total_frames self.fps * (seconds_before seconds_after)start_idx max(0, len(buffer_list) - total_frames)clip_frames buffer_list[start_idx:]if len(clip_frames) self.fps * 2: # 至少2秒return []timestamp datetime.now().strftime(%Y%m%d_%H%M%S_%f)[:-3]filename self.output_dir / f{event_name}_{timestamp}.mp4try:import cv2h, w clip_frames[0].shape[:2]fourcc cv2.VideoWriter_fourcc(*mp4v)out cv2.VideoWriter(str(filename), fourcc, self.fps, (w, h))for frame in clip_frames:out.write(frame)out.release()return [str(filename)]except Exception as e:print(f保存视频失败: {e})return []# 模块7六边形图生成 class HexagonChartGenerator:def generate(self, stats: Dict, output_path: Optional[str] None) - str:try:import matplotlib.pyplot as pltimport numpy as npif output_path is None:output_path tempfile.mktemp(suffix.png)labels [击杀, 生存, 助攻, 经济, 伤害, 视野]values [stats.get(kills, 5),stats.get(survival, 5),stats.get(assists, 5),stats.get(gold, 5),stats.get(damage, 5),stats.get(vision, 5)]values [min(v, 10) for v in values]angles np.linspace(0, 2 * np.pi, len(labels), endpointFalse).tolist()values values[:1]angles angles[:1]fig, ax plt.subplots(figsize(6, 6), subplot_kwdict(polarTrue))ax.plot(angles, values, o-, linewidth2, color#00B4D8)ax.fill(angles, values, alpha0.25, color#00B4D8)ax.set_xticks(angles[:-1])ax.set_xticklabels(labels)ax.set_ylim(0, 10)ax.grid(True)plt.savefig(output_path, dpi100, bbox_inchestight)plt.close()return output_pathexcept:return self._generate_fallback(stats, output_path)def _generate_fallback(self, stats: Dict, output_path: Optional[str]) - str:output_path output_path or tempfile.mktemp(suffix.txt)with open(output_path, w, encodingutf-8) as f:f.write( 选手六边形能力图 \n)labels [击杀, 生存, 助攻, 经济, 伤害, 视野]keys [kills, survival, assists, gold, damage, vision]for label, key in zip(labels, keys):f.write(f{label}: {stats.get(key, 5)}/10\n)return output_path# 模块8消息推送 class MessagePusher:def push_wechat(self, message: str, files: Optional[List[str]] None):webhook CONFIG.get(WECHAT_WEBHOOK)if not webhook:return Falsetry:data {msgtype: text, text: {content: message}}resp requests.post(webhook, jsondata, timeout10)return resp.status_code 200except:return Falsedef push_email(self, subject: str, body: str, attachments: Optional[List[str]] None):import smtplibfrom email.mime.text import MIMETextfrom email.mime.multipart import MIMEMultipartfrom email.mime.base import MIMEBasefrom email import encoderssender CONFIG.get(QQ_EMAIL)password CONFIG.get(SMTP_PASSWORD)target CONFIG.get(TARGET_EMAIL)if not sender or not password or not target:return Falsetry:msg MIMEMultipart()msg[From] sendermsg[To] targetmsg[Subject] subjectmsg.attach(MIMEText(body, plain, utf-8))if attachments:for filepath in attachments:if os.path.exists(filepath):with open(filepath, rb) as f:part MIMEBase(application, octet-stream)part.set_payload(f.read())encoders.encode_base64(part)part.add_header(Content-Disposition,fattachment; filename{os.path.basename(filepath)})msg.attach(part)server smtplib.SMTP_SSL(smtp.qq.com, 465)server.login(sender, password)server.sendmail(sender, [target], msg.as_string())server.quit()return Trueexcept:return Falsedef push(self, title: str, content: str, attachments: Optional[List[str]] None):统一推送接口full_msg f{title}\n{content}if attachments:full_msg f\n附件: {, .join(attachments)}self.push_wechat(full_msg, attachments)self.push_email(title, content, attachments)# 主程序 class CoachAssistant:def __init__(self):self.dongle DongleBinder(CONFIG[DONGLE_VID],CONFIG[DONGLE_PID],CONFIG[DONGLE_SERIAL])self.game_detector GameDetector(CONFIG[LOL_PROCESS_NAMES])self.event_detector EventDetector()self.lcu LCUClient()self.lcu_ws LCUWebSocket(self.event_detector)self.recorder RingBufferRecorder(fpsCONFIG[SCREEN_FPS],buffer_secCONFIG[BUFFER_SECONDS],output_dirCONFIG[RECORD_DIR])self.chart_gen HexagonChartGenerator()self.pusher MessagePusher()self.running Falsedef start(self):print( * 50)print(⚡ 电竞教练辅助系统 v2.0真实LCU适配)print( * 50)# 1. 检测加密狗print( 检测加密狗...)if not self.dongle.detect():print(❌ 未检测到加密狗程序退出。)sys.exit(1)print(✅ 加密狗验证通过)# 2. 等待游戏self.game_detector.wait_for_game()# 3. 连接LCUprint( 连接LCU...)if not self.lcu.connect():print(⚠️ LCU连接失败部分功能不可用)else:print(✅ LCU连接成功)# 4. 启动录屏self.recorder.start()# 5. 连接WebSocketprint( 连接LCU WebSocket...)self.lcu_ws.connect()# 6. 注册事件回调self.lcu_ws.on(critical_strike, self._on_critical_strike)self.lcu_ws.on(penta_kill, self._on_penta_kill)print( * 50)print(✅ 系统运行中按 CtrlC 停止)print( * 50)self.running Truetry:while self.running:time.sleep(1)except KeyboardInterrupt:self.stop()def stop(self):print(\n 正在停止...)self.running Falseself.lcu_ws.disconnect()self.recorder.stop()print(✅ 系统已停止)def _on_critical_strike(self, data: Dict):暴击事件处理rate data.get(rate, 0)print(f⚡ 连环暴击! 速度: {rate:.2f}次/秒)clips self.recorder.save_clip(fcritical_{int(rate*100)},seconds_beforeCONFIG[BUFFER_SECONDS],seconds_afterCONFIG[BUFFER_SECONDS])self.pusher.push(⚡ 连环暴击警报,f选手触发连环暴击速度 {rate:.2f} 次/秒\nf击杀: {data.get(killer, 未知)} → {data.get(victim, 未知)},clips)def _on_penta_kill(self, data: Dict):五杀事件处理killer data.get(killer, 未知)print(f 五杀! 击杀者: {killer})# 保存5段视频每段前后30秒video_paths []for i in range(5):clips self.recorder.save_clip(fpenta_{killer}_{i1},seconds_beforeCONFIG[BUFFER_SECONDS],seconds_afterCONFIG[BUFFER_SECONDS])video_paths.extend(clips)time.sleep(0.3)# 生成六边形图try:stats self.lcu.get_active_player()if stats:chart_path self.chart_gen.generate({kills: stats.get(kills, 5),survival: 10 - stats.get(deaths, 5),assists: stats.get(assists, 5),gold: min(stats.get(gold, 5000) / 1000, 10),damage: min(stats.get(damage, 15000) / 2000, 10),vision: stats.get(vision, 5)})video_paths.append(chart_path)except:passself.pusher.push(f 五杀成就! ({killer}),f选手 {killer} 达成五杀\nf5段击杀视频已录制请查看附件,video_paths)# 入口 if __name__ __main__:# 检查依赖required [requests, opencv-python, mss, pillow, numpy, websocket-client]missing []for pkg in required:try:__import__(pkg.replace(-, _))except ImportError:missing.append(pkg)if missing:print(❌ 缺少依赖库:)for pkg in missing:print(f pip install {pkg})sys.exit(1)assistant CoachAssistant()assistant.start()
