feat(analytics): 数据分析板块四新表+迁移——stat_daily_staff/敏感词三表/fetched_at 索引

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

yunke-server/app/db/models.py

@@ -441,6 +441,13 @@ class Message(Base):
             "owner_account_id",
             "timestamp",
         ),
+        # C1 敏感词增量扫描水位:MAX(fetched_at) / fetched_at > 水位 定位新入库
+        # 消息(按消息 timestamp 做水位会永久漏扫增量拉回的历史消息,spec §3.6)。
+        # 「模型+迁移双声明」防 alembic autogenerate 反 DROP。
+        Index(
+            "ix_message_fetched_at",
+            "fetched_at",
+        ),
     )
 
 
@@ -780,6 +787,117 @@ class MsgDailyStats(Base):
     updated_at: Mapped[dt.datetime] = _updated_at()
 
 
+class StatDailyStaff(Base):
+    """员工×日 汇总(stat_daily_staff)— msg_daily_stats 按账号→员工映射的 rollup。
+
+    数据分析板块 v1 地基(2026-09-04 spec §3.1):A1 员工指标直读;A2 绩效
+    (二批)零基建直读。维护见 app/storage/staff_summary.py——4h 调度刷
+    当日+滚动 3 天 + 夜间全量自愈,**不挂写钩子**(rollup-of-rollup 无需
+    实时,滞后语义与「数据截至」角标一致)。映射不上的账号 user_id 存
+    owner_account_id 原值(读侧标注未映射)。day 口径同 msg_daily_stats
+    (Asia/Shanghai yyyymmdd,Python 侧计算)。
+    """
+
+    __tablename__ = "stat_daily_staff"
+
+    day: Mapped[int] = mapped_column(Integer, primary_key=True)
+    channel: Mapped[str] = mapped_column(String(16), primary_key=True)
+    user_id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    msg_sent: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    msg_recv: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    # 当日活跃去重数(账号合并到员工后跨账号去重 peer)
+    active_friends: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    active_rooms: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    updated_at: Mapped[dt.datetime] = _updated_at()
+
+
+class SensitiveWord(Base):
+    """敏感词库(C1)——词库内容是高敏数据,日志禁止输出词内容。"""
+
+    __tablename__ = "sensitive_word"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    word: Mapped[str] = mapped_column(String(64), unique=True)
+    category: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    enabled: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=True, server_default=sa.text("1")
+    )
+    # 短于此长度不入库(防单字误报;单字词走 LIKE 回退扫描)
+    min_len: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=2, server_default=sa.text("2")
+    )
+    # JSON 数组:命中行文本含任一排除词则跳过(误报抑制)
+    exclude_words: Mapped[list[Any] | dict[str, Any] | None] = mapped_column(
+        JSON, nullable=True
+    )
+    created_at: Mapped[dt.datetime] = _updated_at()
+    updated_at: Mapped[dt.datetime] = _updated_at()
+
+
+class SensitiveHit(Base):
+    """敏感词命中——冗余 channel/owner/talker/room 跳转字段(不回查 message)。
+
+    UNIQUE(word_id, msg_id) 防重扫重复膨胀;disposition ∈
+    pending/confirmed/false_positive/ignored。
+    """
+
+    __tablename__ = "sensitive_hit"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    word_id: Mapped[int] = mapped_column(
+        sa.ForeignKey("sensitive_word.id", ondelete="CASCADE"), nullable=False
+    )
+    msg_id: Mapped[str] = mapped_column(String(128), nullable=False)
+    channel: Mapped[str] = mapped_column(String(16), nullable=False)
+    owner_account_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    talker_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    room_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    ts: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+    # 命中词前后文片段(不含全文,控敏)
+    snippet: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    disposition: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="pending", server_default="pending"
+    )
+    created_at: Mapped[dt.datetime] = _updated_at()
+
+    __table_args__ = (
+        sa.UniqueConstraint("word_id", "msg_id", name="uq_sensitive_hit_word_msg"),
+        sa.Index("ix_sensitive_hit_disp_ts", "disposition", "ts"),
+    )
+
+
+class SensitiveScanState(Base):
+    """敏感词扫描水位——entity v1 只有 'message_text'。
+
+    水位存 message.fetched_at(入库时间,DateTime)而非消息 timestamp:
+    增量补扫拉回的历史消息 timestamp 早于水位但 fetched_at 是新的,按
+    timestamp 做水位会永久漏扫(spec §3.6 P2)。
+    """
+
+    __tablename__ = "sensitive_scan_state"
+
+    entity: Mapped[str] = mapped_column(String(32), primary_key=True)
+    last_fetched_at: Mapped[dt.datetime | None] = mapped_column(
+        sa.DateTime, nullable=True
+    )
+    last_run_at: Mapped[dt.datetime | None] = mapped_column(sa.DateTime, nullable=True)
+    scanned: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    hits: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    updated_at: Mapped[dt.datetime] = _updated_at()
+
+
 __all__ = [
     "ChatGroup",
     "Customer",
@@ -790,8 +908,12 @@ __all__ = [
     "MomentsPost",
     "MsgDailyStats",
     "PhoneCall",
+    "SensitiveHit",
+    "SensitiveScanState",
+    "SensitiveWord",
     "Sms",
     "Staff",
+    "StatDailyStaff",
     "SyncRun",
     "SyncState",
     "VoiceClip",

yunke-server/migrations/versions/f0a1b2c3d4e5_analytics_module_tables.py

@@ -0,0 +1,102 @@
+"""数据分析板块 v1:stat_daily_staff + 敏感词三表 + message.fetched_at 索引
+
+Revision ID: f0a1b2c3d4e5
+Revises: a7d3e9f1c2b4
+Create Date: 2026-09-04
+
+spec: docs/superpowers/specs/2026-09-04-yunke-analytics-module-design.md。
+fetched_at 索引服务 C1 增量水位(MAX(fetched_at) / fetched_at > 水位)。
+幂等:sa.inspect 先查,存在即跳过。SQLite/MySQL 通用。
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "f0a1b2c3d4e5"
+down_revision: str | Sequence[str] | None = "a7d3e9f1c2b4"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _has_index(bind, table: str, index: str) -> bool:
+    insp = sa.inspect(bind)
+    if not insp.has_table(table):
+        return False
+    return index in {ix["name"] for ix in insp.get_indexes(table)}
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    insp = sa.inspect(bind)
+
+    if not insp.has_table("stat_daily_staff"):
+        op.create_table(
+            "stat_daily_staff",
+            sa.Column("day", sa.Integer(), primary_key=True),
+            sa.Column("channel", sa.String(length=16), primary_key=True),
+            sa.Column("user_id", sa.String(length=64), primary_key=True),
+            sa.Column("msg_sent", sa.Integer(), nullable=False, server_default=sa.text("0")),
+            sa.Column("msg_recv", sa.Integer(), nullable=False, server_default=sa.text("0")),
+            sa.Column("active_friends", sa.Integer(), nullable=False, server_default=sa.text("0")),
+            sa.Column("active_rooms", sa.Integer(), nullable=False, server_default=sa.text("0")),
+            sa.Column("updated_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
+        )
+    if not insp.has_table("sensitive_word"):
+        op.create_table(
+            "sensitive_word",
+            sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+            sa.Column("word", sa.String(length=64), nullable=False, unique=True),
+            sa.Column("category", sa.String(length=32), nullable=True),
+            sa.Column("enabled", sa.Boolean(), nullable=False, server_default=sa.text("1")),
+            sa.Column("min_len", sa.Integer(), nullable=False, server_default=sa.text("2")),
+            sa.Column("exclude_words", sa.JSON(), nullable=True),
+            sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
+            sa.Column("updated_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
+        )
+    if not insp.has_table("sensitive_hit"):
+        op.create_table(
+            "sensitive_hit",
+            sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+            sa.Column(
+                "word_id",
+                sa.Integer(),
+                sa.ForeignKey("sensitive_word.id", ondelete="CASCADE"),
+                nullable=False,
+            ),
+            sa.Column("msg_id", sa.String(length=128), nullable=False),
+            sa.Column("channel", sa.String(length=16), nullable=False),
+            sa.Column("owner_account_id", sa.String(length=128), nullable=True),
+            sa.Column("talker_id", sa.String(length=128), nullable=True),
+            sa.Column("room_id", sa.String(length=128), nullable=True),
+            sa.Column("ts", sa.BigInteger(), nullable=True),
+            sa.Column("snippet", sa.String(length=255), nullable=True),
+            sa.Column("disposition", sa.String(length=16), nullable=False, server_default="pending"),
+            sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
+            sa.UniqueConstraint("word_id", "msg_id", name="uq_sensitive_hit_word_msg"),
+        )
+        op.create_index(
+            "ix_sensitive_hit_disp_ts", "sensitive_hit", ["disposition", "ts"]
+        )
+    if not insp.has_table("sensitive_scan_state"):
+        op.create_table(
+            "sensitive_scan_state",
+            sa.Column("entity", sa.String(length=32), primary_key=True),
+            sa.Column("last_fetched_at", sa.DateTime(), nullable=True),
+            sa.Column("last_run_at", sa.DateTime(), nullable=True),
+            sa.Column("scanned", sa.Integer(), nullable=False, server_default=sa.text("0")),
+            sa.Column("hits", sa.Integer(), nullable=False, server_default=sa.text("0")),
+            sa.Column("updated_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
+        )
+    if not _has_index(bind, "message", "ix_message_fetched_at"):
+        op.create_index("ix_message_fetched_at", "message", ["fetched_at"])
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    insp = sa.inspect(bind)
+    if _has_index(bind, "message", "ix_message_fetched_at"):
+        op.drop_index("ix_message_fetched_at", table_name="message")
+    for table in ("sensitive_scan_state", "sensitive_hit", "sensitive_word", "stat_daily_staff"):
+        if insp.has_table(table):
+            op.drop_table(table)

yunke-server/tests/test_analytics_models.py

@@ -0,0 +1,40 @@
+"""新表建模冒烟:create_all 建表 + 唯一约束生效(迁移幂等由 alembic 链保证)。"""
+from __future__ import annotations
+
+import pytest
+from sqlalchemy.exc import IntegrityError
+
+from app.db.models import SensitiveHit, SensitiveWord
+
+
+async def test_sensitive_word_unique(make_client):
+    async def seed(session):
+        session.add(SensitiveWord(word="私单", category="飞单"))
+        await session.flush()
+        # 同词二插必须撞 UNIQUE(word)
+        session.add(SensitiveWord(word="私单"))
+        with pytest.raises(IntegrityError):
+            await session.flush()
+
+    client = await make_client(seed)
+    r = await client.get("/api/health")
+    assert r.status_code == 200  # app boots with new models
+
+
+async def test_hit_unique_word_msg(make_client):
+    async def seed(session):
+        w = SensitiveWord(word="私单")
+        session.add(w)
+        await session.flush()
+        session.add(
+            SensitiveHit(word_id=w.id, msg_id="m1", channel="wechat")
+        )
+        session.add(
+            SensitiveHit(word_id=w.id, msg_id="m1", channel="wechat")
+        )
+        with pytest.raises(IntegrityError):
+            await session.flush()
+
+    client = await make_client(seed)
+    r = await client.get("/api/health")
+    assert r.status_code == 200  # app boots with new models