重构时间戳过滤逻辑：按照思维导图实现3条路 (Task #1686)
问题： - OCR继承机制把旧时间戳(1:21:11)关联到新消息，优先级高于新鲜时间戳(18:21:07) - 导致新消息被误判为陈旧消息丢弃 修复： - 第一关：有5分钟内新鲜时间戳（自身时间戳，不是继承的）→ 通关 - 第二关：没有新鲜时间戳 → 查库 - 有数据库记录：本地时间-记录时间戳≤300秒 → 通关 - 有数据库记录：本地时间-记录时间戳>300秒 → 拦截 - 无数据库记录 → 拦截 新增方法： - get_latest_record_timestamp()：获取该平台最近一条记录的时间戳
This commit is contained in:
youpaishidifu
+58
-39
@@ -965,6 +965,26 @@ class MessageHistory:
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def get_latest_record_timestamp(self, platform: str) -> Optional[datetime]:
|
||||
"""获取该平台最近一条记录的时间戳（用于第二关判断）"""
|
||||
self._ensure_ocr_staging_table()
|
||||
conn = self._get_conn()
|
||||
try:
|
||||
cursor = conn.execute(
|
||||
"SELECT timestamp FROM ocr_staging WHERE platform=? ORDER BY timestamp DESC LIMIT 1",
|
||||
(platform,)
|
||||
)
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
# timestamp 字段格式为 "YYYY-MM-DD HH:MM:SS"
|
||||
try:
|
||||
return datetime.strptime(row[0], "%Y-%m-%d %H:%M:%S")
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
return None
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
class MessageMonitor:
|
||||
"""多线程消息监控器"""
|
||||
@@ -10889,52 +10909,51 @@ class MessageMonitor:
|
||||
logger.debug(f"[{platform}] freshness_score查找: item={item!r:.30}, score={freshness_score:.1f}, "
|
||||
f"在字典中={_in_dict}")
|
||||
|
||||
# ⭐ Task #1647: 陈旧消息过滤策略重构
|
||||
# - 超过1小时的消息：freshness_score == -10.0（扣10分）
|
||||
# - 选择消息时分数必须>0才能处理
|
||||
# - DB有记录的消息：不加分（避免重复处理）
|
||||
# ⭐ Task #1685 修复: 区分"自身时间戳陈旧"和"继承时间戳陈旧"
|
||||
# 继承时间戳的消息不应被强制丢弃（可能是新消息继承了旧时间戳）
|
||||
# ⭐ Task #1686: 按照思维导图的3条路逻辑重新实现
|
||||
# 第一关：有5分钟内新鲜时间戳（自身时间戳，不是继承的）
|
||||
# 第二关：没有新鲜时间戳 → 查库
|
||||
_is_inherited_ts = isinstance(item, tuple) and len(item) == 2 and item[1] is not None
|
||||
if freshness_score >= 5:
|
||||
# ≥5分（5分钟内）：直接放行，确保最终分数至少为5
|
||||
|
||||
# 第一关：检查是否有5分钟内的新鲜时间戳（自身时间戳）
|
||||
# freshness_score >= 5 表示时间戳在5分钟内
|
||||
# 但必须是自身时间戳，不是继承的
|
||||
_has_fresh_own_ts = (freshness_score >= 5) and not _is_inherited_ts
|
||||
if _has_fresh_own_ts:
|
||||
# 第一关通过：有5分钟内新鲜时间戳（自身时间戳）
|
||||
old_score = score
|
||||
score = max(score + 5.0, 5.0)
|
||||
logger.info(f"[{platform}] 时间戳新鲜度≥5直接放行(+5): {old_score:.1f}→{score:.1f}: {item[:50]}")
|
||||
elif freshness_score == -999.0:
|
||||
# ⭐ Task #1661+#1665: 无有效时间戳（整批OCR丢失时间戳）
|
||||
# 完全无法判断消息年龄，直接强制丢弃
|
||||
score = -100.0
|
||||
logger.info(f"[{platform}] 无有效时间戳(OCR丢失?)强制丢弃: {item[:50]}")
|
||||
elif freshness_score < 0:
|
||||
# ⭐ Task #1684: 负分（超过1小时）→ 强制丢弃
|
||||
# ⭐ Task #1685 修复: 继承时间戳的消息使用DB查重兜底，不强制丢弃
|
||||
if _is_inherited_ts:
|
||||
# 继承时间戳的消息：检查DB是否有记录，有记录说明是新消息
|
||||
if self._msg_history and self._msg_history.has_ocr_message_record(platform, cleaned_item):
|
||||
# DB有记录：说明是新消息（刚入库），继承了旧时间戳，放行
|
||||
logger.info(f"[{platform}] 继承时间戳陈旧但DB有记录(新消息)，放行: {item[:50]}")
|
||||
logger.info(f"[{platform}] 第一关通过：有新鲜时间戳(自身)，放行: {old_score:.1f}→{score:.1f}: {item[:50]}")
|
||||
else:
|
||||
# DB无记录：可能是旧消息，强制丢弃
|
||||
score = -100.0
|
||||
logger.info(f"[{platform}] 继承时间戳陈旧且DB无记录，强制丢弃: {item[:50]}")
|
||||
else:
|
||||
# 自身时间戳陈旧：强制丢弃
|
||||
score = -100.0
|
||||
logger.info(f"[{platform}] 自身时间戳陈旧(>{abs(freshness_score)/10:.0f}h)，强制丢弃: {item[:50]}")
|
||||
else:
|
||||
# 0-4分（10分钟-1小时）：查DB决定是否加分
|
||||
if self._msg_history and self._msg_history.has_ocr_message_record(platform, cleaned_item):
|
||||
# DB有记录：不加分
|
||||
logger.info(f"[{platform}] 时间戳新鲜度<5且DB有记录，不加分: {item[:50]}")
|
||||
elif self._msg_history and self._msg_history.has_any_records_for_platform(platform):
|
||||
# DB存在但该消息无记录：补分到5（确保最终分数至少为5，不因负基础分而被丢弃）
|
||||
# 第一关未通过：没有新鲜时间戳，进入第二关
|
||||
logger.debug(f"[{platform}] 第一关未通过：无新鲜时间戳(自身)，进入第二关: {item[:50]}")
|
||||
|
||||
# 第二关：查库判断
|
||||
if self._msg_history:
|
||||
latest_ts = self._msg_history.get_latest_record_timestamp(platform)
|
||||
has_record = self._msg_history.has_ocr_message_record(platform, cleaned_item)
|
||||
|
||||
if latest_ts is not None:
|
||||
# 有数据库记录：计算本地时间与记录时间戳的差值
|
||||
time_diff_seconds = (datetime.now() - latest_ts).total_seconds()
|
||||
logger.debug(f"[{platform}] 第二关：有DB记录，时间差={time_diff_seconds:.0f}秒: {item[:50]}")
|
||||
|
||||
if time_diff_seconds <= 300:
|
||||
# 本地时间 - 记录时间戳 ≤ 300秒：通关
|
||||
old_score = score
|
||||
score = max(score + 5.0, 5.0)
|
||||
logger.info(f"[{platform}] 时间戳新鲜度<5且DB无记录，补分到5: {old_score:.1f}→{score:.1f}: {item[:50]}")
|
||||
logger.info(f"[{platform}] 第二关通过：DB记录时间差{time_diff_seconds:.0f}秒≤300秒，放行: {old_score:.1f}→{score:.1f}: {item[:50]}")
|
||||
else:
|
||||
# ⭐ Task #1662: DB不存在或无记录时跳过，不加分（避免旧消息因补分逻辑被误处理）
|
||||
logger.info(f"[{platform}] 时间戳新鲜度<5且DB无记录/不存在，跳过补分: {item[:50]}")
|
||||
# 本地时间 - 记录时间戳 > 300秒：返团拦截
|
||||
score = -100.0
|
||||
logger.info(f"[{platform}] 第二关拦截：DB记录时间差{time_diff_seconds:.0f}秒>300秒，丢弃: {item[:50]}")
|
||||
else:
|
||||
# 无数据库记录：返团拦截
|
||||
score = -100.0
|
||||
logger.info(f"[{platform}] 第二关拦截：无DB记录，丢弃: {item[:50]}")
|
||||
else:
|
||||
# msg_history不可用：强制丢弃
|
||||
score = -100.0
|
||||
logger.info(f"[{platform}] msg_history不可用，强制丢弃: {item[:50]}")
|
||||
|
||||
# ⭐ 检查分数是否>0：分数<=0的消息不应被处理
|
||||
if score <= 0:
|
||||
|
||||
Reference in New Issue
Block a user