任务#1570+#1572: 重构消息历史数据库为3表结构，改善OCR查重逻辑
1. 数据库重构为3张表： - user_messages: 用户消息原文 - bot_messages: bot消息原文 - ocr_staging: OCR暂存（重启清空防自言自语） 2. 新增数据迁移：自动将旧表数据迁移到新表 3. 改善OCR查重逻辑： - 得分≥2时：检查与bot消息和用户原始消息的重合率 - 重合率≥50%时丢弃（防bot自言自语和OCR残留） - 得分<2时：重合率≥70%丢弃 4. 新增OCR暂存记录功能： - 记录通过所有过滤器的OCR消息 - 用于后续查重和审计 5. 新增get_recent_user_messages方法： - 获取最近的用户原始消息 - 用于查重逻辑
This commit is contained in:
youpaishidifu
+174
-30
@@ -300,34 +300,109 @@ class MessageHistory:
|
||||
self._init_db()
|
||||
|
||||
def _init_db(self):
|
||||
"""初始化数据库表"""
|
||||
"""初始化数据库表 - 重构为3表结构"""
|
||||
conn = sqlite3.connect(self.db_path, check_same_thread=False)
|
||||
conn.execute("PRAGMA journal_mode=WAL")
|
||||
|
||||
# 1. 用户消息原文表
|
||||
conn.execute("""
|
||||
CREATE TABLE IF NOT EXISTS message_history (
|
||||
CREATE TABLE IF NOT EXISTS user_messages (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
platform TEXT NOT NULL,
|
||||
sender_type TEXT NOT NULL,
|
||||
content TEXT NOT NULL,
|
||||
content_hash TEXT NOT NULL,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
""")
|
||||
conn.execute("""
|
||||
CREATE INDEX IF NOT EXISTS idx_content_hash ON message_history(content_hash)
|
||||
CREATE INDEX IF NOT EXISTS idx_user_content_hash ON user_messages(content_hash)
|
||||
""")
|
||||
conn.execute("""
|
||||
CREATE INDEX IF NOT EXISTS idx_platform_time ON message_history(platform, created_at)
|
||||
CREATE INDEX IF NOT EXISTS idx_user_platform_time ON user_messages(platform, created_at)
|
||||
""")
|
||||
|
||||
# 2. Bot消息原文表
|
||||
conn.execute("""
|
||||
CREATE TABLE IF NOT EXISTS bot_messages (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
platform TEXT NOT NULL,
|
||||
content TEXT NOT NULL,
|
||||
content_hash TEXT NOT NULL,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
""")
|
||||
conn.execute("""
|
||||
CREATE INDEX IF NOT EXISTS idx_bot_content_hash ON bot_messages(content_hash)
|
||||
""")
|
||||
conn.execute("""
|
||||
CREATE INDEX IF NOT EXISTS idx_bot_platform_time ON bot_messages(platform, created_at)
|
||||
""")
|
||||
|
||||
# 3. OCR暂存表（重启清空防自言自语）
|
||||
conn.execute("""
|
||||
CREATE TABLE IF NOT EXISTS ocr_staging (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
platform TEXT NOT NULL,
|
||||
content TEXT NOT NULL,
|
||||
content_hash TEXT NOT NULL,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
""")
|
||||
|
||||
# 检查是否存在旧表，如果存在则迁移数据
|
||||
cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='message_history'")
|
||||
if cursor.fetchone():
|
||||
self._migrate_from_old_table(conn)
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
# 启动时清空OCR暂存表
|
||||
self.clear_ocr_staging()
|
||||
|
||||
def _get_conn(self):
|
||||
"""获取数据库连接"""
|
||||
return sqlite3.connect(self.db_path, check_same_thread=False)
|
||||
|
||||
def _migrate_from_old_table(self, conn):
|
||||
"""从旧表迁移数据到新表"""
|
||||
logger.info("检测到旧表，开始迁移数据...")
|
||||
|
||||
# 迁移用户消息
|
||||
try:
|
||||
cursor = conn.execute("SELECT platform, content, content_hash, created_at FROM message_history WHERE sender_type='user'")
|
||||
rows = cursor.fetchall()
|
||||
if rows:
|
||||
conn.executemany(
|
||||
"INSERT INTO user_messages (platform, content, content_hash, created_at) VALUES (?, ?, ?, ?)",
|
||||
rows
|
||||
)
|
||||
logger.info(f"迁移了 {len(rows)} 条用户消息")
|
||||
except Exception as e:
|
||||
logger.warning(f"迁移用户消息失败: {e}")
|
||||
|
||||
# 迁移bot消息
|
||||
try:
|
||||
cursor = conn.execute("SELECT platform, content, content_hash, created_at FROM message_history WHERE sender_type='bot'")
|
||||
rows = cursor.fetchall()
|
||||
if rows:
|
||||
conn.executemany(
|
||||
"INSERT INTO bot_messages (platform, content, content_hash, created_at) VALUES (?, ?, ?, ?)",
|
||||
rows
|
||||
)
|
||||
logger.info(f"迁移了 {len(rows)} 条bot消息")
|
||||
except Exception as e:
|
||||
logger.warning(f"迁移bot消息失败: {e}")
|
||||
|
||||
# 删除旧表
|
||||
try:
|
||||
conn.execute("DROP TABLE IF EXISTS message_history")
|
||||
logger.info("已删除旧表")
|
||||
except Exception as e:
|
||||
logger.warning(f"删除旧表失败: {e}")
|
||||
|
||||
def record_bot_reply(self, platform: str, content: str):
|
||||
"""记录bot回复"""
|
||||
"""记录bot回复到bot_messages表"""
|
||||
content_hash = hashlib.md5(content.strip().encode('utf-8')).hexdigest()
|
||||
conn = self._get_conn()
|
||||
try:
|
||||
@@ -337,7 +412,7 @@ class MessageHistory:
|
||||
from datetime import datetime
|
||||
local_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
conn.execute(
|
||||
"INSERT INTO message_history (platform, sender_type, content, content_hash, created_at) VALUES (?, 'bot', ?, ?, ?)",
|
||||
"INSERT INTO bot_messages (platform, content, content_hash, created_at) VALUES (?, ?, ?, ?)",
|
||||
(platform, content.strip(), content_hash, local_time)
|
||||
)
|
||||
conn.commit()
|
||||
@@ -347,7 +422,7 @@ class MessageHistory:
|
||||
self._last_bot_reply_content[platform] = content.strip()
|
||||
|
||||
def record_user_message(self, platform: str, content: str):
|
||||
"""记录用户消息（用于后续去重）"""
|
||||
"""记录用户消息到user_messages表（用于后续去重）"""
|
||||
content_hash = hashlib.md5(content.strip().encode('utf-8')).hexdigest()
|
||||
conn = self._get_conn()
|
||||
try:
|
||||
@@ -356,7 +431,7 @@ class MessageHistory:
|
||||
from datetime import datetime
|
||||
local_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
conn.execute(
|
||||
"INSERT INTO message_history (platform, sender_type, content, content_hash, created_at) VALUES (?, 'user', ?, ?, ?)",
|
||||
"INSERT INTO user_messages (platform, content, content_hash, created_at) VALUES (?, ?, ?, ?)",
|
||||
(platform, content.strip(), content_hash, local_time)
|
||||
)
|
||||
conn.commit()
|
||||
@@ -371,9 +446,9 @@ class MessageHistory:
|
||||
content_hash = hashlib.md5(content.strip().encode('utf-8')).hexdigest()
|
||||
conn = self._get_conn()
|
||||
try:
|
||||
# 首先检查是否有相同内容的bot回复
|
||||
# 首先检查是否有相同内容的bot回复（查询bot_messages表）
|
||||
cursor = conn.execute(
|
||||
"SELECT COUNT(*) FROM message_history WHERE platform=? AND sender_type='bot' AND content_hash=?",
|
||||
"SELECT COUNT(*) FROM bot_messages WHERE platform=? AND content_hash=?",
|
||||
(platform, content_hash)
|
||||
)
|
||||
count = cursor.fetchone()[0]
|
||||
@@ -389,7 +464,7 @@ class MessageHistory:
|
||||
if user_msg_time > 0:
|
||||
try:
|
||||
c2 = conn.execute(
|
||||
"SELECT created_at FROM message_history WHERE platform=? AND sender_type='bot' AND content=? ORDER BY id DESC LIMIT 1",
|
||||
"SELECT created_at FROM bot_messages WHERE platform=? AND content=? ORDER BY id DESC LIMIT 1",
|
||||
(platform, bot_content)
|
||||
)
|
||||
row = c2.fetchone()
|
||||
@@ -406,7 +481,7 @@ class MessageHistory:
|
||||
# 如果有相同内容的bot回复，但用户消息时间戳晚于最后bot回复时间，则视为新消息
|
||||
if user_msg_time > 0:
|
||||
cursor = conn.execute(
|
||||
"SELECT created_at FROM message_history WHERE platform=? AND sender_type='bot' ORDER BY id DESC LIMIT 1",
|
||||
"SELECT created_at FROM bot_messages WHERE platform=? ORDER BY id DESC LIMIT 1",
|
||||
(platform,)
|
||||
)
|
||||
row = cursor.fetchone()
|
||||
@@ -432,7 +507,7 @@ class MessageHistory:
|
||||
conn = self._get_conn()
|
||||
try:
|
||||
cursor = conn.execute(
|
||||
"SELECT created_at FROM message_history WHERE platform=? AND sender_type='bot' ORDER BY id DESC LIMIT 1",
|
||||
"SELECT created_at FROM bot_messages WHERE platform=? ORDER BY id DESC LIMIT 1",
|
||||
(platform,)
|
||||
)
|
||||
row = cursor.fetchone()
|
||||
@@ -459,17 +534,66 @@ class MessageHistory:
|
||||
from datetime import datetime, timedelta
|
||||
five_minutes_ago = (datetime.now() - timedelta(minutes=within_minutes)).strftime("%Y-%m-%d %H:%M:%S")
|
||||
cursor = conn.execute(
|
||||
"SELECT content FROM message_history WHERE platform=? AND sender_type='bot' AND created_at > ? ORDER BY id DESC LIMIT ?",
|
||||
"SELECT content FROM bot_messages WHERE platform=? AND created_at > ? ORDER BY id DESC LIMIT ?",
|
||||
(platform, five_minutes_ago, limit)
|
||||
)
|
||||
else:
|
||||
cursor = conn.execute(
|
||||
"SELECT content FROM message_history WHERE platform=? AND sender_type='bot' ORDER BY id DESC LIMIT ?",
|
||||
"SELECT content FROM bot_messages WHERE platform=? ORDER BY id DESC LIMIT ?",
|
||||
(platform, limit)
|
||||
)
|
||||
return [row[0] for row in cursor.fetchall() if row[0]]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def get_recent_user_messages(self, platform: str, limit: int = 20, within_minutes: int = None) -> list:
|
||||
"""获取该平台最近的用户原始消息内容列表，用于查库比对
|
||||
|
||||
Args:
|
||||
within_minutes: 只返回最近N分钟内的消息，None表示不过滤时间
|
||||
"""
|
||||
conn = self._get_conn()
|
||||
try:
|
||||
if within_minutes is not None and within_minutes > 0:
|
||||
from datetime import datetime, timedelta
|
||||
five_minutes_ago = (datetime.now() - timedelta(minutes=within_minutes)).strftime("%Y-%m-%d %H:%M:%S")
|
||||
cursor = conn.execute(
|
||||
"SELECT content FROM user_messages WHERE platform=? AND created_at > ? ORDER BY id DESC LIMIT ?",
|
||||
(platform, five_minutes_ago, limit)
|
||||
)
|
||||
else:
|
||||
cursor = conn.execute(
|
||||
"SELECT content FROM user_messages WHERE platform=? ORDER BY id DESC LIMIT ?",
|
||||
(platform, limit)
|
||||
)
|
||||
return [row[0] for row in cursor.fetchall() if row[0]]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def record_ocr_message(self, platform: str, content: str):
|
||||
"""记录OCR识别的消息到暂存表"""
|
||||
content_hash = hashlib.md5(content.strip().encode('utf-8')).hexdigest()
|
||||
conn = self._get_conn()
|
||||
try:
|
||||
from datetime import datetime
|
||||
local_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
conn.execute(
|
||||
"INSERT INTO ocr_staging (platform, content, content_hash, created_at) VALUES (?, ?, ?, ?)",
|
||||
(platform, content.strip(), content_hash, local_time)
|
||||
)
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def clear_ocr_staging(self):
|
||||
"""清空OCR暂存表（重启时调用）"""
|
||||
conn = self._get_conn()
|
||||
try:
|
||||
conn.execute("DELETE FROM ocr_staging")
|
||||
conn.commit()
|
||||
logger.info("已清空OCR暂存表")
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
class MessageMonitor:
|
||||
@@ -4132,33 +4256,47 @@ class MessageMonitor:
|
||||
logger.debug(f"[{platform}] OCR 过滤: 回复内容片段 line={line[:60]!r}")
|
||||
continue
|
||||
|
||||
# ⭐ Task #1441: bot自身消息查库过滤
|
||||
# 检查OCR读到的内容是否与数据库中bot回复的消息重合率
|
||||
# - 得分≥2时：重合率≥50%丢弃（防bot自言自语）
|
||||
# ⭐ Task #1441 + Task #1572: bot自身消息查库过滤（增强版）
|
||||
# 检查OCR读到的内容是否与数据库中bot回复的消息或用户原始消息重合率
|
||||
# - 得分≥2时：重合率≥50%丢弃（防bot自言自语/OCR残留）
|
||||
# - 得分<2时：重合率≥70%丢弃（放宽阈值）
|
||||
# 原理：用户每个问法不可能完全一样，重合率≥50%大概率是OCR抓到残留消息
|
||||
if len(line) > 10 and self._msg_history:
|
||||
try:
|
||||
import difflib
|
||||
# 查询数据库中最近的bot回复（限制20条）
|
||||
# 获取当前消息的关键词得分
|
||||
score = 0
|
||||
threshold = 0.7 # 默认阈值
|
||||
if self._keyword_scorer:
|
||||
try:
|
||||
score = self._keyword_scorer.get_last_score(platform, line)
|
||||
if score >= 2:
|
||||
threshold = 0.5 # 高分消息降低阈值
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 检查bot消息
|
||||
recent_bot_replies = self._msg_history.get_recent_bot_replies(platform, limit=20)
|
||||
for bot_reply in recent_bot_replies:
|
||||
if len(bot_reply) < 10:
|
||||
continue
|
||||
# 计算重合率
|
||||
sm = difflib.SequenceMatcher(None, line, bot_reply)
|
||||
ratio = sm.ratio()
|
||||
# 获取当前消息的关键词得分
|
||||
threshold = 0.7 # 默认阈值
|
||||
if self._keyword_scorer:
|
||||
try:
|
||||
score = self._keyword_scorer.get_last_score(platform, line)
|
||||
if score >= 2:
|
||||
threshold = 0.5 # 高分消息降低阈值
|
||||
except Exception:
|
||||
pass
|
||||
if ratio >= threshold:
|
||||
logger.debug(f"[{platform}] OCR 过滤: bot自身消息(查库重合率{ratio:.2f}≥{threshold:.0%}) line={line[:60]!r}")
|
||||
break # 匹配到任一bot回复即丢弃
|
||||
else:
|
||||
# 检查用户原始消息（仅高分消息才检查，避免误杀）
|
||||
if score >= 2:
|
||||
recent_user_msgs = self._msg_history.get_recent_user_messages(platform, limit=20)
|
||||
for user_msg in recent_user_msgs:
|
||||
if len(user_msg) < 10:
|
||||
continue
|
||||
sm = difflib.SequenceMatcher(None, line, user_msg)
|
||||
ratio = sm.ratio()
|
||||
if ratio >= 0.5: # 与用户消息重合率≥50%即丢弃
|
||||
logger.debug(f"[{platform}] OCR 过滤: 用户残留消息(查库重合率{ratio:.2f}≥50%) line={line[:60]!r}")
|
||||
break
|
||||
except Exception as e:
|
||||
logger.debug(f"[{platform}] OCR 查库过滤异常: {e}")
|
||||
|
||||
@@ -9104,6 +9242,12 @@ class MessageMonitor:
|
||||
self._last_msg[platform] = _best_cleaned
|
||||
engine_tag = self._last_chosen_engine.get(platform, "")
|
||||
logger.info(f"[{platform}] 新消息[{engine_tag}]: {_best_raw[:60]}")
|
||||
# ⭐ Task #1572: 记录OCR消息到暂存表（用于后续查重）
|
||||
if self._msg_history:
|
||||
try:
|
||||
self._msg_history.record_ocr_message(platform, _best_cleaned)
|
||||
except Exception as e:
|
||||
logger.debug(f"[{platform}] OCR暂存记录失败: {e}")
|
||||
# ⚠️ 急停检查：回复前最后检查，防止急停后仍然发送
|
||||
if not self._running:
|
||||
return
|
||||
|
||||
Reference in New Issue
Block a user