refactor ai_auto_response into xiaoniu group bot

plugins/ai_auto_response/triggers.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from typing import Dict, List
+
+
+QUESTION_PATTERNS = [
+    r"\?$", r"？$", r"怎么", r"如何", r"咋弄", r"为啥", r"为什么",
+    r"有人知道", r"谁知道", r"能不能", r"可以吗", r"报错", r"怎么解决",
+]
+SOCIAL_PATTERNS = [r"小牛", r"在吗", r"出来", r"帮忙看", r"看看"]
+
+
+@dataclass
+class TriggerResult:
+    trigger_type: str = "none"
+    priority: float = 0.0
+    is_question: bool = False
+    is_followup: bool = False
+    is_social_call: bool = False
+    is_returning_member: bool = False
+    should_respond: bool = False
+    topic: str = ""
+    reasons: List[str] = field(default_factory=list)
+
+
+class TriggerRouter:
+    def __init__(self, config: Dict):
+        self.config = config or {}
+        self.topic_keywords = [str(item).lower() for item in self.config.get("focus", [])]
+
+    def route(self, message: Dict, memory_hints: Dict) -> TriggerResult:
+        content = str(message.get("content", "")).strip()
+        content_lower = content.lower()
+        result = TriggerResult()
+        if message.get("is_at"):
+            result.trigger_type = "at_trigger"
+            result.priority = float(self.config.get("at_bot", 1.0))
+            result.should_respond = True
+            result.reasons.append("is_at")
+        if self._is_question(content):
+            if result.priority < float(self.config.get("explicit_question", 0.95)):
+                result.trigger_type = "question_trigger"
+                result.priority = float(self.config.get("explicit_question", 0.95))
+            result.is_question = True
+            result.should_respond = True
+            result.reasons.append("question")
+        if memory_hints.get("is_followup"):
+            if result.priority < float(self.config.get("followup", 0.90)):
+                result.trigger_type = "followup_trigger"
+                result.priority = float(self.config.get("followup", 0.90))
+            result.is_followup = True
+            result.should_respond = True
+            result.reasons.append("followup")
+        topic = self._detect_topic(content_lower)
+        if topic:
+            result.topic = topic
+            result.reasons.append("topic")
+            if result.priority < float(self.config.get("casual_topic", 0.35)):
+                result.trigger_type = result.trigger_type if result.trigger_type != "none" else "topic_trigger"
+                result.priority = max(result.priority, float(self.config.get("casual_topic", 0.35)))
+        if self._is_social_call(content_lower):
+            if result.priority < float(self.config.get("social_call", 0.65)):
+                result.trigger_type = result.trigger_type if result.trigger_type != "none" else "social_trigger"
+                result.priority = max(result.priority, float(self.config.get("social_call", 0.65)))
+            result.is_social_call = True
+            result.reasons.append("social_call")
+        if memory_hints.get("returning_member_state") in {"returning_member", "long_absent_member"}:
+            result.is_returning_member = True
+            result.reasons.append(memory_hints.get("returning_member_state"))
+            if result.trigger_type == "none":
+                result.trigger_type = "returning_member"
+                result.priority = max(result.priority, float(self.config.get("casual_topic", 0.35)))
+        return result
+
+    def _is_question(self, content: str) -> bool:
+        return any(re.search(pattern, content, flags=re.IGNORECASE) for pattern in QUESTION_PATTERNS)
+
+    def _is_social_call(self, content: str) -> bool:
+        return any(re.search(pattern, content, flags=re.IGNORECASE) for pattern in SOCIAL_PATTERNS)
+
+    def _detect_topic(self, content_lower: str) -> str:
+        for keyword in self.topic_keywords:
+            if keyword and keyword in content_lower:
+                return keyword
+        return ""