"""GUI — customtkinter 界面，开关 + 知识库路径 + 日志 """ import tkinter as tk from tkinter import filedialog, messagebox from typing import Optional import logging import queue import threading import os import customtkinter as ctk from config import load_config, save_config from monitor import MessageMonitor from knowledge_base import KnowledgeBase from auto_reply import AutoReply logger = logging.getLogger("msgbot.gui") ctk.set_appearance_mode("System") ctk.set_default_color_theme("green") class MsgBotApp(ctk.CTk): """主窗口""" def __init__(self): super().__init__() self.title("消息问答助手 v1.0") self.geometry("720x560") self.minsize(640, 480) # 配置 self.cfg = load_config() # 知识库 self.kb = KnowledgeBase(self.cfg.get("knowledge_base_path", "")) # 监控器 self.monitor = MessageMonitor(on_message=self._on_new_message) self.monitor.set_poll_interval(self.cfg.get("poll_interval_seconds", 2.0)) self.monitor.set_enabled_platforms( wechat=self.cfg.get("wechat_enabled", True), qq=self.cfg.get("qq_enabled", False), ) self.monitor.set_emergency_callback(self._on_emergency_stop) # 自动回复 self.ar = AutoReply(self.monitor) self.ar.set_enabled(self.cfg.get("auto_reply_enabled", False)) self.ar.set_min_score(self.cfg.get("auto_reply_min_score", 0.35)) # 日志队列（线程安全） self._log_queue: queue.Queue[str] = queue.Queue() # 自动回复执行队列（转发到主线程，避免 daemon 线程 SetForegroundWindow 失败） self._reply_queue: queue.Queue[dict] = queue.Queue() self._build_ui() # 定时从队列取日志刷新 self.after(200, self._flush_log) # 定时处理自动回复队列（主线程才能可靠 SetForegroundWindow） self.after(200, self._flush_reply_queue) # 窗口关闭 self.protocol("WM_DELETE_WINDOW", self._on_close) # ── UI 构建 ────────────────────────────── def _build_ui(self): self.grid_columnconfigure(0, weight=1) self.grid_rowconfigure(2, weight=1) # ── 顶部：开关 + 状态 ── header = ctk.CTkFrame(self) header.grid(row=0, column=0, padx=12, pady=(12, 4), sticky="ew") header.grid_columnconfigure(1, weight=1) ctk.CTkLabel(header, text="🤖 消息问答助手", font=("Microsoft YaHei", 18, "bold")).grid( row=0, column=0, padx=(8, 16), pady=8, sticky="w" ) self._status_label = ctk.CTkLabel(header, text="⏸ 已暂停", font=("Microsoft YaHei", 13)) self._status_label.grid(row=0, column=1, padx=4, pady=8, sticky="w") self._toggle_btn = ctk.CTkSwitch( header, text="", command=self._toggle_monitor, onvalue=True, offvalue=False, ) self._toggle_btn.grid(row=0, column=2, padx=(4, 8), pady=8, sticky="e") ctk.CTkLabel(header, text="启用监控", font=("Microsoft YaHei", 12)).grid( row=0, column=3, padx=(0, 12), pady=8, sticky="w" ) # ── 设置区域 ── settings = ctk.CTkFrame(self) settings.grid(row=1, column=0, padx=12, pady=4, sticky="ew") settings.grid_columnconfigure(1, weight=1) row = 0 # 知识库路径 ctk.CTkLabel(settings, text="📂 知识库路径:", font=("Microsoft YaHei", 12)).grid( row=row, column=0, padx=(8, 4), pady=6, sticky="w" ) self._kb_path_var = ctk.StringVar(value=self.cfg.get("knowledge_base_path", "")) self._kb_entry = ctk.CTkEntry( settings, textvariable=self._kb_path_var, placeholder_text="选择知识库文件夹（.md / .txt）" ) self._kb_entry.grid(row=row, column=1, padx=4, pady=6, sticky="ew") self._kb_browse_btn = ctk.CTkButton( settings, text="浏览", width=64, command=self._browse_kb ) self._kb_browse_btn.grid(row=row, column=2, padx=(4, 8), pady=6, sticky="e") self._kb_status = ctk.CTkLabel(settings, text="未加载", font=("Microsoft YaHei", 11)) self._kb_status.grid(row=row, column=3, padx=4, pady=6, sticky="w") row += 1 # 平台选择 ctk.CTkLabel(settings, text="📱 监控平台:", font=("Microsoft YaHei", 12)).grid( row=row, column=0, padx=(8, 4), pady=6, sticky="w" ) self._wx_var = ctk.BooleanVar(value=self.cfg.get("wechat_enabled", True)) self._wx_cb = ctk.CTkCheckBox(settings, text="微信", variable=self._wx_var, command=self._save_plat_settings) self._wx_cb.grid(row=row, column=1, padx=4, pady=6, sticky="w") self._qq_var = ctk.BooleanVar(value=self.cfg.get("qq_enabled", False)) self._qq_cb = ctk.CTkCheckBox(settings, text="QQ", variable=self._qq_var, command=self._save_plat_settings) self._qq_cb.grid(row=row, column=1, padx=(90, 4), pady=6, sticky="w") row += 1 # 轮询间隔 ctk.CTkLabel(settings, text="⏱ 轮询间隔(秒):", font=("Microsoft YaHei", 12)).grid( row=row, column=0, padx=(8, 4), pady=6, sticky="w" ) self._interval_var = ctk.DoubleVar(value=self.cfg.get("poll_interval_seconds", 2.0)) self._interval_slider = ctk.CTkSlider( settings, from_=0.5, to=10.0, number_of_steps=19, variable=self._interval_var, command=self._save_interval ) self._interval_slider.grid(row=row, column=1, padx=4, pady=6, sticky="ew") self._interval_label = ctk.CTkLabel( settings, text=f"{self._interval_var.get():.1f}s", width=40 ) self._interval_label.grid(row=row, column=2, padx=(4, 8), pady=6, sticky="w") # 应用按钮 self._apply_btn = ctk.CTkButton( settings, text="应用设置", width=80, command=self._apply_settings ) self._apply_btn.grid(row=row, column=3, padx=4, pady=6, sticky="e") row += 1 # ── 自动回复 ── ctk.CTkLabel(settings, text="🤖 自动回复:", font=("Microsoft YaHei", 12)).grid( row=row, column=0, padx=(8, 4), pady=6, sticky="w" ) ar_frame = ctk.CTkFrame(settings, fg_color="transparent") ar_frame.grid(row=row, column=1, columnspan=2, padx=4, pady=6, sticky="ew") ar_frame.grid_columnconfigure(2, weight=1) self._ar_var = ctk.BooleanVar(value=self.cfg.get("auto_reply_enabled", False)) self._ar_toggle = ctk.CTkSwitch( ar_frame, text="", variable=self._ar_var, command=self._save_ar_settings, onvalue=True, offvalue=False, ) self._ar_toggle.grid(row=0, column=0, padx=(0, 6), pady=0, sticky="w") ctk.CTkLabel(ar_frame, text="阈值:", font=("Microsoft YaHei", 12)).grid( row=0, column=1, padx=(0, 4), pady=0, sticky="w" ) self._ar_threshold_var = ctk.DoubleVar( value=self.cfg.get("auto_reply_min_score", 0.35) ) self._ar_threshold_slider = ctk.CTkSlider( ar_frame, from_=0.10, to=0.90, number_of_steps=16, variable=self._ar_threshold_var, command=self._save_ar_threshold, ) self._ar_threshold_slider.grid(row=0, column=2, padx=2, pady=0, sticky="ew") self._ar_threshold_label = ctk.CTkLabel( ar_frame, text=f"{self._ar_threshold_var.get():.2f}", width=36 ) self._ar_threshold_label.grid(row=0, column=3, padx=(6, 0), pady=0, sticky="w") # ── 日志区域 ── log_frame = ctk.CTkFrame(self) log_frame.grid(row=2, column=0, padx=12, pady=(4, 12), sticky="nsew") log_frame.grid_rowconfigure(0, weight=1) log_frame.grid_columnconfigure(0, weight=1) self._log_text = ctk.CTkTextbox(log_frame, font=("Consolas", 11), wrap="word") self._log_text.grid(row=0, column=0, padx=4, pady=4, sticky="nsew") self._log_text.configure(state="disabled") btn_frame = ctk.CTkFrame(log_frame, fg_color="transparent") btn_frame.grid(row=1, column=0, pady=4, sticky="ew") ctk.CTkButton(btn_frame, text="清空日志", width=80, command=self._clear_log).pack(side="right", padx=4) # 初始更新知识库状态 self._update_kb_status() # ── 回调 ────────────────────────────── def _toggle_monitor(self): enabled = self._toggle_btn.get() if enabled: # 检查知识库是否已加载 if not self.kb.is_ready: self._log("⚠ 知识库未加载，请先设置知识库路径") self._toggle_btn.deselect() return # 用户手动点击启用 → 彻底重启（重新检测内存，尝试恢复GPU） self.monitor.start(full_restart=True) self._status_label.configure(text="▶ 运行中", text_color="#2ecc71") self._log("✅ 消息监控已启动") else: self.monitor.stop() self._status_label.configure(text="⏸ 已暂停") self._log("⏸ 消息监控已暂停") def _browse_kb(self): path = filedialog.askdirectory(title="选择知识库文件夹") if path: self._kb_path_var.set(path) self._apply_settings() def _apply_settings(self): # 保存知识库路径 path = self._kb_path_var.get().strip() self.cfg["knowledge_base_path"] = path self.kb.set_path(path) self._update_kb_status() # 平台 self.cfg["wechat_enabled"] = self._wx_var.get() self.cfg["qq_enabled"] = self._qq_var.get() # 自动回复 self.cfg["auto_reply_enabled"] = self._ar_var.get() self.cfg["auto_reply_min_score"] = round(self._ar_threshold_var.get(), 2) self.ar.set_enabled(self.cfg["auto_reply_enabled"]) self.ar.set_min_score(self.cfg["auto_reply_min_score"]) # 间隔 self.cfg["poll_interval_seconds"] = round(self._interval_var.get(), 1) save_config(self.cfg) # 更新监控器 self.monitor.set_poll_interval(self.cfg["poll_interval_seconds"]) self.monitor.set_enabled_platforms( wechat=self.cfg["wechat_enabled"], qq=self.cfg["qq_enabled"], ) self._log(f"📝 设置已保存 | 知识库: {self.kb.get_doc_count()} 个文档 | " f"轮询间隔: {self.cfg['poll_interval_seconds']}s") def _save_plat_settings(self): # 只保存平台选择，不触发完整应用 self.cfg["wechat_enabled"] = self._wx_var.get() self.cfg["qq_enabled"] = self._qq_var.get() save_config(self.cfg) self.monitor.set_enabled_platforms( wechat=self.cfg["wechat_enabled"], qq=self.cfg["qq_enabled"], ) def _save_interval(self, _=None): self._interval_label.configure(text=f"{self._interval_var.get():.1f}s") def _save_ar_settings(self): """自动回复开关即时保存""" self.cfg["auto_reply_enabled"] = self._ar_var.get() save_config(self.cfg) self.ar.set_enabled(self.cfg["auto_reply_enabled"]) status = "开" if self._ar_var.get() else "关" self._log(f"🤖 自动回复已切换为{status}") def _save_ar_threshold(self, _=None): val = round(self._ar_threshold_var.get(), 2) self._ar_threshold_label.configure(text=f"{val:.2f}") def _update_kb_status(self): if self.kb.is_ready: self._kb_status.configure( text=f"✅ {self.kb.get_doc_count()} 个文档", text_color="#2ecc71" ) else: self._kb_status.configure(text="❌ 未加载", text_color="#e74c3c") # ── 消息处理 ── def _on_new_message(self, platform: str, msg: str): """收到新消息 -> 搜索知识库 -> 日志输出""" if not self.kb.is_ready: self._log_queue.put(f"[{platform.upper()}] {msg}") self._log_queue.put(" ⚠ 知识库未加载，无法搜索答案") return # 在后台线程搜索 threading.Thread(target=self._search_and_log, args=(platform, msg), daemon=True).start() def _search_and_log(self, platform: str, msg: str): """搜索知识库并输出结果，匹配成功时队列自动回复到主线程""" results = self.kb.search(msg, top_k=3) lines = [f"[{platform.upper()}] {msg}"] if results: answer_title = results[0][0]["title"] answer_score = results[0][1] lines.append(f" 🔍 找到 {len(results)} 条相关结果:") for doc, score in results: title = doc["title"] snippet = doc["content"][:200].replace("\n", " ") if len(snippet) > 200: snippet = snippet[:197] + "..." lines.append(f" 📄 [{title}] (匹配度 {score:.2f})") lines.append(f" {snippet}") # 诊断日志：记录自动回复触发条件 if not self.ar.enabled: lines.append(" ⚠ 自动回复已关闭，不触发回复") elif answer_score < self.ar.min_score: lines.append(f" ⚠ 匹配度 {answer_score:.2f} < 阈值 {self.ar.min_score}，跳过") else: # 检查内容相关性 has_relevance = self.ar._has_content_relevance(msg, results[0][0], answer_score) if not has_relevance: heading = results[0][0].get("heading", "")[:40] lines.append(f" ⚠ 内容相关性检查失败 (heading={heading!r})，跳过") else: lines.append(" ✅ 匹配通过，准备发送自动回复") # 自动回复：队列到主线程执行（daemon 线程不能可靠 SetForegroundWindow） reply_data = self.ar.get_reply_data(platform, msg, results[0][0], results[0][1]) if reply_data: self._reply_queue.put(reply_data) else: lines.append(" ⚠ get_reply_data 返回 None（重复命中或内部检查失败）") else: lines.append(" ❌ 知识库中未找到相关答案") for line in lines: self._log_queue.put(line) def _log(self, text: str): """直接写日志（主线程）""" self._log_queue.put(text) def _flush_log(self): """从队列消费日志到文本框""" try: while True: line = self._log_queue.get_nowait() self._log_text.configure(state="normal") self._log_text.insert("end", line + "\n") self._log_text.see("end") self._log_text.configure(state="disabled") except queue.Empty: pass self.after(200, self._flush_log) def _flush_reply_queue(self): """从自动回复队列消费——主线程执行，SetForegroundWindow 可靠。""" try: while True: data = self._reply_queue.get_nowait() try: self._execute_reply(data) except Exception as e: logger.error(f"自动回复执行异常: {e}", exc_info=True) self.monitor._sending_event.clear() except queue.Empty: pass self.after(200, self._flush_reply_queue) def _execute_reply(self, data: dict): """主线程执行自动回复发送。加上发送锁防止监控轮询干扰。""" platform = data["platform"] answer = data["answer"] sender = data.get("sender") # 设发送锁，阻止监控轮询切走焦点 self.monitor._sending_event.set() try: if sender: # 搜索联系人切换会话 found = self.monitor._search_contact(platform, sender) if not found: self._log(f"[{platform}] ⚠ 找不到联系人 {sender}，当前会话回复") # 发送（主线程 SetForegroundWindow 可靠，且不受监控干扰） ok = self.monitor._paste_and_send(platform, answer) if ok: self._log(f"[{platform}] 🤖 已自动回复: {answer[:60]}...") else: self._log(f"[{platform}] ⚠ 自动回复发送失败（目标窗口无法带到前台）") finally: self.monitor._sending_event.clear() def _clear_log(self): self._log_text.configure(state="normal") self._log_text.delete("1.0", "end") self._log_text.configure(state="disabled") # ── 急停回调 ── def _on_emergency_stop(self): """双击 ESC 急停的回调（从 daemon 线程调用，日志队列安全）。""" self._log_queue.put("⚠️⚠️⚠️ 双击ESC急停！消息监控已强制终止") # 安全更新 toggle 状态（主线程 after 调度） self.after(0, self._update_stop_ui) def _update_stop_ui(self): """主线程更新急停后的 UI。""" self.monitor.stop() self._toggle_btn.deselect() self._status_label.configure(text="⏸ 急停") self._log_queue.put("🔄 可点击「启用监控」重新启动") # ── 窗口关闭 ── def _on_close(self): self.monitor.stop() self.destroy()