121 lines
4.4 KiB
Python
121 lines
4.4 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):
|
||
heading = top_doc.get("heading", "")[:40]
|
||
logger.debug(
|
||
f"[{platform}] 内容相关性检查失败，跳过: msg={msg[:40]!r} doc_heading={heading!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}
|
||
|
||
@staticmethod
|
||
def _has_content_relevance(query: str, doc: dict) -> bool:
|
||
"""查询词与匹配文档标题至少有一个共同2字+词，防偶然匹配。
|
||
短查询(≤2个2字+词)直接放行，不做严格检查。
|
||
"""
|
||
q_tokens = set(t for t in jieba.cut(query) if len(t) >= 2)
|
||
if len(q_tokens) <= 2:
|
||
return True # 短查询不拦截
|
||
doc_text = doc.get("heading", "") + " " + doc.get("title", "")
|
||
d_tokens = set(t for t in jieba.cut(doc_text) if len(t) >= 2)
|
||
if not (q_tokens & d_tokens):
|
||
return False
|
||
return True
|
||
|
||
def _get_answer(self, platform: str, msg: str, doc: dict, score: float) -> str or None:
|
||
"""提取答案 + 防重复检查。
|
||
优先返回 doc['answer']（Q&A 段），兜底使用 doc['content']。
|
||
KB 搜索已按 section 切分，doc['content'] 即最相关的片段。
|
||
只需截断到 300 字符避免消息过长。
|
||
"""
|
||
answer = doc.get("answer")
|
||
if answer:
|
||
answer = answer.strip()
|
||
else:
|
||
content = doc.get("content", "").strip()
|
||
answer = content[:297] + "..." if len(content) > 300 else content
|
||
|
||
if not answer:
|
||
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
|