youpaishidifu
fd147862c7
根因： should_skip_message 先 record_message_score 落库(时间=now)， 再 _should_skip_by_time 按 message_hash 反查最后一条 → 命中的是 刚插入的自己 → db_time=now → '当前<=DB+5分' 永远拦截。 新消息首次出现即被拦，5分钟窗口永远走不完。 例：'那服务器有哪些基本信息，L ~' score=2(命中共用词'服务'/'务器') 时间比较命中自己 id=54(14:40:15) → 拦截，消息丢失。 修复： record_message_score 返回 lastrowid， _should_skip_by_time 增加 exclude_id 参数，查询排除自身。 验证： - 新消息首次出现 → 放行(不再自拦截) - 同文本5分内重现 → 拦截(去重仍生效) - 同文本5分后重现 → 放行(时间超窗) - 真实库下游测3/3 全部符合预期
327 lines
13 KiB
Python
327 lines
13 KiB
Python
"""关键词打分去重系统 - 替代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:
|
||
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 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（防止自拦截）
|
||
"""
|
||
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()
|