From 565006239b8ab5e3d87ba2a41438e177da2b6e00 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Fri, 4 Sep 2026 15:04:33 +0800 Subject: [PATCH] =?UTF-8?q?feat(analytics):=20=E6=95=B0=E6=8D=AE=E5=88=86?= =?UTF-8?q?=E6=9E=90=E6=9D=BF=E5=9D=97=E5=9B=9B=E6=96=B0=E8=A1=A8+?= =?UTF-8?q?=E8=BF=81=E7=A7=BB=E2=80=94=E2=80=94stat=5Fdaily=5Fstaff/?= =?UTF-8?q?=E6=95=8F=E6=84=9F=E8=AF=8D=E4=B8=89=E8=A1=A8/fetched=5Fat=20?= =?UTF-8?q?=E7=B4=A2=E5=BC=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- yunke-server/app/db/models.py | 122 ++++++++++++++++++ .../f0a1b2c3d4e5_analytics_module_tables.py | 102 +++++++++++++++ yunke-server/tests/test_analytics_models.py | 40 ++++++ 3 files changed, 264 insertions(+) create mode 100644 yunke-server/migrations/versions/f0a1b2c3d4e5_analytics_module_tables.py create mode 100644 yunke-server/tests/test_analytics_models.py diff --git a/yunke-server/app/db/models.py b/yunke-server/app/db/models.py index 053b311..28e8cef 100644 --- a/yunke-server/app/db/models.py +++ b/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", diff --git a/yunke-server/migrations/versions/f0a1b2c3d4e5_analytics_module_tables.py b/yunke-server/migrations/versions/f0a1b2c3d4e5_analytics_module_tables.py new file mode 100644 index 0000000..e7783c3 --- /dev/null +++ b/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) diff --git a/yunke-server/tests/test_analytics_models.py b/yunke-server/tests/test_analytics_models.py new file mode 100644 index 0000000..92a18d6 --- /dev/null +++ b/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