"""关键词打分去重系统 - 替代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 _ensure_keyword_scores_table(self): """确保keyword_scores表存在(防御性编程)""" conn = sqlite3.connect(self.db_path, check_same_thread=False) try: cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='keyword_scores'") if not cursor.fetchone(): logger.warning("[keyword-scorer] keyword_scores表不存在,重新创建") 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() logger.warning("[keyword-scorer] keyword_scores表已重新创建") finally: conn.close() 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, [] # 防御性编程:确保表存在 self._ensure_keyword_scores_table() 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) """ # 防御性编程:确保表存在 self._ensure_keyword_scores_table() 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: cursor = 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}") return cursor.lastrowid finally: conn.close() def get_last_score(self, platform: str, text: str) -> int: """获取指定文本最近一次的关键词得分(不记录新行)""" keywords = self.extract_keywords(text) if not keywords: return 0 # 防御性编程:确保表存在 self._ensure_keyword_scores_table() conn = sqlite3.connect(self.db_path, check_same_thread=False) try: cursor = conn.execute(""" SELECT score FROM keyword_scores WHERE platform = ? AND message_text = ? ORDER BY id DESC LIMIT 1 """, (platform, text.strip())) row = cursor.fetchone() return row[0] if row else 0 finally: conn.close() def should_skip_message(self, text: str, platform: str, ocr_time: Optional[str] = None, threshold: int = 2) -> Tuple[bool, int, List[str]]: """判断是否应该跳过该消息(基于关键词得分 + 时间感知) 流程: 1. 得分≤1 → 直接放行(新消息) 2. 得分≥1 → 比较三个时间源: - DB落库时间(历史最后一次处理时间) - OCR识别时间(来自时间戳) - 系统当前时间 3. 时间源选择: - 若OCR时间与系统时间差 ≤ ±10秒 → 用OCR时间(OCR时间更准) - 否则 → 用系统时间 4. 判断: - 若当前时间 > DB落库时间 + 5分钟 → 放行(5分钟前已处理过) - 否则 → 拦截(重复消息) Args: text: 消息文本 platform: 平台标识 ocr_time: OCR识别到的时间戳(如'22:54:16'),可选 threshold: 得分阈值 Returns: (是否跳过, 得分, 匹配关键词) """ keywords = self.extract_keywords(text) if not keywords: return False, 0, [] score, matched = self.score_against_history(keywords, platform) # 记录本次得分(先记录用于后续检查) rec_id = self.record_message_score(platform, text, keywords, score, matched, is_new=True) # 得分≤1:直接放行(新消息) if score <= 1: return False, score, matched # 得分≥1:时间比较逻辑 # ⭐ 排除刚插入的自身记录:否则同 hash 反查命中自己 → db_time=now → 永远拦截 should_skip = self._should_skip_by_time(text, platform, score, matched, ocr_time, exclude_id=rec_id) if should_skip: logger.info(f"[keyword-scorer] 跳过消息(得分{score},重复): {text[:40]}... " f"匹配关键词: {matched[:5]}") else: logger.info(f"[keyword-scorer] 放行消息(得分{score},时间超时): {text[:40]}... " f"匹配关键词: {matched[:5]}") return should_skip, score, matched def _should_skip_by_time(self, text: str, platform: str, score: int, matched: List[str], ocr_time: Optional[str], exclude_id: Optional[int] = None) -> bool: """时间比较逻辑:判断是否拦截重复消息 规则: 1. 时间源选择: - 若OCR时间与系统时间差 ≤ ±10秒 → 用OCR时间(OCR时间更准) - 否则 → 用系统时间 2. 判断: - 当前时间 > DB落库时间 + 5分钟 → 放行 - 否则 → 拦截 exclude_id: 排除刚插入的当前消息记录id(防止自拦截) """ # 防御性编程:确保表存在 self._ensure_keyword_scores_table() import hashlib from datetime import timedelta # 获取DB落库时间(同平台、同hash的最后一条记录,排除自身) message_hash = hashlib.md5(text.strip().encode('utf-8')).hexdigest() conn = sqlite3.connect(self.db_path, check_same_thread=False) try: # 查询同hash的最新记录 if exclude_id is not None: cursor = conn.execute(""" SELECT local_time, score FROM keyword_scores WHERE platform = ? AND message_hash = ? AND id != ? ORDER BY id DESC LIMIT 1 """, (platform, message_hash, exclude_id)) else: cursor = conn.execute(""" SELECT local_time, score FROM keyword_scores WHERE platform = ? AND message_hash = ? ORDER BY id DESC LIMIT 1 """, (platform, message_hash)) row = cursor.fetchone() if not row: # 无历史同hash记录 → 放行 logger.debug(f"[keyword-scorer] 无历史同hash记录,放行: hash={message_hash[:8]}") return False db_time_str = row[0] db_time = datetime.strptime(db_time_str, "%Y-%m-%d %H:%M:%S") logger.debug(f"[keyword-scorer] DB落库时间: {db_time_str}, score={row[1]}") # 确定当前时间(OCR时间 vs 系统时间) now = datetime.now() current_time = now if ocr_time: # 解析OCR时间(格式如'22:54:16') try: ocr_dt = datetime.strptime(ocr_time, "%H:%M:%S") # 同一天 ocr_dt = ocr_dt.replace(year=now.year, month=now.month, day=now.day) # 时间差检查 time_diff = abs((now - ocr_dt).total_seconds()) if time_diff <= 10: current_time = ocr_dt logger.debug(f"[keyword-scorer] 使用OCR时间(差值{time_diff:.0f}秒≤10秒)") else: logger.debug(f"[keyword-scorer] OCR时间差值{time_diff:.0f}秒>10秒,使用系统时间") except ValueError: logger.debug(f"[keyword-scorer] OCR时间解析失败'{ocr_time}',使用系统时间") # 判断:当前时间 > DB落库时间 + 5分钟 → 放行 cutoff_time = db_time + timedelta(minutes=5) if current_time > cutoff_time: logger.debug(f"[keyword-scorer] 当前时间{current_time.strftime('%H:%M:%S')} > DB+5分{cutoff_time.strftime('%H:%M:%S')},放行") return False else: logger.debug(f"[keyword-scorer] 当前时间{current_time.strftime('%H:%M:%S')} <= DB+5分{cutoff_time.strftime('%H:%M:%S')},拦截") return True finally: conn.close()