youpaishidifu
c3f12cd6dc
核心变化:
1. 完全移除OCR时间戳丢弃逻辑（_is_timestamp_old不再用于丢弃消息）
2. 改为关键词打分+时间感知去重:
- 得分≤1 → 直接放行（新消息）
- 得分≥1 → 检查最近一次打分时间:
* 最近300秒内有同类记录 → 丢弃（同一用户重复）
* 超过300秒或无记录 → 放行（不同用户的新提问）
3. 修复问题:
- 旧消息被陈旧时间戳继承导致误杀（如13:32:36继承给14:42:57的消息）
- 不同用户相同问题的误过滤（如"php管理员密码"被统一丢弃）
4. OCR时间戳仅作为参考，±10秒容差，超时用本地时间
修改文件:
- _keyword_scorer.py: 新增时间感知去重逻辑
- monitor.py: 移除时间戳丢弃分支
265 lines
10 KiB
Python
265 lines
10 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:
|
||
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]]:
|
||
"""判断是否应该跳过该消息（基于关键词得分 + 时间感知）
|
||
|
||
⭐ Task #1558: 新逻辑
|
||
- 得分≤1 → 直接放行（新消息）
|
||
- 得分≥2 → 检查最近一次打分时间：
|
||
* 最近打分在5分钟内 → 丢弃（同一用户重复）
|
||
* 最近打分超过5分钟 → 放行（不同用户的新提问）
|
||
这样避免一刀切≥2分就丢弃，因为不同用户可能提出相同问题。
|
||
|
||
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)
|
||
|
||
# 记录本次得分（先记录用于后续检查）
|
||
self.record_message_score(platform, text, keywords, score, matched, is_new=True)
|
||
|
||
# 得分≤1：直接放行（新消息）
|
||
if score <= 1:
|
||
return False, score, matched
|
||
|
||
# 得分≥2：检查最近一次同类消息的打分时间
|
||
should_skip = self._should_skip_by_recent_score(platform, score, matched)
|
||
|
||
if should_skip:
|
||
logger.info(f"[keyword-scorer] 跳过消息(得分{score}≥{threshold},5分钟内): {text[:40]}... "
|
||
f"匹配关键词: {matched[:5]}")
|
||
else:
|
||
logger.info(f"[keyword-scorer] 放行消息(得分{score}≥{threshold},超过5分钟或无记录): "
|
||
f"{text[:40]}... 匹配关键词: {matched[:5]}")
|
||
|
||
return should_skip, score, matched
|
||
|
||
def _should_skip_by_recent_score(self, platform: str, score: int, matched: List[str]) -> bool:
|
||
"""检查最近一次同类消息的打分时间，超过5分钟则放行"""
|
||
conn = sqlite3.connect(self.db_path, check_same_thread=False)
|
||
try:
|
||
from datetime import timedelta
|
||
cutoff_time = (datetime.now() - timedelta(seconds=300)).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
# 查询最近300秒内得分≥1的消息（不限score阈值，只看时间）
|
||
cursor = conn.execute("""
|
||
SELECT local_time, score
|
||
FROM keyword_scores
|
||
WHERE platform = ?
|
||
AND score >= 1
|
||
AND matched_keywords != ''
|
||
AND local_time > ?
|
||
ORDER BY id DESC
|
||
LIMIT 3
|
||
""", (platform, cutoff_time))
|
||
|
||
recent = cursor.fetchall()
|
||
if not recent:
|
||
# 无近期记录 → 放行（可能是新用户的相同提问）
|
||
logger.debug(f"[keyword-scorer] 无近期记录，放行: score={score}, matched={matched[:3]}")
|
||
return False
|
||
|
||
# 有近期记录 → 丢弃（同一用户重复提问）
|
||
logger.debug(f"[keyword-scorer] 发现近期记录，跳过: count={len(recent)}, "
|
||
f"最近时间={recent[0][0]}, score={recent[0][1]}")
|
||
return True
|
||
|
||
finally:
|
||
conn.close()
|