youpaishidifu
438ba08365
1. 历史消息二次处理 (Task #1490): - 启动时从消息历史数据库加载最近50条bot回复到_replied_texts - 降低查库过滤器重合率阈值0.9→0.7 - 防止重启后旧回复被OCR重新识别处理 2. 回复截断 (Task #1491): - 表格内容截断长度300→1000字符 - 普通文本保持300字符 - 避免长表格回复被截断
212 lines
9.5 KiB
Python
212 lines
9.5 KiB
Python
"""自动回复 — 消息匹配知识库后准备答案，由主线程发送"""
|
||
|
||
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:
|
||
# 表格内容允许更长，避免截断重要信息
|
||
max_len = 1000 if '|' in body else 300
|
||
answer = body[:max_len-3] + "..." if len(body) > max_len else body
|
||
else:
|
||
content = doc.get("content", "").strip()
|
||
max_len = 1000 if '|' in content else 300
|
||
answer = content[:max_len-3] + "..." if len(content) > max_len 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
|