实现消息回复状态机制替代_seen_messages集合

- Task #1673: 在user_messages表添加reply_status和fail_count字段
- 实现mark_message_replied和mark_message_send_failed方法
- SQLite UPDATE语法修复:改用子查询方式避免ORDER BY+LIMIT错误
- 在gui.py中集成发送成功/失败状态标记
- 事不过三原则:连续3次发送失败后强制标记为已回复
This commit is contained in:
youpaishidifu
2026-09-02 12:34:35 +08:00
parent 52fae0c1e2
commit f43ab86308
2 changed files with 113 additions and 2 deletions
+6
View File
@@ -1380,6 +1380,9 @@ class MsgBotApp(ctk.CTk):
ok = self.monitor._paste_and_send(platform, answer, force_retry=(attempt > 0)) ok = self.monitor._paste_and_send(platform, answer, force_retry=(attempt > 0))
if ok: if ok:
self._log(f"[{platform}] 🤖 已自动回复: {answer[:60]}...") self._log(f"[{platform}] 🤖 已自动回复: {answer[:60]}...")
# ⭐ Task #1673: 发送成功,标记消息已回复
if msg:
self.monitor.mark_message_replied(platform, msg)
return # 成功直接退出 return # 成功直接退出
else: else:
if attempt < max_retries - 1: if attempt < max_retries - 1:
@@ -1394,6 +1397,9 @@ class MsgBotApp(ctk.CTk):
self.after(int(retry_delay * 1000), lambda d=data: self._execute_reply(d)) self.after(int(retry_delay * 1000), lambda d=data: self._execute_reply(d))
else: else:
self._log(f"[{platform}] ⚠ 自动回复发送失败(连续{max_retries}次),用户可能在接管,本轮跳过") self._log(f"[{platform}] ⚠ 自动回复发送失败(连续{max_retries}次),用户可能在接管,本轮跳过")
# ⭐ Task #1673: 连续3次发送失败,标记消息发送失败
if msg:
self.monitor.mark_message_send_failed(platform, msg)
finally: finally:
self.monitor._sending_event.clear() self.monitor._sending_event.clear()
# ⭐ 回复完成或失败后清除进行中标志 # ⭐ 回复完成或失败后清除进行中标志
+107 -2
View File
@@ -558,12 +558,25 @@ class MessageHistory:
platform TEXT NOT NULL, platform TEXT NOT NULL,
content TEXT NOT NULL, content TEXT NOT NULL,
content_hash TEXT NOT NULL, content_hash TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
reply_status INTEGER DEFAULT 1,
fail_count INTEGER DEFAULT 0
) )
""") """)
conn.execute("CREATE INDEX IF NOT EXISTS idx_user_content_hash ON user_messages(content_hash)") conn.execute("CREATE INDEX IF NOT EXISTS idx_user_content_hash ON user_messages(content_hash)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_user_platform_time ON user_messages(platform, created_at)") conn.execute("CREATE INDEX IF NOT EXISTS idx_user_platform_time ON user_messages(platform, created_at)")
conn.commit() conn.commit()
else:
# ⭐ Task #1673: 检查并添加 reply_status 和 fail_count 字段(如果不存在)
cursor = conn.execute("PRAGMA table_info(user_messages)")
columns = [row[1] for row in cursor.fetchall()]
if 'reply_status' not in columns:
logger.info("[msg-history] user_messages表添加 reply_status 字段")
conn.execute("ALTER TABLE user_messages ADD COLUMN reply_status INTEGER DEFAULT 1")
if 'fail_count' not in columns:
logger.info("[msg-history] user_messages表添加 fail_count 字段")
conn.execute("ALTER TABLE user_messages ADD COLUMN fail_count INTEGER DEFAULT 0")
conn.commit()
finally: finally:
conn.close() conn.close()
@@ -609,6 +622,93 @@ class MessageHistory:
finally: finally:
conn.close() conn.close()
def mark_message_replied(self, platform: str, content: str):
"""标记消息已回复(reply_status=0)"""
# ⭐ Task #1673: 用数据库状态机制替代内存中的_seen_messages集合
self._ensure_user_messages_table()
conn = self._get_conn()
try:
# ⭐ Task #1676: SQLite不支持UPDATE+ORDER BY+LIMIT,改用子查询找到最大id
# 先找到最近一条匹配消息的id
cursor = conn.execute(
"""SELECT id FROM user_messages
WHERE platform=? AND content=? AND reply_status=1
ORDER BY id DESC LIMIT 1""",
(platform, content.strip())
)
row = cursor.fetchone()
if row:
msg_id = row[0]
# 更新该条消息的 reply_status 为 0(已回复)
conn.execute(
"UPDATE user_messages SET reply_status=0 WHERE id=?",
(msg_id,)
)
conn.commit()
except Exception as e:
logger.error(f"[{platform}] 标记消息已回复失败: {e}")
finally:
conn.close()
def mark_message_send_failed(self, platform: str, content: str):
"""标记消息发送失败(增加 fail_count,连续3次失败则标记为已回复)"""
# ⭐ Task #1673: 发送失败的事不过三原则
self._ensure_user_messages_table()
conn = self._get_conn()
try:
# ⭐ Task #1676: SQLite不支持UPDATE+ORDER BY+LIMIT,改用子查询找到最大id
# 先找到最近一条匹配消息的id
cursor = conn.execute(
"""SELECT id FROM user_messages
WHERE platform=? AND content=? AND reply_status=1
ORDER BY id DESC LIMIT 1""",
(platform, content.strip())
)
row = cursor.fetchone()
if row:
msg_id = row[0]
# 增加 fail_count
conn.execute(
"UPDATE user_messages SET fail_count=fail_count+1 WHERE id=?",
(msg_id,)
)
# 检查 fail_count 是否达到3次
cursor = conn.execute(
"SELECT fail_count FROM user_messages WHERE id=?",
(msg_id,)
)
row = cursor.fetchone()
if row and row[0] >= 3:
# 连续3次失败,强制标记为已回复
conn.execute(
"UPDATE user_messages SET reply_status=0 WHERE id=?",
(msg_id,)
)
logger.info(f"[{platform}] 消息连续3次发送失败,强制标记为已回复: {content[:40]}")
conn.commit()
except Exception as e:
logger.error(f"[{platform}] 标记消息发送失败失败: {e}")
finally:
conn.close()
def is_message_already_replied(self, platform: str, content: str) -> bool:
"""检查消息是否已回复(reply_status=0)"""
# ⭐ Task #1673: 轮询时检查 reply_status,为0则跳过
self._ensure_user_messages_table()
conn = self._get_conn()
try:
cursor = conn.execute(
"""SELECT COUNT(*) FROM user_messages
WHERE platform=? AND content=? AND reply_status=0""",
(platform, content.strip())
)
return cursor.fetchone()[0] > 0
except Exception as e:
logger.error(f"[{platform}] 检查消息是否已回复失败: {e}")
return False
finally:
conn.close()
def record_user_message(self, platform: str, content: str): def record_user_message(self, platform: str, content: str):
"""记录用户消息到user_messages表(环形缓冲区,max 10条/平台)""" """记录用户消息到user_messages表(环形缓冲区,max 10条/平台)"""
# ⭐ Task #1655: 确保表存在(防御性编程) # ⭐ Task #1655: 确保表存在(防御性编程)
@@ -651,8 +751,9 @@ class MessageHistory:
" ORDER BY created_at ASC LIMIT ?" " ORDER BY created_at ASC LIMIT ?"
")", (platform, count - 9) ")", (platform, count - 9)
) )
# ⭐ Task #1673: 添加 reply_status 和 fail_count 字段
conn.execute( conn.execute(
"INSERT INTO user_messages (platform, content, content_hash, created_at) VALUES (?, ?, ?, ?)", "INSERT INTO user_messages (platform, content, content_hash, created_at, reply_status, fail_count) VALUES (?, ?, ?, ?, 1, 0)",
(platform, content.strip(), content_hash, local_time) (platform, content.strip(), content_hash, local_time)
) )
conn.commit() conn.commit()
@@ -10335,6 +10436,10 @@ class MessageMonitor:
if matched_r: if matched_r:
logger.debug(f"[{platform}] 回复公共子串跳过: {ts[:50]}") logger.debug(f"[{platform}] 回复公共子串跳过: {ts[:50]}")
continue continue
# ⭐ Task #1673: 先检查数据库中的 reply_status,为0则跳过
if self._msg_history and self._msg_history.is_message_already_replied(platform, ts):
logger.debug(f"[{platform}] 数据库查重跳过(reply_status=0): {ts[:50]}")
continue
if ts in seen: if ts in seen:
logger.debug(f"[{platform}] seen去重跳过: {ts[:50]}") logger.debug(f"[{platform}] seen去重跳过: {ts[:50]}")
continue continue