"""自动回复 — 消息匹配知识库后准备答案，由主线程发送""" import logging import threading import jieba logger = logging.getLogger("msgbot.autoreply") class AutoReply: """自动回复控制器""" def __init__(self, monitor): self.monitor = monitor self._enabled = False self._min_score = 0.30 # 已回复的消息指纹（防重复） self._replied: dict = {} self._lock = threading.Lock() def set_enabled(self, enabled: bool): self._enabled = enabled logger.info(f"自动回复 {'已开启' if enabled else '已关闭'}") def set_min_score(self, score: float): self._min_score = max(0.0, min(10.0, score)) @property def enabled(self) -> bool: return self._enabled @property def min_score(self) -> float: return self._min_score def on_kb_match(self, platform: str, msg: str, top_doc: dict, score: float) -> str or None: """知识库匹配命中时调用。线程安全。 返回要发送的回答文本，返回 None 表示不发送。 """ if not self._enabled: return None if score < self._min_score: logger.debug( f"[{platform}] 匹配度 {score:.2f} < 阈值 {self._min_score}，跳过回复" ) return None # 内容相关性检查：查询词与匹配文档至少有一个共同2字+词，防偶然匹配 if not self._has_content_relevance(msg, top_doc, score): heading = top_doc.get("heading", "")[:40] body_preview = (top_doc.get("content", "") or top_doc.get("body", ""))[:60] logger.debug( f"[{platform}] 内容相关性检查失败，跳过: " f"msg={msg[:40]!r} score={score:.2f} doc_heading={heading!r} " f"doc_body={body_preview!r}" ) return None with self._lock: return self._get_answer(platform, msg, top_doc, score) def get_reply_data(self, platform: str, msg: str, top_doc: dict, score: float) -> dict or None: """返回 {platform, answer, sender} 供主线程发送。""" answer = self.on_kb_match(platform, msg, top_doc, score) if not answer: return None # 获取发送者（在同一线程提取，不涉及窗口操作） raw = self.monitor.get_raw_clipboard(platform) sender = self.monitor._parse_last_sender(raw) # 只有发送者是有效姓名且不在消息内容中时才切会话 if sender and sender not in msg and score >= self._min_score + 0.05: pass else: sender = None return {"platform": platform, "answer": answer, "sender": sender, "msg": msg} @staticmethod def _has_content_relevance(query: str, doc: dict, score: float = 0.0) -> bool: """查询词与匹配文档至少有一个共同2字+词，防偶然匹配。 短查询(≤2个2字+词)直接放行，不做严格检查。 检查顺序：heading → title → content/body。 动作/配置类查询额外要求正文含核心词，不能只命中标题。 高匹配度(≥10.0)：如果heading+title+body都无共同词，直接拒绝。 """ import re q_tokens = set(t for t in jieba.cut(query) if len(t) >= 2) # ⭐ 修复：短查询不再无条件放行，改为必须有子串命中才放行 # 原逻辑导致"下载"(1 token)→"前置依赖"误匹配（无任何共同token却放行） _is_short_query = len(q_tokens) <= 2 # 1. 检查 heading + title（排除链接文本） heading = doc.get("heading", "") title = doc.get("title", "") # 清理 heading 中的 Markdown 链接：[text](url) → text _link_pattern = re.compile(r'\[([^\]]*)\]\([^\)]+\)') heading_clean = _link_pattern.sub(r'\1', heading) # 进一步清理：去掉链接文本本身 heading_no_link = _link_pattern.sub('', heading) doc_text = heading_clean + " " + title d_tokens = set(t for t in jieba.cut(doc_text) if len(t) >= 2) has_heading_hit = bool(q_tokens & d_tokens) # 2. 检查 content/body（排除链接文本） body = doc.get("content", "") or doc.get("body", "") b_tokens = set() has_body_hit = False if body: # 清理 body 中的链接文本，防止 "[sdf1_login登录插件](sdf1_login.md)" 误匹配 body_clean = _link_pattern.sub(r'\1', body) body_no_link = _link_pattern.sub('', body) # 截断前500字防超长文档 b_tokens = set(t for t in jieba.cut(body_no_link[:500]) if len(t) >= 2) has_body_hit = bool(q_tokens & b_tokens) # 动作/配置类查询：不能只命中标题，正文必须也含核心词 # 否则"# Sdf1系列登录插件使用说明"会因标题含"登录/插件/使用"被误放 action_keywords = {"怎么", "如何", "使用", "配置", "修改", "设置", "安装", "用法"} if action_keywords & q_tokens: core_q_tokens = q_tokens - {"插件", "使用", "怎么", "如何", "吗", "呢", "吧"} if core_q_tokens and not (core_q_tokens & b_tokens): logger.debug( f"动作查询正文相关性不足: query={query[:40]!r} " f"heading_hit={has_heading_hit} body_hit={has_body_hit}" ) return False if has_heading_hit or has_body_hit: return True # 3. 子串回退：jieba分词可能产生"包是""还是"等碎片token， # 导致token交集为空但文档确实包含查询核心词（如"资源包"）。 # 检查查询中任意2~6字子串是否出现在文档文本中。 # ⭐ 短查询(≤2 token)：检查整个query是否出现在文档中（"下载"必须在文档中出现） # ⭐ 使用清理后的文本（排除链接文本） full_text = doc_text + " " + body_no_link[:500] if body else doc_text stripped_query = re.sub(r'[？?！!。，,\s]', '', query) # 短查询特殊处理：整个query必须作为子串出现在文档中 if _is_short_query and len(stripped_query) >= 2: if stripped_query in full_text: logger.debug(f"短查询子串命中: query={stripped_query!r} in doc={doc.get('heading','')[:30]!r}") return True else: logger.debug(f"短查询子串未命中: query={stripped_query!r} 不在文档中，拒绝") return False if len(stripped_query) >= 4: for qlen in range(min(len(stripped_query), 6), 1, -1): for start in range(len(stripped_query) - qlen + 1): phrase = stripped_query[start:start + qlen] if len(phrase) >= 2 and phrase in full_text: logger.debug( f"子串回退命中: phrase={phrase!r} in doc={doc.get('heading','')[:30]!r}" ) return True # 4. 高匹配度但无共同词：可能是TF-IDF误匹配，拒绝 # 用户问"PVP击杀播报"，KB返回"json格式说明"（匹配度9.53）→ 应拒绝 if score >= 10.0 and not has_heading_hit and not has_body_hit: return False return False def _get_answer(self, platform: str, msg: str, doc: dict, score: float) -> str or None: """提取答案 + 防重复检查。 优先返回 doc['answer']（Q&A 段），兜底使用 doc['body']（仅正文，不含 heading）。 KB 搜索已按 section 切分，doc['body'] 即最相关的片段正文。 只需截断到 300 字符避免消息过长。 """ answer = doc.get("answer") if answer: answer = answer.strip() else: # 优先用 body（纯正文），兜底用 content（heading+正文） body = doc.get("body", "").strip() if body and len(body) > 20: answer = body[:297] + "..." if len(body) > 300 else body else: content = doc.get("content", "").strip() answer = content[:297] + "..." if len(content) > 300 else content if not answer: return None # 拒绝把纯标题/免责声明当作答案发送 import re if answer.startswith("# ") and len(answer) < 60: logger.debug(f"[{platform}] 答案为纯标题，跳过: {answer[:60]!r}") return None if re.search(r'特别说明|免责声明|本插件.*自行修改|仅供参考|版权所有', answer): logger.debug(f"[{platform}] 答案为免责声明/通用说明，跳过: {answer[:60]!r}") return None # 防重复指纹 fp = hash(f"{platform}:{msg}:{answer[:80]}") replied_set = self._replied.setdefault(platform, set()) if fp in replied_set: logger.debug(f"[{platform}] 重复命中，跳过: {msg[:40]}") return None replied_set.add(fp) # 控制缓存上限 if len(replied_set) > 200: self._replied[platform] = set(list(replied_set)[-100:]) logger.info( f"[{platform}] 🎯 自动回复触发 | 匹配度 {score:.2f} | " f"消息: {msg[:40]}" ) return answer