feat(analytics): C1 敏感词库 CRUD+加词预览——防一加就淹没

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

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

@@ -0,0 +1,130 @@
+"""词库服务——CRUD + 加词前命中量预览(防「一加就淹没」,spec P3)。"""
+from __future__ import annotations
+
+import sqlalchemy as sa
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models import Message, SensitiveWord
+
+PREVIEW_WARN_THRESHOLD = 500  # 预览超此数警示收窄(前端弹提示)
+
+
+async def list_words(session: AsyncSession) -> list[SensitiveWord]:
+    return (
+        (await session.execute(sa.select(SensitiveWord).order_by(SensitiveWord.id)))
+        .scalars().all()
+    )
+
+
+async def create_word(
+    session: AsyncSession, *, word: str, category: str | None = None,
+    min_len: int = 2, exclude_words: list[str] | None = None,
+) -> SensitiveWord:
+    word = (word or "").strip()
+    if not word:
+        raise ValueError("词不能为空")
+    dup = (
+        await session.execute(
+            sa.select(SensitiveWord).where(SensitiveWord.word == word)
+        )
+    ).scalar_one_or_none()
+    if dup is not None:
+        raise ValueError(f"词已存在: id={dup.id}")  # 词是用户输入,回显 id 可
+    obj = SensitiveWord(word=word, category=category, min_len=min_len,
+                        exclude_words=exclude_words)
+    session.add(obj)
+    await session.flush()
+    return obj
+
+
+async def update_word(session: AsyncSession, word_id: int, **fields) -> SensitiveWord:
+    obj = await session.get(SensitiveWord, word_id)
+    if obj is None:
+        raise LookupError("word not found")
+    for k in ("enabled", "category", "min_len", "exclude_words"):
+        if k in fields and fields[k] is not None:
+            setattr(obj, k, fields[k])
+    await session.flush()
+    return obj
+
+
+async def delete_word(session: AsyncSession, word_id: int) -> None:
+    """删词 = 级联删命中(SQLite FK 不强制,显式删;UI 二次确认)。"""
+    obj = await session.get(SensitiveWord, word_id)
+    if obj is None:
+        raise LookupError("word not found")
+    from app.db.models import SensitiveHit
+
+    await session.execute(
+        sa.delete(SensitiveHit).where(SensitiveHit.word_id == word_id)
+    )
+    await session.delete(obj)
+    await session.flush()
+
+
+async def preview_count(
+    session: AsyncSession, word: str, exclude_words: list[str] | None = None
+) -> int:
+    """加词前 COUNT——MySQL 走 ngram MATCH,其他方言 LIKE(测试)。
+
+    口径注:MySQL MATCH 路径不扣排除词(预览=上限,排除词在扫描时精确
+    生效);LIKE 路径(测试库)精确扣。前端文案「预计命中」体现上限语义。
+    """
+    word = (word or "").strip()
+    if not word:
+        return 0
+    conds = [Message.text.is_not(None), Message.text != ""]
+    dialect = session.bind.dialect.name if session.bind else ""
+    if dialect == "mysql" and len(word) >= 2:
+        conds.append(
+            sa.text("MATCH(message.text) AGAINST (:w IN BOOLEAN MODE)")
+            .bindparams(w=f'"{word}"')
+        )
+        stmt = sa.select(sa.func.count()).select_from(Message).where(*conds)
+        n = (await session.execute(stmt)).scalar_one()
+    else:
+        escaped = word.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+        like = f"%{escaped}%"
+        stmt = sa.select(Message.msg_id, Message.text).where(
+            Message.text.is_not(None), Message.text.like(like, escape="\\")
+        )
+        rows = (await session.execute(stmt)).all()
+        n = sum(1 for _mid, t in rows if not _hit_exclude(t, word, exclude_words))
+    return int(n)
+
+
+def _hit_exclude(text: str | None, word: str, exclude_words: list[str] | None) -> bool:
+    """命中行含任一排除词 → 视为误报跳过。"""
+    if not exclude_words or not text:
+        return False
+    return any(ex in text for ex in exclude_words)
+
+
+# 内置起步词库(spec §3.6):幂等补种(CLI sensitive-scan 首跑时调),上线后
+# 管理员在词库页自补行业词。
+STARTER_WORDS: list[tuple[str, str]] = [
+    ("私单", "飞单"),
+    ("飞单", "飞单"),
+    ("加私人微信", "私联"),
+    ("红包", "资金"),
+    ("转账", "资金"),
+    ("离职", "人员风险"),
+]
+
+
+async def ensure_starter_words(session: AsyncSession) -> int:
+    """缺则补(按 word 去重幂等),返回新种入数。"""
+    existing = set(
+        (
+            await session.execute(sa.select(SensitiveWord.word))
+        ).scalars().all()
+    )
+    n = 0
+    for word, category in STARTER_WORDS:
+        if word in existing:
+            continue
+        session.add(SensitiveWord(word=word, category=category))
+        n += 1
+    if n:
+        await session.flush()
+    return n