diff --git a/yunke-server/app/analytics/sensitive/__init__.py b/yunke-server/app/analytics/sensitive/__init__.py
new file mode 100644
index 0000000..b9cdfe3
--- /dev/null
+++ b/yunke-server/app/analytics/sensitive/__init__.py
@@ -0,0 +1,5 @@
+"""C1 敏感词子系统(spec §3.6)——words(词库)/scanner(扫描)/hits(处置)。
+
+内容源抽象:scanner 目前扫 message.text;预留 refer_msg 引文与 D4 ASR
+转写文本接入位(二/三批)。日志禁打词内容与命中正文。
+"""
diff --git a/yunke-server/app/analytics/sensitive/words.py b/yunke-server/app/analytics/sensitive/words.py
new file mode 100644
index 0000000..935f2c0
--- /dev/null
+++ b/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
diff --git a/yunke-server/app/api/routers/analytics.py b/yunke-server/app/api/routers/analytics.py
index feb34e1..6a12617 100644
--- a/yunke-server/app/api/routers/analytics.py
+++ b/yunke-server/app/api/routers/analytics.py
@@ -4,10 +4,12 @@ from __future__ import annotations
 from typing import Annotated, Any, Literal
 
 from fastapi import APIRouter, Depends, HTTPException, Query
+from pydantic import BaseModel
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.analytics import health as health_svc
 from app.analytics import metrics as metrics_svc
+from app.analytics.sensitive import words as words_svc
 from app.api.deps import get_session
 
 router = APIRouter(prefix="/api/analytics", tags=["analytics"])
@@ -69,3 +71,68 @@ async def unreplied_list(
     return await metrics_svc.unreplied(
         session, channel=channel, hours=hours, page=page, page_size=page_size
     )
+
+
+class WordIn(BaseModel):
+    word: str
+    category: str | None = None
+    min_len: int = 2
+    exclude_words: list[str] | None = None
+
+
+class WordPreviewIn(BaseModel):
+    word: str
+    exclude_words: list[str] | None = None
+
+
+@router.get("/sensitive/words")
+async def list_words(session: SessionDep) -> list[dict]:
+    """C1 词库清单(enabled=false 的词 scanner 跳过)。"""
+    return [
+        {"id": w.id, "word": w.word, "category": w.category, "enabled": w.enabled,
+         "min_len": w.min_len, "exclude_words": w.exclude_words or []}
+        for w in await words_svc.list_words(session)
+    ]
+
+
+@router.post("/sensitive/words")
+async def create_word(session: SessionDep, body: WordIn) -> dict:
+    """加词(重复 409)。词内容是敏感数据,错误信息只回 id 不回词。"""
+    try:
+        w = await words_svc.create_word(
+            session, word=body.word, category=body.category,
+            min_len=body.min_len, exclude_words=body.exclude_words,
+        )
+        await session.commit()
+        return {"id": w.id, "word": w.word, "enabled": w.enabled}
+    except ValueError as exc:
+        raise HTTPException(409, str(exc)) from exc
+
+
+@router.post("/sensitive/words/preview")
+async def preview_word(session: SessionDep, body: WordPreviewIn) -> dict:
+    """加词前命中量预览(count 为 MySQL 上限语义;warn=超阈值建议收窄)。"""
+    n = await words_svc.preview_count(session, body.word, body.exclude_words)
+    return {"count": n, "warn": n >= words_svc.PREVIEW_WARN_THRESHOLD}
+
+
+@router.patch("/sensitive/words/{word_id}")
+async def update_word(session: SessionDep, word_id: int, body: dict) -> dict:
+    """改词(启停/分类/排除词);字段 None=不改。"""
+    try:
+        w = await words_svc.update_word(session, word_id, **body)
+        await session.commit()
+        return {"id": w.id, "enabled": w.enabled, "category": w.category}
+    except LookupError as exc:
+        raise HTTPException(404, str(exc)) from exc
+
+
+@router.delete("/sensitive/words/{word_id}")
+async def delete_word(session: SessionDep, word_id: int) -> dict:
+    """删词=级联删命中(UI 二次确认)。"""
+    try:
+        await words_svc.delete_word(session, word_id)
+        await session.commit()
+        return {"ok": True}
+    except LookupError as exc:
+        raise HTTPException(404, str(exc)) from exc
diff --git a/yunke-server/tests/test_sensitive.py b/yunke-server/tests/test_sensitive.py
new file mode 100644
index 0000000..79de177
--- /dev/null
+++ b/yunke-server/tests/test_sensitive.py
@@ -0,0 +1,61 @@
+"""C1 词库 CRUD + 预览。"""
+from __future__ import annotations
+
+import datetime as dt
+
+from app.analytics.sensitive import words as words_svc
+from app.db.models import Message
+
+
+async def test_create_and_duplicate(make_client):
+    async def seed(session):
+        w = await words_svc.create_word(
+            session, word="飞单", category="私单", exclude_words=["飞单群"]
+        )
+        assert w.id is not None and w.enabled is True
+        try:
+            await words_svc.create_word(session, word="飞单")
+            raise AssertionError("should raise")
+        except ValueError:
+            pass
+
+    await make_client(seed)
+
+
+async def test_preview_count_and_excludes(make_client):
+    async def seed(session):
+        ts = 1756900000000
+        for i, text in enumerate(["今天聊飞单的事", "飞单群公告", "正常聊天"]):
+            session.add(Message(msg_id=f"m{i}", channel="wechat",
+                                owner_account_id="wx1", talker_id="t1",
+                                timestamp=ts + i, is_send=False, type="text",
+                                text=text, fetched_at=dt.datetime(2026, 9, 4)))
+        await session.commit()
+        n1 = await words_svc.preview_count(session, "飞单", None)
+        assert n1 == 2  # SQLite 测试库走 LIKE 路径
+        n2 = await words_svc.preview_count(session, "飞单", ["飞单群"])
+        assert n2 == 1  # 排除词剔除「飞单群公告」
+
+    await make_client(seed)
+
+
+async def test_api_words_crud(make_client):
+    async def seed(session):
+        ...
+
+    client = await make_client(seed)
+    r = await client.post("/api/analytics/sensitive/words",
+                          json={"word": "私单", "category": "飞单"})
+    assert r.status_code == 200 and r.json()["word"] == "私单"
+    r = await client.post("/api/analytics/sensitive/words",
+                          json={"word": "私单"})
+    assert r.status_code == 409
+    r = await client.post("/api/analytics/sensitive/words/preview",
+                          json={"word": "私单"})
+    assert r.status_code == 200 and r.json()["count"] == 0
+    wid = 1
+    r = await client.patch(f"/api/analytics/sensitive/words/{wid}",
+                           json={"enabled": False})
+    assert r.status_code == 200 and r.json()["enabled"] is False
+    r = await client.delete(f"/api/analytics/sensitive/words/{wid}")
+    assert r.status_code == 200