feat: 实现关键词打分去重系统 (Task #1556)

2026-08-18 12:45:04 +08:00
parent da619dd3da
commit c7afa13bc5
2 changed files with 236 additions and 0 deletions
@@ -0,0 +1,216 @@
+"""关键词打分去重系统 - 替代OCR时间戳判断消息新旧
+
+核心思路：
+1. 每次处理消息时，提取关键词（两字一组）
+2. 与数据库中的历史回复匹配，每个匹配+1分
+3. 得分≥2：视为已知回复，跳过
+4. 得分≤1：视为新消息，继续处理
+5. 记录得分和关键词到数据库，用于后续分析
+6. 完全依赖本地时间，不依赖OCR识别的时间戳
+"""
+
+import re
+import sqlite3
+import logging
+from datetime import datetime
+from typing import List, Dict, Tuple, Optional
+
+logger = logging.getLogger("msgbot.keyword_scorer")
+
+
+class KeywordScorer:
+    """关键词打分器：通过关键词匹配判断消息是否为已知回复"""
+
+    def __init__(self, db_path: str):
+        self.db_path = db_path
+        self._init_db()
+
+    def _init_db(self):
+        """初始化打分数据库表"""
+        conn = sqlite3.connect(self.db_path, check_same_thread=False)
+        try:
+            # 关键词得分记录表
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS keyword_scores (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    platform TEXT NOT NULL,
+                    message_text TEXT NOT NULL,
+                    message_hash TEXT NOT NULL,
+                    score INTEGER NOT NULL DEFAULT 0,
+                    matched_keywords TEXT NOT NULL DEFAULT '',
+                    local_time TIMESTAMP NOT NULL,
+                    is_new_message INTEGER NOT NULL DEFAULT 1
+                )
+            """)
+            # 为快速查询创建索引
+            conn.execute("""
+                CREATE INDEX IF NOT EXISTS idx_keyword_scores_platform_time
+                ON keyword_scores(platform, local_time)
+            """)
+            conn.execute("""
+                CREATE INDEX IF NOT EXISTS idx_keyword_scores_hash
+                ON keyword_scores(message_hash)
+            """)
+            conn.commit()
+        finally:
+            conn.close()
+        logger.debug(f"[keyword-scorer] DB初始化完成: {self.db_path}")
+
+    def extract_keywords(self, text: str) -> List[str]:
+        """提取两字一组的关键词
+
+        规则：
+        1. 中文：两字一组滑动窗口
+        2. 英文：忽略大小写，提取有意义片段
+        3. 数字：4位以上连续数字作为关键词
+        """
+        if not text:
+            return []
+
+        keywords = set()
+
+        # 提取中文关键词（两字一组）
+        chinese_chars = re.findall(r'[\u4e00-\u9fff]', text)
+        for i in range(len(chinese_chars) - 1):
+            keywords.add(chinese_chars[i] + chinese_chars[i + 1])
+
+        # 提取英文关键词（忽略大小写）
+        english_words = re.findall(r'[a-zA-Z]{3,}', text)
+        for word in english_words:
+            keywords.add(word.lower())
+            # 也添加首字母缩写
+            if len(word) >= 4:
+                keywords.add(word[:2].lower())
+
+        # 提取数字关键词
+        numbers = re.findall(r'\d{4,}', text)
+        for num in numbers:
+            keywords.add(num)
+
+        # 过滤太短或太常见的词
+        stop_words = {'的', '了', '是', '在', '我', '有', '和', '就', '不', '人',
+                      '都', '一', '一个', '这', '那', '你', '他', '她', '它', '们',
+                      'php', 'git', 'com', 'cn'}
+        keywords = {kw for kw in keywords if kw not in stop_words and len(kw) >= 2}
+
+        return list(keywords)
+
+    def score_against_history(self, keywords: List[str], platform: str,
+                              lookback_minutes: int = 10) -> Tuple[int, List[str]]:
+        """计算关键词与历史回复的匹配得分
+
+        Args:
+            keywords: 当前消息的关键词列表
+            platform: 平台标识 (qq/wechat)
+            lookback_minutes: 回溯时间范围（分钟）
+
+        Returns:
+            (得分, 匹配的关键词列表)
+        """
+        if not keywords:
+            return 0, []
+
+        conn = sqlite3.connect(self.db_path, check_same_thread=False)
+        try:
+            # 获取最近N分钟的bot回复关键词
+            from datetime import timedelta
+            cutoff_time = (datetime.now() - timedelta(minutes=lookback_minutes)).strftime("%Y-%m-%d %H:%M:%S")
+
+            # 查询历史回复的关键词（简化：直接查message_text）
+            cursor = conn.execute("""
+                SELECT message_text, matched_keywords
+                FROM keyword_scores
+                WHERE platform = ?
+                  AND local_time > ?
+                  AND is_new_message = 0
+                ORDER BY id DESC
+                LIMIT 100
+            """, (platform, cutoff_time))
+
+            history = cursor.fetchall()
+            if not history:
+                return 0, []
+
+            matched = []
+            score = 0
+
+            # 对每个关键词，检查是否在历史回复中出现
+            for kw in keywords:
+                for text, hist_kws in history:
+                    # 精确匹配
+                    if kw in hist_kws:
+                        score += 1
+                        if kw not in matched:
+                            matched.append(kw)
+                        break
+                    # 子串匹配（抵抗OCR噪声）
+                    if kw.lower() in text.lower():
+                        score += 1
+                        if kw not in matched:
+                            matched.append(kw)
+                        break
+
+            return score, matched
+
+        finally:
+            conn.close()
+
+    def record_message_score(self, platform: str, message_text: str,
+                             keywords: List[str], score: int,
+                             matched_keywords: List[str], is_new: bool = True):
+        """记录消息的关键词得分到数据库
+
+        Args:
+            platform: 平台标识
+            message_text: 原始消息文本
+            keywords: 提取的所有关键词
+            score: 匹配得分
+            matched_keywords: 匹配到的关键词
+            is_new: 是否为新消息（score<=1为True）
+        """
+        import hashlib
+        message_hash = hashlib.md5(message_text.strip().encode('utf-8')).hexdigest()
+        local_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+
+        conn = sqlite3.connect(self.db_path, check_same_thread=False)
+        try:
+            conn.execute("""
+                INSERT INTO keyword_scores
+                (platform, message_text, message_hash, score, matched_keywords, local_time, is_new_message)
+                VALUES (?, ?, ?, ?, ?, ?, ?)
+            """, (platform, message_text.strip(), message_hash, score,
+                  ','.join(matched_keywords), local_time, 1 if is_new else 0))
+            conn.commit()
+            logger.debug(f"[keyword-scorer] 记录得分: platform={platform}, score={score}, "
+                         f"keywords={matched_keywords[:5]}, is_new={is_new}")
+        finally:
+            conn.close()
+
+    def should_skip_message(self, text: str, platform: str,
+                            threshold: int = 2) -> Tuple[bool, int, List[str]]:
+        """判断是否应该跳过该消息（基于关键词得分）
+
+        Args:
+            text: 消息文本
+            platform: 平台标识
+            threshold: 得分阈值，≥threshold 则跳过
+
+        Returns:
+            (是否跳过, 得分, 匹配关键词)
+        """
+        keywords = self.extract_keywords(text)
+        if not keywords:
+            return False, 0, []
+
+        score, matched = self.score_against_history(keywords, platform)
+
+        # 记录本次得分
+        is_new = score < threshold
+        self.record_message_score(platform, text, keywords, score, matched, is_new)
+
+        should_skip = score >= threshold
+        if should_skip:
+            logger.info(f"[keyword-scorer] 跳过消息(得分{score}≥{threshold}): {text[:40]}... "
+                        f"匹配关键词: {matched}")
+
+        return should_skip, score, matched