From 956f243fcefa62ee6d8f85c543ac54a0ebf2db30 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Fri, 4 Sep 2026 20:30:34 +0800 Subject: [PATCH] =?UTF-8?q?fix(analytics):=20=E6=89=AB=E6=8F=8F=E5=99=A8?= =?UTF-8?q?=E6=B0=B4=E4=BD=8D>=3D/=E6=97=A0=E6=B0=B4=E4=BD=8D=E8=87=AA?= =?UTF-8?q?=E5=8D=AB/=E8=BF=9F=E5=88=B0=E6=B5=8B=E8=AF=95=E5=88=A4?= =?UTF-8?q?=E5=88=AB=E5=8A=9B/=E5=88=86=E7=89=87flush?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- .../app/analytics/sensitive/scanner.py | 35 ++++++++++++++++--- yunke-server/tests/test_sensitive.py | 5 +-- 2 files changed, 33 insertions(+), 7 deletions(-) diff --git a/yunke-server/app/analytics/sensitive/scanner.py b/yunke-server/app/analytics/sensitive/scanner.py index a99f996..190340d 100644 --- a/yunke-server/app/analytics/sensitive/scanner.py +++ b/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", diff --git a/yunke-server/tests/test_sensitive.py b/yunke-server/tests/test_sensitive.py index d4670bb..b8ccfd6 100644 --- a/yunke-server/tests/test_sensitive.py +++ b/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()