From aa59461b03892d1955a8cd35dc5b5ae6a688668e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Fri, 4 Sep 2026 19:55:14 +0800 Subject: [PATCH] =?UTF-8?q?feat(analytics):=20C1=20=E6=89=AB=E6=8F=8F?= =?UTF-8?q?=E5=99=A8=E2=80=94=E2=80=94fetched=5Fat=20=E6=B0=B4=E4=BD=8D?= =?UTF-8?q?=E5=A2=9E=E9=87=8F/=E8=BF=9F=E5=88=B0=E5=85=8D=E7=96=AB/?= =?UTF-8?q?=E5=B9=82=E7=AD=89/=E5=85=A8=E9=87=8F=E5=9B=9E=E5=A1=AB?= 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 | 181 ++++++++++++++++++ yunke-server/tests/test_sensitive.py | 97 +++++++++- 2 files changed, 277 insertions(+), 1 deletion(-) create mode 100644 yunke-server/app/analytics/sensitive/scanner.py diff --git a/yunke-server/app/analytics/sensitive/scanner.py b/yunke-server/app/analytics/sensitive/scanner.py new file mode 100644 index 0000000..a99f996 --- /dev/null +++ b/yunke-server/app/analytics/sensitive/scanner.py @@ -0,0 +1,181 @@ +"""敏感词扫描器——增量水位(fetched_at)+ 首次全量回填。 + +水位语义(spec §3.6 P2):last_fetched_at = MAX(message.fetched_at)。增量 +补扫拉回的历史消息 timestamp 早于水位但 fetched_at 是新的,按 timestamp +做水位会永久漏扫。等值边界行可能重扫——UNIQUE(word_id,msg_id)+已有命中 +跳过兜底,幂等。 + +方言:MySQL ngram MATCH(≥2字,快);单字词与其他方言走 LIKE。语义差 +异(MATCH 分词 vs LIKE 子串)可接受——命中可处置,误报治理闭环在。 + +分批:增量直接拉水位后行集(4h 增量仅数千);全量回填按词扫——MATCH 在 +服务端把 86 万行预筛成命中行(每词仅数千),整批取回后按 500 行分片插 +入/flush 控挂起对象数。不流式消费游标:MySQL 驱动(asyncmy)下结果未 +取尽时同连接发下一条语句会 out-of-sync,全量路径无测试覆盖,取保守形。 +""" +from __future__ import annotations + +import datetime as dt +import logging + +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import AsyncSession + +from app.analytics.sensitive.words import _hit_exclude +from app.db.models import Message, SensitiveHit, SensitiveScanState, SensitiveWord + +logger = logging.getLogger(__name__) + +ENTITY = "message_text" +_SNIPPET_RADIUS = 30 # 命中词前后各取 30 字 + + +def _snippet(text: str, word: str) -> str: + i = text.find(word) + if i < 0: + return text[: _SNIPPET_RADIUS * 2] + lo, hi = max(0, i - _SNIPPET_RADIUS), min(len(text), i + len(word) + _SNIPPET_RADIUS) + return text[lo:hi] + + +def _text_match_cond(word: str, dialect: str) -> list: + """全量回填路径:MySQL ≥2字 → ngram MATCH;否则 LIKE(单字 ngram 不可用)。""" + if dialect == "mysql" and len(word) >= 2: + phrase = word.replace('"', "").replace("*", "") + return [ + sa.text("MATCH(message.text) AGAINST (:w IN BOOLEAN MODE)") + .bindparams(w=f'"{phrase}"') + ] + escaped = word.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + return [Message.text.like(f"%{escaped}%", escape="\\")] + + +async def _get_state(session: AsyncSession) -> SensitiveScanState: + st = await session.get(SensitiveScanState, ENTITY) + if st is None: + st = SensitiveScanState(entity=ENTITY) + session.add(st) + await session.flush() + return st + + +async def _insert_hits( + session: AsyncSession, word: SensitiveWord, rows: list, existing: set[str] +) -> int: + n = 0 + for msg_id, channel, owner, talker, room, ts, text in rows: + if msg_id in existing: + continue + if _hit_exclude(text, word.word, word.exclude_words or None): + continue + session.add( + SensitiveHit( + word_id=word.id, msg_id=msg_id, channel=channel, + owner_account_id=owner, talker_id=talker, room_id=room, + ts=ts, snippet=_snippet(text or "", word.word), + ) + ) + existing.add(msg_id) + n += 1 + return n + + +async def _scan_batch( + session: AsyncSession, word: SensitiveWord, rows: list +) -> int: + """对一个词扫一批 (msg_id..text) 行:查已有命中、插新命中。""" + sel_rows = [tuple(r[:7]) for r in rows if word.word in (r[6] or "")] + if not sel_rows: + return 0 + existing = set( + ( + await session.execute( + sa.select(SensitiveHit.msg_id).where( + SensitiveHit.word_id == word.id, + SensitiveHit.msg_id.in_( + [r[0] for r in sel_rows] or ["-"] + ), + ) + ) + ).scalars().all() + ) + return await _insert_hits(session, word, sel_rows, existing) + + +async def run_incremental(session: AsyncSession) -> dict: + """增量扫描:fetched_at > 水位 的行 × 全部 enabled 词。 + + 增量行少(4h 数千)——统一 Python 侧子串判定(=LIKE 语义,精确;MATCH + 只在全量回填的 MySQL 路径用,见 _text_match_cond)。 + """ + words = ( + await session.execute( + sa.select(SensitiveWord).where(SensitiveWord.enabled.is_(True)) + ) + ).scalars().all() + st = await _get_state(session) + + 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) + rows = ( + await session.execute( + sa.select( + Message.msg_id, Message.channel, Message.owner_account_id, + Message.talker_id, Message.room_id, Message.timestamp, + Message.text, Message.fetched_at, + ).where(*conds) + ) + ).all() + + scanned = len(rows) + hits = 0 + for w in words: + hits += await _scan_batch(session, w, rows) + + new_wm = max((r[7] for r in rows), default=None) if rows else None + if new_wm is not None: + st.last_fetched_at = new_wm + st.last_run_at = dt.datetime.now(dt.UTC) + st.scanned += scanned + st.hits += hits + await session.flush() + return {"scanned": scanned, "hits": hits, "new_watermark": new_wm} + + +async def run_full(session: AsyncSession) -> dict: + """全量回填(CLI 一次性):每词单独扫全库;水位推进同增量。 + + 每词结果整批取回(MATCH/LIKE 已在服务端预筛成命中行)再分片复用 + _scan_batch——不在未取尽的结果上发新语句/flush(MySQL 驱动会 + out-of-sync)。日志只打数量,不打词与正文。 + """ + dialect = session.bind.dialect.name if session.bind else "" + words = ( + await session.execute( + sa.select(SensitiveWord).where(SensitiveWord.enabled.is_(True)) + ) + ).scalars().all() + st = await _get_state(session) + total_hits = 0 + for w in words: + conds = [Message.text.is_not(None), Message.text != ""] + conds += _text_match_cond(w.word, dialect) + stmt = sa.select( + Message.msg_id, Message.channel, Message.owner_account_id, + Message.talker_id, Message.room_id, Message.timestamp, Message.text, + ).where(*conds) + rows = (await session.execute(stmt)).all() + for i in range(0, len(rows), 500): + total_hits += await _scan_batch(session, w, rows[i:i + 500]) + await session.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.last_run_at = dt.datetime.now(dt.UTC) + await session.flush() + logger.info("[sensitive-scan] full backfill done: words=%s hits=%s", + len(words), total_hits) + return {"words": len(words), "hits": total_hits} diff --git a/yunke-server/tests/test_sensitive.py b/yunke-server/tests/test_sensitive.py index 1779dc5..d4670bb 100644 --- a/yunke-server/tests/test_sensitive.py +++ b/yunke-server/tests/test_sensitive.py @@ -3,8 +3,9 @@ from __future__ import annotations import datetime as dt +from app.analytics.sensitive import scanner from app.analytics.sensitive import words as words_svc -from app.db.models import Message +from app.db.models import Message, SensitiveHit, SensitiveScanState async def test_create_and_duplicate(make_client): @@ -92,3 +93,97 @@ async def test_api_create_empty_word_409(make_client): r = await client.post("/api/analytics/sensitive/words", json={"word": " "}) assert r.status_code == 409 + + +# ---------------------------------------------------------------- C1 scanner + +def _msg(msg_id, text, fetched, ts=1756900000000, talker="t1"): + return Message(msg_id=msg_id, channel="wechat", owner_account_id="wx1", + talker_id=talker, timestamp=ts, is_send=False, type="text", + text=text, fetched_at=fetched) + + +async def test_scan_incremental_and_idempotent(make_client): + async def seed(session): + f1 = dt.datetime(2026, 9, 4, 10, 0) + session.add(_msg("m1", "聊聊私单", f1)) + session.add(_msg("m2", "正常", f1)) + await words_svc.create_word(session, word="私单") + await session.commit() + r1 = await scanner.run_incremental(session) + await session.commit() + assert r1["hits"] == 1 + assert r1["new_watermark"] == f1 + r2 = await scanner.run_incremental(session) # 再跑:水位之后无新行 + await session.commit() + assert r2["hits"] == 0 + from sqlalchemy import func, select + + n = (await session.execute( + select(func.count()).select_from(SensitiveHit) + )).scalar_one() + assert n == 1 # 幂等不重复 + + await make_client(seed) + + +async def test_late_arriving_history_scanned(make_client): + """迟到数据:timestamp 很老但 fetched_at 新 → 必须被扫到(P2 水位设计)。""" + async def seed(session): + old = dt.datetime(2026, 9, 1, 8, 0) + session.add(_msg("m_old", "历史私单消息", old)) + await words_svc.create_word(session, word="私单") + await session.commit() + r1 = await scanner.run_incremental(session) + await session.commit() + assert r1["hits"] == 1 + # 迟到行:timestamp 很老(fetched_at 现在) + session.add(_msg("m_late", "迟到的私单", dt.datetime(2026, 9, 4, 12, 0), + ts=1767000000000)) + await session.commit() + r2 = await scanner.run_incremental(session) + await session.commit() + assert r2["hits"] == 1 # 按 fetched_at 水位,迟到历史不漏 + + await make_client(seed) + + +async def test_single_char_like_fallback(make_client): + async def seed(session): + session.add(_msg("m1", "坑爹", dt.datetime(2026, 9, 4, 10, 0))) + await words_svc.create_word(session, word="坑") # 单字 + await session.commit() + r = await scanner.run_incremental(session) + await session.commit() + assert r["hits"] == 1 # 单字走 LIKE 子串,增量行少语义精确 + + await make_client(seed) + + +async def test_disposition_preserved(make_client): + """重扫已处置行不覆盖 disposition。""" + async def seed(session): + session.add(_msg("m1", "私单", dt.datetime(2026, 9, 4, 10, 0))) + await words_svc.create_word(session, word="私单") + await session.commit() + await scanner.run_incremental(session) + await session.commit() + from sqlalchemy import select + + hit = (await session.execute( + select(SensitiveHit) + )).scalars().one() + hit.disposition = "false_positive" + await session.commit() + # 重扫同词同消息(水位回拨模拟) + st = await session.get(SensitiveScanState, "message_text") + st.last_fetched_at = None + await session.commit() + await scanner.run_incremental(session) + await session.commit() + hit2 = (await session.execute( + select(SensitiveHit) + )).scalars().one() + assert hit2.disposition == "false_positive" + + await make_client(seed)