fix(analytics): 扫描器水位>=/无水位自卫/迟到测试判别力/分片flush

Co-Authored-By: Claude Code <noreply@anthropic.com>

yunke-server/app/analytics/sensitive/scanner.py

@@ -2,8 +2,10 @@
 
 水位语义(spec §3.6 P2):last_fetched_at = MAX(message.fetched_at)。增量
 补扫拉回的历史消息 timestamp 早于水位但 fetched_at 是新的,按 timestamp
-做水位会永久漏扫。等值边界行可能重扫——UNIQUE(word_id,msg_id)+已有命中
-跳过兜底,幂等。
+做水位会永久漏扫。比较用 >=:fetched_at 是 MySQL DATETIME 秒精度,SELECT
+快照后同秒 commit 的行 fetched_at == 水位,严格 > 会永久丢弃(只丢不重
+扫)——>= 每 tick 重扫 ≤1 秒边界行,UNIQUE(word_id,msg_id)+已有命中
+跳过去重,幂等(st.scanned 计数随边界重扫微膨胀,可接受)。
 
 方言:MySQL ngram MATCH(≥2字,快);单字词与其他方言走 LIKE。语义差
 异(MATCH 分词 vs LIKE 子串)可接受——命中可处置,误报治理闭环在。
@@ -39,7 +41,12 @@ def _snippet(text: str, word: str) -> str:
 
 
 def _text_match_cond(word: str, dialect: str) -> list:
-    """全量回填路径:MySQL ≥2字 → ngram MATCH;否则 LIKE(单字 ngram 不可用)。"""
+    """全量回填路径:MySQL ≥2字 → ngram MATCH;否则 LIKE(单字 ngram 不可用)。
+
+    已知漏召:ngram 对嵌入标点的拉丁词可能分词不齐(如 "we-chat" 不含
+    "we-chat" 整词 token)——增量路径 Python 子串精确无此问题;起步词
+    全 CJK,当前无影响。
+    """
     if dialect == "mysql" and len(word) >= 2:
         phrase = word.replace('"', "").replace("*", "")
         return [
@@ -115,9 +122,23 @@ async def run_incremental(session: AsyncSession) -> dict:
     ).scalars().all()
     st = await _get_state(session)
 
+    # 无水位自卫:无水位首跑=全表(86 万行 MEDIUMTEXT)进 API 进程内存。
+    # 增量由调度 tick 直调,大表首跑必须走 CLI sensitive-scan --full 建水位。
+    if st.last_fetched_at is None:
+        n = (await session.execute(
+            sa.select(sa.func.count()).select_from(Message)
+        )).scalar_one()
+        if n > 20_000:
+            raise RuntimeError(
+                "message 表超阈值且无扫描水位——首跑请先 CLI sensitive-scan"
+                " --full 建水位"
+            )
+
     conds = [Message.text.is_not(None), Message.text != ""]
     if st.last_fetched_at is not None:
-        conds.append(Message.fetched_at > st.last_fetched_at)
+        # >=(非 >):fetched_at 秒精度,快照后同秒 commit 的行等值——严格 >
+        # 永久丢弃;>= 每 tick 重扫 ≤1s 边界行,existing/UNIQUE 去重。
+        conds.append(Message.fetched_at >= st.last_fetched_at)
     rows = (
         await session.execute(
             sa.select(
@@ -158,6 +179,7 @@ async def run_full(session: AsyncSession) -> dict:
     ).scalars().all()
     st = await _get_state(session)
     total_hits = 0
+    scanned_rows = 0
     for w in words:
         conds = [Message.text.is_not(None), Message.text != ""]
         conds += _text_match_cond(w.word, dialect)
@@ -166,14 +188,17 @@ async def run_full(session: AsyncSession) -> dict:
             Message.talker_id, Message.room_id, Message.timestamp, Message.text,
         ).where(*conds)
         rows = (await session.execute(stmt)).all()
+        scanned_rows += len(rows)  # MATCH 预筛返回行数,按词累计(非去重消息数)
         for i in range(0, len(rows), 500):
             total_hits += await _scan_batch(session, w, rows[i:i + 500])
-        await session.flush()
+            await session.flush()  # 行已 .all() 物化,分片 flush 控挂起对象数
     new_wm = (
         await session.execute(sa.select(sa.func.max(Message.fetched_at)))
     ).scalar_one_or_none()
     if new_wm is not None:
         st.last_fetched_at = new_wm
+    st.scanned += scanned_rows
+    st.hits += total_hits
     st.last_run_at = dt.datetime.now(dt.UTC)
     await session.flush()
     logger.info("[sensitive-scan] full backfill done: words=%s hits=%s",

yunke-server/tests/test_sensitive.py

@@ -137,9 +137,10 @@ async def test_late_arriving_history_scanned(make_client):
         r1 = await scanner.run_incremental(session)
         await session.commit()
         assert r1["hits"] == 1
-        # 迟到行:timestamp 很老(fetched_at 现在)
+        # 迟到行:timestamp 比 m_old 的还老(fetched_at 现在)——
+        # 错误的 timestamp 水位实现会漏掉它,fetched_at 水位必须抓到
         session.add(_msg("m_late", "迟到的私单", dt.datetime(2026, 9, 4, 12, 0),
-                         ts=1767000000000))
+                         ts=1735000000000))
         await session.commit()
         r2 = await scanner.run_incremental(session)
         await session.commit()