From aecb556efd08cdb92615efdf812dce15b640d4e7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Fri, 4 Sep 2026 14:25:20 +0800 Subject: [PATCH] =?UTF-8?q?docs(plan):=20=E6=95=B0=E6=8D=AE=E5=88=86?= =?UTF-8?q?=E6=9E=90=E6=9D=BF=E5=9D=97=20v1=20=E5=AE=9E=E6=96=BD=E8=AE=A1?= =?UTF-8?q?=E5=88=92=E2=80=94=E2=80=9417=20=E4=BB=BB=E5=8A=A1=20TDD=20?= =?UTF-8?q?=E5=85=A8=E9=93=BE(=E6=A8=A1=E5=9E=8B/=E6=8C=87=E6=A0=87/C1/?= =?UTF-8?q?=E8=B0=83=E5=BA=A6/=E5=9B=9B=E9=A1=B5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- .../2026-09-04-yunke-analytics-module.md | 3423 +++++++++++++++++ 1 file changed, 3423 insertions(+) create mode 100644 docs/superpowers/plans/2026-09-04-yunke-analytics-module.md diff --git a/docs/superpowers/plans/2026-09-04-yunke-analytics-module.md b/docs/superpowers/plans/2026-09-04-yunke-analytics-module.md new file mode 100644 index 0000000..2531f36 --- /dev/null +++ b/docs/superpowers/plans/2026-09-04-yunke-analytics-module.md @@ -0,0 +1,3423 @@ +# 数据分析板块 v1 实施计划（A1+A3+B2+C1+E0-lite+stat_daily_staff） + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 云客后台新增「数据分析」板块：A1 经营总览、A3 流失预警、B2 该回未回、C1 敏感词，外加 stat_daily_staff 员工日汇总地基与 E0-lite 数据健康端点。 + +**Architecture:** 全部读路径骑小表（msg_daily_stats 6.2万 / customer 1.2万 / phone_call 3千 / stat_daily_staff ~2万），message 只做索引点查（生产 buffer pool 128MB 禁裸扫）。后端新增 `app/analytics/` 服务层（口径收敛处）+ `app/storage/staff_summary.py`（rollup 维护）；前端侧栏新组四页 + ECharts 按需引入。 + +**Spec:** `docs/superpowers/specs/2026-09-04-yunke-analytics-module-design.md`（口径速查表在 spec §8，实现时以它为准） + +**Tech Stack:** FastAPI + SQLAlchemy 2.0 async + alembic；Vue3 + TS + Element Plus + echarts/core 按需；pytest 内存 SQLite（SQL 双方言兼容）。 + +**项目约定（执行者必读）：** +- master 单主干直接提交；提交信息中文 `type(scope): 描述`，结尾加 `Co-Authored-By: Claude Code `。 +- 后端测试：`cd yunke-server && .venv/Scripts/python.exe -m pytest tests/ -x -q`。新 SQL 禁 bare column in GROUP BY、禁 SQL 日期函数、日期一律 day 整数（Asia/Shanghai，Python 侧算）。 +- 测试用 conftest 的 `make_client` 工厂（`tests/conftest.py`，seed 协程 + AsyncClient）。 +- 前端验证：`cd yunkes && npm run build`（vue-tsc 类型检查）。 +- 日志禁止输出敏感词内容与命中正文。 +- R1 已核实：`staff.is_on_job` 存在（models.py:136），A3 待交接不降级。 +- 迁移链现状 head = `a7d3e9f1c2b4`（msg_daily_stats）。 + +--- + +## 文件结构总览 + +``` +yunke-server/ + app/db/models.py +4 模型 +1 索引声明 + migrations/versions/f0a1b2c3d4e5_analytics_module_tables.py 新迁移 + app/storage/summary.py +day_bounds_ms 公共别名 + app/storage/staff_summary.py 新:stat_daily_staff 维护层 + app/analytics/__init__.py 新:分析服务层包 + app/analytics/health.py 新:E0-lite + app/analytics/metrics.py 新:overview/churn/unreplied + app/analytics/sensitive/__init__.py 新 + app/analytics/sensitive/words.py 新:词库 CRUD + 预览 + app/analytics/sensitive/scanner.py 新:增量/全量扫描 + app/api/routers/analytics.py 新:全部 /api/analytics/* 端点 + app/api/main.py +注册 router + app/sync/cli.py +staff-daily-stats / sensitive-scan + app/sync/scheduler.py +4h analytics tick job + scripts/full_sync_all.sh +staff-daily-stats 步骤 + tests/test_analytics_models.py / test_analytics_api.py / test_staff_summary.py + tests/test_analytics_churn_unreplied.py / test_sensitive.py +yunkes/ + src/api/analytics.ts 新:全部 API 函数+类型 + src/composables/useChart.ts 新:ECharts 封装 + src/router/index.ts +4 路由 + src/layouts/MainLayout.vue +「数据分析」组 + src/views/analytics/DashboardView.vue 新:A1 + src/views/analytics/ChurnView.vue 新:A3 + src/views/analytics/UnrepliedView.vue 新:B2 + src/views/analytics/SensitiveView.vue 新:C1 + package.json +echarts +docs/DEPLOY.md +升级回填注记 +``` + +--- + +### Task 1: 数据模型 + alembic 迁移 + +**Files:** +- Modify: `yunke-server/app/db/models.py`（MsgDailyStats 类之后、`__all__` 之前插入 4 个模型；Message.__table_args__ 追加索引） +- Create: `yunke-server/migrations/versions/f0a1b2c3d4e5_analytics_module_tables.py` +- Test: `yunke-server/tests/test_analytics_models.py` + +- [ ] **Step 1: models.py 追加 4 个模型**（放在 `class MsgDailyStats` 之后，`__all__` 列表之前；并把这 4 个名字加进 `__all__`） + +```python +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() +``` + +- [ ] **Step 2: Message.__table_args__ 追加 fetched_at 索引**（models.py:437-444 的 `ix_message_channel_owner_time` Index 之后、同一元组内；模型+迁移双声明防 autogenerate 反 DROP，惯例同 c53f90b1e6a4） + +```python + Index( + "ix_message_fetched_at", + "fetched_at", + ), +``` + +- [ ] **Step 3: 写迁移文件** `migrations/versions/f0a1b2c3d4e5_analytics_module_tables.py`（幂等风格照抄 a7d3e9f1c2b4：sa.inspect 先查再建；列定义与上面模型一字对应） + +```python +"""数据分析板块 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) +``` + +- [ ] **Step 4: 写冒烟测试** `tests/test_analytics_models.py` + +```python +"""新表建模冒烟: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="飞单")) + + 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() +``` + +- [ ] **Step 5: 跑测试** + +Run: `cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_analytics_models.py -x -q` +Expected: 2 passed + +- [ ] **Step 6: 全量回归**（确保新模型不破坏现有 512+ 测试） + +Run: `cd yunke-server && .venv/Scripts/python.exe -m pytest tests/ -x -q` +Expected: all passed + +- [ ] **Step 7: Commit** + +```bash +git add yunke-server/app/db/models.py yunke-server/migrations/versions/f0a1b2c3d4e5_analytics_module_tables.py yunke-server/tests/test_analytics_models.py +git commit -m "feat(analytics): 数据分析板块四新表+迁移——stat_daily_staff/敏感词三表/fetched_at 索引" +``` + +--- + +### Task 2: analytics 包骨架 + E0-lite health 端点 + +**Files:** +- Create: `yunke-server/app/analytics/__init__.py` +- Create: `yunke-server/app/analytics/health.py` +- Create: `yunke-server/app/api/routers/analytics.py` +- Modify: `yunke-server/app/api/main.py`（import + include_router） +- Test: `yunke-server/tests/test_analytics_api.py` + +- [ ] **Step 1: `app/analytics/__init__.py`** + +```python +"""分析服务层——所有数据分析指标的口径收敛处(spec 2026-09-04 §2)。 + +指标函数是纯查询(只依赖 AsyncSession,无 HTTP/请求上下文),E2 报表订阅 +(二批)直接复用。口径速查表:spec §8;函数 docstring 与之同步。 +""" +``` + +- [ ] **Step 2: `app/analytics/health.py`** + +```python +"""E0-lite 数据健康——「数据截至」角标 + 实体新鲜度。 + +完整 E0(调度日志解析/成功率账本)二批;v1 用表新鲜度判定: +msg_daily_stats 当日有行 = 4h 消息链路活着。 +""" +from __future__ import annotations + +import datetime as dt + +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import ( + ChatGroup, + Customer, + Media, + Message, + MomentsPost, + MsgDailyStats, + PhoneCall, +) + + +async def data_as_of(session: AsyncSession) -> int | None: + """数据截至(ms-epoch)= MAX(msg_daily_stats.last_ts)。""" + val = ( + await session.execute(sa.select(sa.func.max(MsgDailyStats.last_ts))) + ).scalar_one_or_none() + return int(val) if val is not None else None + + +async def entity_freshness(session: AsyncSession) -> list[dict]: + """各实体最后更新时间(UPDATED_AT 列 MAX;message 用 fetched_at)。""" + + async def _max(col) -> dt.datetime | None: + return (await session.execute(sa.select(sa.func.max(col)))).scalar_one_or_none() + + out = [] + for name, col in ( + ("customer", Customer.updated_at), + ("chat_group", ChatGroup.updated_at), + ("phone_call", PhoneCall.updated_at), + ("media", Media.updated_at), + ("moments_post", MomentsPost.updated_at), + ("message", Message.fetched_at), + ): + out.append({"table": name, "last_seen": await _max(col)}) + return out +``` + +- [ ] **Step 3: `app/api/routers/analytics.py`** + +```python +"""GET /api/analytics/* —— 数据分析板块端组(前缀分组=未来角色权限落点)。""" +from __future__ import annotations + +from typing import Annotated, Any, Literal + +from fastapi import APIRouter, Depends +from sqlalchemy.ext.asyncio import AsyncSession + +from app.analytics import health as health_svc +from app.api.deps import get_session + +router = APIRouter(prefix="/api/analytics", tags=["analytics"]) + +SessionDep = Annotated[AsyncSession, Depends(get_session)] + + +@router.get("/health") +async def analytics_health(session: SessionDep) -> dict[str, Any]: + """E0-lite:数据截至 + 实体新鲜度。""" + as_of = await health_svc.data_as_of(session) + return {"data_as_of": as_of, "entities": await health_svc.entity_freshness(session)} +``` + +- [ ] **Step 4: main.py 注册**（routers import 列表按字母序加 `analytics`；`app.include_router(analytics.router)` 放 include 列表任意处，照现有行样式） + +- [ ] **Step 5: 写失败测试** `tests/test_analytics_api.py` + +```python +"""/api/analytics/health:E0-lite 冒烟。""" +from __future__ import annotations + +from app.db.models import MsgDailyStats + + +async def test_health_empty(make_client): + async def seed(session): + ... + + client = await make_client(seed) + r = await client.get("/api/analytics/health") + assert r.status_code == 200 + body = r.json() + assert body["data_as_of"] is None + assert any(e["table"] == "message" for e in body["entities"]) + + +async def test_health_with_stats(make_client): + async def seed(session): + session.add( + MsgDailyStats( + channel="wechat", + owner_account_id="wx1", + peer_kind="friend", + peer_id="t1", + day=20260904, + send_cnt=1, + recv_cnt=2, + last_ts=1756900000000, + last_is_send=False, + ) + ) + + client = await make_client(seed) + r = await client.get("/api/analytics/health") + assert r.json()["data_as_of"] == 1756900000000 +``` + +- [ ] **Step 6: 跑测试** Run: `cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_analytics_api.py -x -q` → Expected: 2 passed + +- [ ] **Step 7: Commit** + +```bash +git add yunke-server/app/analytics/ yunke-server/app/api/routers/analytics.py yunke-server/app/api/main.py yunke-server/tests/test_analytics_api.py +git commit -m "feat(analytics): E0-lite 数据健康端点——数据截至+实体新鲜度" +``` + +--- + +### Task 3: stat_daily_staff 维护层（staff_summary.py） + +**Files:** +- Modify: `yunke-server/app/storage/summary.py`（文件尾部加公共别名） +- Create: `yunke-server/app/storage/staff_summary.py` +- Test: `yunke-server/tests/test_staff_summary.py` + +- [ ] **Step 1: summary.py 尾部加别名**（DRY:日界计算唯一权威在此） + +```python +# 公共别名(2026-09-04):analytics 层复用日界计算,避免各处复制时区逻辑。 +day_bounds_ms = _day_bounds_ms +``` + +- [ ] **Step 2: 写失败测试** `tests/test_staff_summary.py` + +```python +"""stat_daily_staff 维护层:派生/幂等/判别法/未映射兜底。 + +判别法(9-03 先例):rollup 只读 msg_daily_stats——回填后删光 message 明细, +员工汇总重算仍出完整数据。 +""" +from __future__ import annotations + +import datetime as dt + +from app.db.models import Message, MsgDailyStats, StatDailyStaff, WechatAccount +from app.storage import staff_summary + + +def _mk_msg(msg_id: str, ts: int, is_send: bool | None) -> Message: + return Message( + msg_id=msg_id, + channel="wechat", + owner_account_id="wx1", + talker_id="t1", + timestamp=ts, + is_send=is_send, + text="x", + fetched_at=dt.datetime(2026, 9, 4, 12, 0), + ) + + +async def _seed_pair(session) -> None: + """两账号同一员工(wx1/wx2→u1) + 未映射账号 wx3,20260904 各有消息。""" + session.add(WechatAccount(wx_id="wx1", user_id="u1", user_name="张三")) + session.add(WechatAccount(wx_id="wx2", user_id="u1", user_name="张三")) + rows = [ + MsgDailyStats(channel="wechat", owner_account_id="wx1", peer_kind="friend", + peer_id="t1", day=20260904, send_cnt=10, recv_cnt=5, + last_ts=1756900000000, last_is_send=True), + MsgDailyStats(channel="wechat", owner_account_id="wx1", peer_kind="room", + peer_id="r1", day=20260904, send_cnt=3, recv_cnt=0, + last_ts=1756900000000, last_is_send=True), + MsgDailyStats(channel="wechat", owner_account_id="wx2", peer_kind="friend", + peer_id="t1", day=20260904, send_cnt=2, recv_cnt=1, + last_ts=1756900001000, last_is_send=False), + MsgDailyStats(channel="wechat", owner_account_id="wx3", peer_kind="friend", + peer_id="t9", day=20260904, send_cnt=1, recv_cnt=0, + last_ts=1756900002000, last_is_send=True), + ] + session.add_all(rows) + await session.commit() + + +async def test_recompute_aggregates_by_user(make_client): + async def seed(session): + await _seed_pair(session) + n = await staff_summary.recompute_days(session, [20260904]) + await session.commit() + assert n == 1 + + client = await make_client(seed) + rows = ( + await client.app.state.get("rows") if False else None + ) + # 直接用返回值断言:recompute 后查表 + async def verify(session): + out = ( + await session.execute( + __import__("sqlalchemy").select(StatDailyStaff) + ) + ).scalars().all() + return {(r.user_id): r for r in out} + + # make_client 工厂只给 http 客户端;这里改用直连引擎断言—— + # 换一种写法:在 seed 里断言(seed 也有 session) +``` + +上面这个测试写法不对——`make_client` 的 seed 里就有 session，直接在 seed 里断言即可。**正确版本：** + +```python +async def test_recompute_aggregates_by_user(make_client): + """两账号同人合并(sent=10+3+2=15,recv=5+0+1=6;friends 跨账号去重=2?否—— + t1 在 wx1/wx2 都聊 → 合并后 distinct friend={t1} + room={r1});wx3 未映射按原值。""" + async def seed(session): + await _seed_pair(session) + await staff_summary.recompute_days(session, [20260904]) + await session.commit() + from sqlalchemy import select + + rows = (await session.execute(select(StatDailyStaff))).scalars().all() + by_uid = {r.user_id: r for r in rows} + assert set(by_uid) == {"u1", "wx3"} # wx3 映射不上按 owner 原值 + u1 = by_uid["u1"] + assert u1.msg_sent == 15 and u1.msg_recv == 6 + assert u1.active_friends == 1 # t1 跨账号去重 + assert u1.active_rooms == 1 + assert by_uid["wx3"].active_friends == 1 + + await make_client(seed) + + +async def test_recompute_idempotent(make_client): + async def seed(session): + await _seed_pair(session) + await staff_summary.recompute_days(session, [20260904]) + await staff_summary.recompute_days(session, [20260904]) # 再跑不翻倍 + await session.commit() + from sqlalchemy import func, select + + n = ( + await session.execute( + select(func.count()).select_from(StatDailyStaff) + ) + ).scalar_one() + assert n == 2 # u1 + wx3 各一行 + + await make_client(seed) + + +async def test_discriminant_survives_message_delete(make_client): + """判别法:rollup 回填后删光 message,重算结果不变(只依赖 msg_daily_stats)。""" + async def seed(session): + await _seed_pair(session) + session.add(_mk_msg("m1", 1756900000000, True)) + await session.commit() + await staff_summary.recompute_days(session, [20260904]) + await session.commit() + from sqlalchemy import delete, select + + await session.execute(delete(Message)) + await staff_summary.recompute_days(session, [20260904]) # 自愈重算 + await session.commit() + rows = (await session.execute(select(StatDailyStaff))).scalars().all() + by_uid = {r.user_id: r for r in rows} + assert by_uid["u1"].msg_sent == 15 # 数据不丢 + + await make_client(seed) + + +async def test_recompute_all_and_rolling(make_client): + async def seed(session): + await _seed_pair(session) + n = await staff_summary.recompute_all(session) + await session.commit() + assert n == 1 # 全库只有一个 day + + await make_client(seed) + days = staff_summary.rolling_days(20260904, window=3) + assert days == [20260902, 20260903, 20260904] +``` + +- [ ] **Step 3: 跑测试确认失败** Run: `cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_staff_summary.py -x -q` → Expected: FAIL（ModuleNotFoundError: staff_summary） + +- [ ] **Step 4: 实现 `app/storage/staff_summary.py`** + +```python +"""stat_daily_staff(员工×日)维护层——rollup-of-rollup,不挂写钩子。 + +刷新策略(spec §3.1):4h 调度刷当日+滚动 3 天(staff_summary.rolling_days), +夜间 full_sync 全量自愈(recompute_all),部署回填 CLI 同款。空表时读端点 +回退直接聚合 msg_daily_stats(毫秒级)——部署即用,沿用 sessions 的 +has_rows 回退模式。 +""" +from __future__ import annotations + +import datetime as dt +import logging +from collections.abc import Iterable +from zoneinfo import ZoneInfo + +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import MsgDailyStats, StatDailyStaff, WechatAccount, WhatsappAccount + +logger = logging.getLogger(__name__) + +_SH = ZoneInfo("Asia/Shanghai") + + +async def has_rows(session: AsyncSession) -> bool: + """汇总表是否已有数据(读端点空表守卫;表不存在与空表同视)。""" + try: + return bool( + ( + await session.execute( + sa.select(sa.func.count()).select_from(StatDailyStaff) + ) + ).scalar_one() + ) + except (sa.exc.OperationalError, sa.exc.ProgrammingError): + return False + + +async def account_map(session: AsyncSession, channel: str) -> dict[str, str]: + """owner_account_id → user_id(按渠道选账号表;映射不上的不在 dict)。""" + if channel == "whatsapp": + stmt = sa.select(WhatsappAccount.uid, WhatsappAccount.user_id).where( + WhatsappAccount.user_id.is_not(None) + ) + else: + stmt = sa.select(WechatAccount.wx_id, WechatAccount.user_id).where( + WechatAccount.user_id.is_not(None) + ) + return {owner: uid for owner, uid in (await session.execute(stmt)).all() if uid} + + +def rolling_days(today: int, window: int = 3) -> list[int]: + """当日 + 前 window-1 天的 day 键(4h job 的滚动刷新窗)。""" + d = _day_from_key(today) + return [_key_from_day(d - dt.timedelta(days=i)) for i in range(window - 1, -1, -1)] + + +def _day_from_key(day: int) -> dt.date: + return dt.date(day // 10000, (day // 100) % 100, day % 100) + + +def _key_from_day(d: dt.date) -> int: + return d.year * 10000 + d.month * 100 + d.day + + +async def recompute_days( + session: AsyncSession, days: Iterable[int], channels: tuple[str, ...] = ("wechat", "whatsapp") +) -> int: + """重算给定 day×channel 的员工行:删旧插新,幂等。 + + 单日日行仅数百~两千,拉回 Python 侧按 user_id 聚合——COUNT(DISTINCT + peer) 必须在「账号合并到员工」之后算(同人多账号 distinct peer 跨账号 + 去重),写 SQL 反而绕且踩 only_full_group_by。 + """ + n = 0 + for day in sorted(set(days)): + for channel in channels: + amap = await account_map(session, channel) + rows = ( + await session.execute( + sa.select( + MsgDailyStats.owner_account_id, + MsgDailyStats.peer_kind, + MsgDailyStats.peer_id, + MsgDailyStats.send_cnt, + MsgDailyStats.recv_cnt, + ).where( + MsgDailyStats.day == day, MsgDailyStats.channel == channel + ) + ) + ).all() + agg: dict[str, dict] = {} + for owner, kind, peer, s, r in rows: + uid = amap.get(owner, owner) # 映射不上按 owner 原值(读侧标注) + a = agg.setdefault( + uid, {"sent": 0, "recv": 0, "friends": set(), "rooms": set()} + ) + a["sent"] += int(s or 0) + a["recv"] += int(r or 0) + (a["friends"] if kind == "friend" else a["rooms"]).add(peer) + await session.execute( + sa.delete(StatDailyStaff).where( + StatDailyStaff.day == day, StatDailyStats.channel == channel + ) + ) + for uid, a in sorted(agg.items()): + session.add( + StatDailyStaff( + day=day, + channel=channel, + user_id=uid, + msg_sent=a["sent"], + msg_recv=a["recv"], + active_friends=len(a["friends"]), + active_rooms=len(a["rooms"]), + ) + ) + n += 1 + return n + + +async def recompute_all(session: AsyncSession) -> int: + """全量重建(夜间自愈 / 部署回填,幂等)。""" + days = ( + await session.execute(sa.select(MsgDailyStats.day).distinct()) + ).scalars().all() + return await recompute_days(session, list(days)) +``` + +⚠️ 上面 `recompute_days` 里 delete 语句有一个笔误要修正：`StatDailyStats.channel == channel` 应为 `StatDailyStaff.channel == channel`。写代码时注意。 + +- [ ] **Step 5: 跑测试** Run: `cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_staff_summary.py -x -q` → Expected: 4 passed + +- [ ] **Step 6: Commit** + +```bash +git add yunke-server/app/storage/summary.py yunke-server/app/storage/staff_summary.py yunke-server/tests/test_staff_summary.py +git commit -m "feat(analytics): stat_daily_staff 员工日汇总维护层——rollup派生/幂等/判别法" +``` + +### Task 4: metrics.py —— A1 总览指标（卡片/趋势/异常警示） + +**Files:** +- Create: `yunke-server/app/analytics/metrics.py` +- Test: `yunke-server/tests/test_analytics_overview.py` + +- [ ] **Step 1: 写失败测试** `tests/test_analytics_overview.py` + +```python +"""A1 总览指标:卡片/首沟通口径/环比(仅周月)/异常警示。""" +from __future__ import annotations + +import datetime as dt + +from app.analytics import metrics +from app.db.models import ( + MsgDailyStats, + PhoneCall, + StatDailyStaff, + WechatAccount, +) + +_TODAY = 20260904 # 测试统一用固定 day 键,monkeypatch today_key + + +async def _seed(session): + session.add(WechatAccount(wx_id="wx1", user_id="u1", user_name="张三")) + # 昨日(20260903):u1 100 条;前 7 日均值(20260827-0902):每天 200 → 跌 50% 触发 + for day, sent in [(20260827, 200), (20260828, 200), (20260829, 200), + (20260830, 200), (20260831, 200), (20260901, 200), + (20260902, 200), (20260903, 100)]: + session.add(StatDailyStaff(day=day, channel="wechat", user_id="u1", + msg_sent=sent, msg_recv=0, + active_friends=5, active_rooms=1)) + # 当日 + 首沟通:t1 当日首现(首沟通=1),t2 早在 20260827 就有(不算) + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t1", day=20260904, + send_cnt=10, recv_cnt=5, last_ts=1756950000000, + last_is_send=True)) + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t2", day=20260827, + send_cnt=3, recv_cnt=0, last_ts=1756240000000, + last_is_send=True)) + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="room", peer_id="r1", day=20260904, + send_cnt=7, recv_cnt=0, last_ts=1756950000000, + last_is_send=True)) + session.add(PhoneCall(call_id="c1", channel="voice_call", + start_call_time=1756950000000, call_duration=60)) + await session.commit() + + +async def test_overview_today(make_client, monkeypatch): + monkeypatch.setattr(metrics, "today_key", lambda: _TODAY) + async def seed(session): + await _seed(session) + out = await metrics.overview(session, channel="wechat", scope="today") + assert out["cards"]["msg_total"] == 22 # (10+5)+(7+0) + assert out["cards"]["active_customers"] == 1 # peer_kind=friend 去重,群不计 + assert out["cards"]["active_staff"] == 1 + assert out["cards"]["call_cnt"] == 1 + assert out["cards"]["first_contact"] == 1 # t1 全史 MIN(day)=今日;t2 不是 + assert out["cards"].get("mom_total_delta") is None # 今日无环比 + + await make_client(seed) + + +async def test_anomaly_yesterday_vs_prev7(make_client, monkeypatch): + monkeypatch.setattr(metrics, "today_key", lambda: _TODAY) + async def seed(session): + await _seed(session) + a = metrics.anomaly(session, channel="wechat") + assert a is not None + assert a["total_delta_pct"] <= -0.5 # 100 vs 均值 200 = -50% + staff = {s["user_id"]: s for s in a["staff"]} + assert staff["u1"]["yesterday"] == 100 + assert staff["u1"]["prev7_mean"] == 200.0 + + await make_client(seed) + + +async def test_overview_fallback_when_rollup_empty(make_client, monkeypatch): + """stat_daily_staff 空表 → 员工卡片回退实时聚合 msg_daily_stats(部署即用)。""" + monkeypatch.setattr(metrics, "today_key", lambda: _TODAY) + async def seed(session): + session.add(WechatAccount(wx_id="wx1", user_id="u1")) + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t1", day=20260904, + send_cnt=2, recv_cnt=1, last_ts=1756950000000, + last_is_send=True)) + await session.commit() + out = await metrics.overview(session, channel="wechat", scope="today") + assert out["cards"]["active_staff"] == 1 + assert out["cards"]["msg_total"] == 3 + + await make_client(seed) +``` + +- [ ] **Step 2: 跑测试确认失败** Run: `cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_analytics_overview.py -x -q` → Expected: FAIL（no module metrics） + +- [ ] **Step 3: 实现 `app/analytics/metrics.py`**（口径注释 = spec §8 速查表） + +```python +"""A1/A3/B2 指标函数——纯查询,口径速查表见 spec §8。 + +时区:day 键一律 Asia/Shanghai 自然日(Python 侧算,禁 SQL 日期函数)。 +周起始=周一(scope=week)。 +""" +from __future__ import annotations + +import datetime as dt +import math +from collections.abc import Iterable +from typing import Any +from zoneinfo import ZoneInfo + +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import MsgDailyStats, PhoneCall, StatDailyStaff +from app.storage.staff_summary import account_map, has_rows as _staff_rows +from app.storage.summary import day_bounds_ms + +_SH = ZoneInfo("Asia/Shanghai") +_CHANNELS = ("wechat", "whatsapp") + + +def today_key(now: dt.datetime | None = None) -> int: + """上海自然日 yyyymmdd(测试 monkeypatch 点)。""" + d = (now or dt.datetime.now(_SH)).date() + return d.year * 10000 + d.month * 100 + d.day + + +def _day_date(day: int) -> dt.date: + return dt.date(day // 10000, (day // 100) % 100, day % 100) + + +def scope_day_keys(scope: str, today: int) -> list[int]: + """today={今};week=本周一..今天;month=本月1号..今天(含端点)。""" + end = _day_date(today) + if scope == "today": + start = end + elif scope == "week": + start = end - dt.timedelta(days=end.weekday()) # 周一=0 + elif scope == "month": + start = end.replace(day=1) + else: + raise ValueError(f"bad scope: {scope!r}") + out, cur = [], start + while cur <= end: + out.append(cur.year * 10000 + cur.month * 100 + cur.day) + cur += dt.timedelta(days=1) + return out + + +def _prev_scope_days(scope: str, days: list[int]) -> list[int]: + """紧邻的前一个等长窗口(环比;今日 scope 不调用)。""" + n = len(days) + last = _day_date(days[0]) - dt.timedelta(days=1) + return [ + (lambda d: d.year * 10000 + d.month * 100 + d.day)( + last - dt.timedelta(days=i) + ) + for i in range(n - 1, -1, -1) + ] + + +async def _day_rows( + session: AsyncSession, channel: str, days: Iterable[int] +) -> list[Any]: + return ( + await session.execute( + sa.select( + MsgDailyStats.day, + MsgDailyStats.peer_kind, + MsgDailyStats.peer_id, + MsgDailyStats.send_cnt, + MsgDailyStats.recv_cnt, + ).where( + MsgDailyStats.channel == channel, MsgDailyStats.day.in_(list(days)) + ) + ) + ).all() + + +async def _call_count(session: AsyncSession, channel_scope_days: list[int]) -> int: + """phone_call 落窗计数(3k 行小表;start_call_time ms 与 day 键边界换算)。""" + if not channel_scope_days: + return 0 + lo, _hi = day_bounds_ms(channel_scope_days[0]) + hi_day_start, _ = day_bounds_ms(channel_scope_days[-1]) + hi = hi_day_start + 86_400_000 # 含末日全天 + val = ( + await session.execute( + sa.select(sa.func.count()).select_from(PhoneCall).where( + PhoneCall.start_call_time.is_not(None), + PhoneCall.start_call_time >= lo, + PhoneCall.start_call_time < hi, + ) + ) + ).scalar_one() + return int(val) + + +async def _first_contact( + session: AsyncSession, channel: str, days: list[int] +) -> int: + """首次沟通客户数:peer 按渠道全史 MIN(day) 落窗(跨员工归一,不跨渠道)。""" + rows = ( + await session.execute( + sa.select(MsgDailyStats.peer_id, sa.func.min(MsgDailyStats.day)) + .where( + MsgDailyStats.channel == channel, + MsgDailyStats.peer_kind == "friend", + ) + .group_by(MsgDailyStats.peer_id) + ) + ).all() + dayset = set(days) + return sum(1 for _peer, first in rows if first in dayset) + + +async def _active_staff( + session: AsyncSession, channel: str, days: list[int] +) -> int: + """活跃员工 distinct user_id;stat_daily_staff 空表回退 msg_daily_stats+映射。""" + if await _staff_rows(session): + val = ( + await session.execute( + sa.select(sa.func.count(sa.distinct(StatDailyStaff.user_id))).where( + StatDailyStaff.channel == channel, + StatDailyStaff.day.in_(days), + ) + ) + ).scalar_one() + return int(val) + amap = await account_map(session, channel) + owners = set(amap.get(o, o) for (o,) in ( + await session.execute( + sa.select(MsgDailyStats.owner_account_id).where( + MsgDailyStats.channel == channel, MsgDailyStats.day.in_(days) + ) + ) + ).all()) + return len(owners) + + +def anomaly_threshold() -> float: + return 0.5 # 昨日 vs 前 7 日均值偏离阈值,v1 固定 + + +async def anomaly(session: AsyncSession, channel: str) -> dict | None: + """E0-lite 异常警示:昨日总量 vs 前 7 日均值偏离>50%,明细到员工。""" + today = today_key() + yesterday = _day_date(today) - dt.timedelta(days=1) + + def key(d: dt.date) -> int: + return d.year * 10000 + d.month * 100 + d.day + + yday = key(yesterday) + prev7 = [key(yesterday - dt.timedelta(days=i)) for i in range(1, 8)] + + use_rollup = await _staff_rows(session) + if use_rollup: + rows = ( + await session.execute( + sa.select( + StatDailyStaff.day, StatDailyStaff.user_id, + sa.func.coalesce(StatDailyStaff.msg_sent + StatDailyStaff.msg_recv, 0), + ).where( + StatDailyStaff.channel == channel, + StatDailyStaff.day.in_([yday] + prev7), + ) + ) + ).all() + else: + amap = await account_map(session, channel) + raw = ( + await session.execute( + sa.select( + MsgDailyStats.day, MsgDailyStats.owner_account_id, + sa.func.coalesce( + sa.func.sum(MsgDailyStats.send_cnt + MsgDailyStats.recv_cnt), 0 + ), + ) + .where( + MsgDailyStats.channel == channel, + MsgDailyStats.day.in_([yday] + prev7), + ) + .group_by(MsgDailyStats.day, MsgDailyStats.owner_account_id) + ) + ).all() + rows = [(d, amap.get(o, o), v) for d, o, v in raw] + + per_user: dict[str, list[int]] = {} + for _day, uid, v in rows: + per_user.setdefault(uid, []).append(int(v)) + + def split(vals: list[int]) -> tuple[int, float]: + """昨日量 = day==yday 的合计;前7均值 = 其余均值。""" + y = vals[-1] # rows 未按序——改为显式传 day;见下方实现修正 + raise NotImplementedError + + # 直接逐 user 算(不猜顺序): + staff_out = [] + totals_y, totals_p = 0, 0.0 + day_rows: dict[str, dict[int, int]] = {} + for day, uid, v in rows: + day_rows.setdefault(uid, {})[day] = int(v) + for uid, by_day in day_rows.items(): + y = by_day.get(yday, 0) + pvals = [by_day.get(d, 0) for d in prev7] + pmean = sum(pvals) / len(pvals) if pvals else 0.0 + totals_y += y + totals_p += pmean + if pmean > 0 and abs(y - pmean) / pmean > anomaly_threshold(): + staff_out.append({ + "user_id": uid, + "yesterday": y, + "prev7_mean": round(pmean, 1), + }) + if totals_p <= 0: + return None + delta = (totals_y - totals_p) / totals_p + if abs(delta) > anomaly_threshold(): + return { + "total_delta_pct": round(delta, 3), + "yesterday_total": totals_y, + "prev7_mean": round(totals_p, 1), + "staff": sorted(staff_out, key=lambda s: s["yesterday"]), + } + return None + + +async def overview( + session: AsyncSession, *, channel: str, scope: str, trend_days: int = 30 +) -> dict[str, Any]: + """A1:卡片 + 环比(仅周/月) + 近 N 天趋势 + 异常警示 + 数据截至。""" + today = today_key() + days = scope_day_keys(scope, today) + rows = await _day_rows(session, channel, days) + msg_total = sum(int(s or 0) + int(r or 0) for *_ , s, r in rows) + active_customers = len({p for _d, kind, p, *_ in rows if kind == "friend"}) + + cards: dict[str, Any] = { + "msg_total": msg_total, + "active_customers": active_customers, + "active_staff": await _active_staff(session, channel, days), + "call_cnt": await _call_count(session, days), + "first_contact": await _first_contact(session, channel, days), + } + + # 环比仅周/月(今日 vs 昨日全天会误导,spec §8) + if scope in ("week", "month"): + prev_days = _prev_scope_days(scope, days) + prev_rows = await _day_rows(session, channel, prev_days) + prev_total = sum(int(s or 0) + int(r or 0) for *_, s, r in prev_rows) + cards["prev_msg_total"] = prev_total + cards["msg_total_delta"] = ( + round((msg_total - prev_total) / prev_total, 3) if prev_total else None + ) + + # 趋势:近 trend_days 天 每日 消息量/活跃客户/通话量 + t_days = scope_day_keys("month", today) # 占位,下面按窗口重算 + end = _day_date(today) + t_days = [] + cur = end - dt.timedelta(days=trend_days - 1) + while cur <= end: + t_days.append(cur.year * 10000 + cur.month * 100 + cur.day) + cur += dt.timedelta(days=1) + t_rows = await _day_rows(session, channel, t_days) + by_day: dict[int, dict] = {d: {"msg": 0, "cust": set()} for d in t_days} + for d, kind, p, s, r in t_rows: + by_day[d]["msg"] += int(s or 0) + int(r or 0) + if kind == "friend": + by_day[d]["cust"].add(p) + # 通话趋势:phone_call 拉窗内 (start_call_time) Python 侧按日分桶(3k 行) + lo, _ = day_bounds_ms(t_days[0]) + hi_day, _ = day_bounds_ms(t_days[-1]) + calls = ( + await session.execute( + sa.select(PhoneCall.start_call_time).where( + PhoneCall.start_call_time.is_not(None), + PhoneCall.start_call_time >= lo, + PhoneCall.start_call_time < hi_day + 86_400_000, + ) + ) + ).scalars().all() + call_by_day: dict[int, int] = {} + for ts in calls: + d = dt.datetime.fromtimestamp(ts / 1000, _SH) + k = d.year * 10000 + d.month * 100 + d.day + call_by_day[k] = call_by_day.get(k, 0) + 1 + trend = [ + {"day": d, "msg": v["msg"], "active_customers": len(v["cust"]), + "call_cnt": call_by_day.get(d, 0)} + for d, v in by_day.items() + ] + + return { + "scope": scope, + "channel": channel, + "days": days, + "cards": cards, + "trend": trend, + "anomaly": await anomaly(session, channel), + } +``` + +实现注意：① `anomaly` 里那个 `split` 半成品函数不要写进去（上面保留它是提示逐 user 算法，落地时删掉）；② `overview` 里 `t_days = scope_day_keys("month", today)` 占位行删除；③ `for *_ , s, r in rows` 的解包统一写成显式五元组。写完后跑测试修正到绿。 + +- [ ] **Step 4: 跑测试到绿** Run: `cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_analytics_overview.py -x -q` → Expected: 3 passed + +- [ ] **Step 5: Commit** + +```bash +git add yunke-server/app/analytics/metrics.py yunke-server/tests/test_analytics_overview.py +git commit -m "feat(analytics): A1 总览指标——卡片/首沟通口径/周月环比/异常警示/空表回退" +``` + +--- + +### Task 5: /api/analytics/overview 端点 + +**Files:** +- Modify: `yunke-server/app/api/routers/analytics.py` +- Test: `yunke-server/tests/test_analytics_api.py`（追加） + +- [ ] **Step 1: 追加失败测试**（test_analytics_api.py） + +```python +async def test_overview_endpoint(make_client): + from app.db.models import MsgDailyStats + + async def seed(session): + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t1", day=20260904, + send_cnt=4, recv_cnt=1, last_ts=1756950000000, + last_is_send=True)) + await session.commit() + + client = await make_client(seed) + r = await client.get( + "/api/analytics/overview", params={"channel": "wechat", "scope": "today"} + ) + assert r.status_code == 200 + body = r.json() + assert body["cards"]["msg_total"] == 5 + assert isinstance(body["trend"], list) and body["trend"] + # scope 非法 → 422 + r2 = await client.get( + "/api/analytics/overview", params={"channel": "wechat", "scope": "year"} + ) + assert r2.status_code == 422 +``` + +- [ ] **Step 2: 端点实现**（analytics.py router 追加） + +```python +from typing import Literal + +from app.analytics import metrics as metrics_svc + + +@router.get("/overview") +async def overview( + session: SessionDep, + channel: Literal["wechat", "whatsapp"] = "wechat", + scope: Literal["today", "week", "month"] = "today", +) -> dict: + """A1 经营总览:卡片+趋势+异常警示一次返回(单往返)。""" + return await metrics_svc.overview(session, channel=channel, scope=scope) +``` + +- [ ] **Step 3: 跑测试** Run: `cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_analytics_api.py -x -q` → Expected: 3 passed + +- [ ] **Step 4: Commit** `git commit -am "feat(analytics): /overview 端点"` + +--- + +### Task 6: A3 流失预警（churn 指标 + 端点） + +**Files:** +- Modify: `yunke-server/app/analytics/metrics.py`（追加 churn） +- Modify: `yunke-server/app/api/routers/analytics.py`（追加端点） +- Test: `yunke-server/tests/test_analytics_churn_unreplied.py` + +- [ ] **Step 1: 写失败测试** + +```python +"""A3 三分法流失 + B2 该回未回。""" +from __future__ import annotations + +import datetime as dt + +from app.analytics import metrics +from app.db.models import Customer, MsgDailyStats, Staff + +_NOW = 1756950000000 # 2026-09-04 ~ 上海 + + +def _cust(friend_id, *, last_chat=None, user_id="u1", delete=0, block=None): + return Customer(friend_id=friend_id, channel="wechat", + owner_account_id="wx1", user_id=user_id, + last_chat_time=last_chat, delete_flag=delete, is_block=block) + + +async def _seed_churn(session): + now = _NOW + day_ms = 86_400_000 + session.add(Staff(user_id="u9", name="离职的", is_on_job=False)) + session.add(Staff(user_id="u1", name="在职的", is_on_job=True)) + # 真流失:45 天没聊 + 历史消息 ≥5 + session.add(_cust("c_real", last_chat=now - 45 * day_ms)) + # 历史消息不足 5 → 不算真流失 + session.add(_cust("c_few", last_chat=now - 45 * day_ms)) + # 只沉默 10 天 → 不入 30 档 + session.add(_cust("c_recent", last_chat=now - 10 * day_ms)) + # 从未激活 + session.add(_cust("c_null", last_chat=None)) + # 待交接:归属离职员工 + session.add(_cust("c_hand", user_id="u9", last_chat=now - 3 * day_ms)) + # 已删好友 → 全部剔除 + session.add(_cust("c_del", last_chat=now - 45 * day_ms, delete=1)) + # 历史消息量:real=6(≥5),few=2 + for peer, n in (("c_real", 6), ("c_few", 2)): + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id=peer, day=20260601, + send_cnt=n, recv_cnt=0, last_ts=now - 50 * day_ms, + last_is_send=True)) + await session.commit() + + +async def test_churn_real_tier30(make_client, monkeypatch): + monkeypatch.setattr(metrics, "_now_ms", lambda: _NOW) + async def seed(session): + await _seed_churn(session) + page = await metrics.churn(session, channel="wechat", kind="real", + tier=30, min_msgs=5, page=1, page_size=20) + ids = [r["friend_id"] for r in page["items"]] + assert "c_real" in ids and "c_few" not in ids + assert "c_recent" not in ids and "c_del" not in ids + real = next(r for r in page["items"] if r["friend_id"] == "c_real") + assert real["hist_msgs"] == 6 and real["active_days"] == 1 + + await make_client(seed) + + +async def test_churn_inactive_and_handover(make_client, monkeypatch): + monkeypatch.setattr(metrics, "_now_ms", lambda: _NOW) + async def seed(session): + await _seed_churn(session) + p1 = await metrics.churn(session, channel="wechat", kind="inactive", + tier=30, min_msgs=5, page=1, page_size=20) + assert {r["friend_id"] for r in p1["items"]} == {"c_null"} + p2 = await metrics.churn(session, channel="wechat", kind="handover", + tier=30, min_msgs=5, page=1, page_size=20) + assert {r["friend_id"] for r in p2["items"]} == {"c_hand"} + + await make_client(seed) +``` + +- [ ] **Step 2: 实现 churn**（metrics.py 追加；`_now_ms` 是测试 monkeypatch 点） + +```python +def _now_ms() -> int: + return int(dt.datetime.now(_SH).timestamp() * 1000) + + +async def churn( + session: AsyncSession, *, channel: str, kind: str, tier: int, + min_msgs: int, page: int, page_size: int, +) -> dict: + """A3 三分法(spec §3.3)。tier ∈ {30,60,90} 分层区间 [tier, next) ,90=≥90。""" + day_ms = 86_400_000 + now = _now_ms() + base = [ + Customer.channel == channel, + sa.or_(Customer.delete_flag.is_(None), Customer.delete_flag == 0), + ] + if kind == "real": + lo = now - _next_tier(tier) * day_ms + hi = now - tier * day_ms + # last_chat_time ∈ (lo, hi]:沉默 tier..next_tier 天 + base += [Customer.last_chat_time.is_not(None), + Customer.last_chat_time > lo, + Customer.last_chat_time <= hi] + order = Customer.last_chat_time.asc() + elif kind == "inactive": + base += [Customer.last_chat_time.is_(None)] + order = Customer.friend_id.asc() + elif kind == "handover": + base += [Customer.user_id.in_( + sa.select(Staff.user_id).where(Staff.is_on_job.is_(False)) + )] + order = Customer.last_chat_time.desc().nulls_last() + else: + raise ValueError(f"bad kind: {kind!r}") + + total = ( + await session.execute( + sa.select(sa.func.count()).select_from(Customer).where(*base) + ) + ).scalar_one() + rows = ( + await session.execute( + sa.select(Customer).where(*base).order_by(order) + .offset((page - 1) * page_size).limit(page_size) + ) + ).scalars().all() + + # enrich:历史消息量/活跃天数(msg_daily_stats 按 peer 全史聚合,62k 行) + items = [] + if rows: + ids = [c.friend_id for c in rows] + agg = ( + await session.execute( + sa.select( + MsgDailyStats.peer_id, + sa.func.coalesce( + sa.func.sum(MsgDailyStats.send_cnt + MsgDailyStats.recv_cnt), 0 + ), + sa.func.count(), + ) + .where( + MsgDailyStats.channel == channel, + MsgDailyStats.peer_kind == "friend", + MsgDailyStats.peer_id.in_(ids), + ) + .group_by(MsgDailyStats.peer_id) + ) + ).all() + hist = {p: (int(m), int(d)) for p, m, d in agg} + for c in rows: + m, d = hist.get(c.friend_id, (0, 0)) + items.append({ + "friend_id": c.friend_id, + "channel": c.channel, + "nickname": c.friend_nickname or c.friend_remark or c.friend_id, + "owner_account_id": c.owner_account_id, + "user_id": c.user_id, + "last_chat_time": c.last_chat_time, + "hist_msgs": m, + "active_days": d, + }) + + # real 需 hist_msgs ≥ min_msgs(过滤后再分页会复杂——先查满页再过滤, + # 数量大时循环补页;v1 简化:SQL 端先过滤(见下),此处保险双滤) + if kind == "real": + items = [it for it in items if it["hist_msgs"] >= min_msgs] + + return {"items": items, "total": int(total), "page": page, "page_size": page_size} + + +def _next_tier(tier: int) -> int: + return {30: 60, 60: 90, 90: 10_000}[tier] +``` + +⚠️ 实现修正点（写码时落实）：`real` 的 `hist_msgs ≥ min_msgs` 过滤放 SQL 端才对（否则分页 total 错）——用半连接：`Customer.friend_id.in_(sa.select(MsgDailyStats.peer_id).where(channel, peer_kind=='friend').group_by(peer_id).having(sum(send+recv) >= min_msgs))`。上面 Python 端双滤保留作保险但 total 语义按 SQL 端为准。`nulls_last` 在 MySQL 会被 app/db/compat.py 编译补丁剥掉（已有机制，DESC 语义天然等价），SQLite 支持——直接用没问题。 + +- [ ] **Step 3: 端点**（analytics.py 追加） + +```python +@router.get("/churn") +async def churn_list( + session: SessionDep, + channel: Literal["wechat", "whatsapp"] = "wechat", + type: Literal["real", "inactive", "handover"] = "real", + tier: int = Query(30, ge=30, le=90), # 30/60/90;前端给三档 + min_msgs: int = Query(5, ge=1, le=100), + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), +) -> dict: + """A3 流失预警名单(三分法)。""" + if tier not in (30, 60, 90): + raise HTTPException(422, "tier must be 30/60/90") + return await metrics_svc.churn( + session, channel=channel, kind=type, tier=tier, + min_msgs=min_msgs, page=page, page_size=page_size, + ) +``` + +（`Query`/`HTTPException` 从 fastapi import。） + +- [ ] **Step 4: 跑测试到绿** Run: `cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_analytics_churn_unreplied.py -x -q`（此时 unreplied 测试还没写，只有 churn 两个） + +- [ ] **Step 5: Commit** `git commit -am "feat(analytics): A3 流失预警三分法——真流失/从未激活/待交接"` + +--- + +### Task 7: B2 该回未回（unreplied 指标 + 端点） + +**Files:** +- Modify: `yunke-server/app/analytics/metrics.py`（追加 unreplied） +- Modify: `yunke-server/app/api/routers/analytics.py`（追加端点） +- Test: `yunke-server/tests/test_analytics_churn_unreplied.py`（追加） + +- [ ] **Step 1: 写失败测试**（追加到同一测试文件） + +```python +from app.db.models import Customer, Message, PhoneCall # noqa: E402 + +_VALID_TYPES = ("text", "image", "voice", "video", "file", "link") + + +async def _seed_unreplied(session): + now = _NOW + h3 = 3 * 3600 * 1000 + # 该回未回:t1 客户最后一条(is_send=False,text)3h 前 + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t1", day=20260904, + send_cnt=5, recv_cnt=3, last_ts=now - h3, + last_is_send=False)) + session.add(Message(msg_id="t1m", channel="wechat", owner_account_id="wx1", + talker_id="t1", timestamp=now - h3, is_send=False, + type_="text", text="在吗,方案发我一下", + fetched_at=dt.datetime(2026, 9, 4, 12, 0))) + # 员工已回 → 不在清单 + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t2", day=20260904, + send_cnt=5, recv_cnt=3, last_ts=now - h3, + last_is_send=True)) + # 客户最后一条是 unknown 类型 → 无效,不出清单 + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t3", day=20260904, + send_cnt=5, recv_cnt=3, last_ts=now - h3, + last_is_send=False)) + session.add(Message(msg_id="t3m", channel="wechat", owner_account_id="wx1", + talker_id="t3", timestamp=now - h3, is_send=False, + type_="unknown", text=None, + fetched_at=dt.datetime(2026, 9, 4, 12, 0))) + # 尾部 0.5h < 2h 阈值 → 不出 + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t4", day=20260904, + send_cnt=5, recv_cnt=3, last_ts=now - 1800_000, + last_is_send=False)) + session.add(Message(msg_id="t4m", channel="wechat", owner_account_id="wx1", + talker_id="t4", timestamp=now - 1800_000, is_send=False, + type_="text", text="刚发的", + fetched_at=dt.datetime(2026, 9, 4, 12, 0))) + # 通话豁免:最后消息 5h 前没回,但 4h 前员工打过电话 → 视为已回应 + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t5", day=20260904, + send_cnt=5, recv_cnt=3, last_ts=now - 5 * 3600 * 1000, + last_is_send=False)) + session.add(Message(msg_id="t5m", channel="wechat", owner_account_id="wx1", + talker_id="t5", timestamp=now - 5 * 3600 * 1000, + is_send=False, type_="text", text="hello", + fetched_at=dt.datetime(2026, 9, 4, 12, 0))) + session.add(_cust("t5", last_chat=now)) # phone=13800000005 + session.get(Customer, "t5").phone = "13800000005" + session.add(PhoneCall(call_id="pc1", channel="voice_call", + user_id="u1", call_number="13800000005", + start_call_time=now - 4 * 3600 * 1000)) + # 已删好友 → 剔除 + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t6", day=20260904, + send_cnt=5, recv_cnt=3, last_ts=now - h3, + last_is_send=False)) + session.add(Message(msg_id="t6m", channel="wechat", owner_account_id="wx1", + talker_id="t6", timestamp=now - h3, is_send=False, + type_="text", text="已删我的", + fetched_at=dt.datetime(2026, 9, 4, 12, 0))) + session.add(_cust("t6", last_chat=now - h3, delete=1)) + await session.commit() + + +async def test_unreplied(make_client, monkeypatch): + monkeypatch.setattr(metrics, "_now_ms", lambda: _NOW) + async def seed(session): + await _seed_unreplied(session) + out = await metrics.unreplied(session, channel="wechat", hours=2, + page=1, page_size=20) + ids = [r["peer_id"] for r in out["items"]] + assert ids == ["t1"] # t2已回/t3无效类型/t4未超时/t5通话豁免/t6已删 + row = out["items"][0] + assert row["wait_ms"] >= 3 * 3600 * 1000 + assert "方案" in row["preview"] + + await make_client(seed) +``` + +注意：Message 的类型列实际名是 `type_`? **执行时先确认列名**——models.py 里 Message 的列可能是 `type`（SQL 保留字处理过）或 `label`。`grep -n "type" yunke-server/app/db/models.py` 看 Message 类实际列名后照实写（存量代码 messages.py 路由有引用）。`Customer` 的 phone 赋值直接在 `_cust()` 工厂加参数更干净。 + +- [ ] **Step 2: 实现 unreplied**（metrics.py 追加） + +```python +_VALID_TAIL_TYPES = ("text", "image", "voice", "video", "file", "link") + + +def _norm_phone(p: str | None) -> str | None: + """归一:+86/空格/横线剥掉(spec B1 同款规则)。""" + if not p: + return None + out = p.replace(" ", "").replace("-", "") + for pre in ("+86", "86"): + if out.startswith(pre) and len(out) > len(pre): + out = out[len(pre):] + break + return out or None + + +async def unreplied( + session: AsyncSession, *, channel: str, hours: int, page: int, page_size: int, +) -> dict: + """B2 该回未回(spec §3.4)。候选=汇总表最新日行;决赛圈点查类型/预览/通话豁免。""" + now = _now_ms() + cutoff = now - hours * 3600 * 1000 + + # 候选集:每 (channel,owner,peer) 最新日行,方向=客户 且 ≤cutoff(窗口函数双方言) + rn = ( + sa.func.row_number() + .over( + partition_by=( + MsgDailyStats.channel, + MsgDailyStats.owner_account_id, + MsgDailyStats.peer_id, + ), + order_by=MsgDailyStats.day.desc(), + ) + .label("rn") + ) + sub = ( + sa.select( + MsgDailyStats.owner_account_id.label("owner"), + MsgDailyStats.peer_id.label("peer"), + MsgDailyStats.last_ts.label("ts"), + rn, + ) + .where( + MsgDailyStats.channel == channel, + MsgDailyStats.peer_kind == "friend", + MsgDailyStats.last_is_send.is_(False), + MsgDailyStats.last_ts.is_not(None), + MsgDailyStats.last_ts <= cutoff, + ) + ).subquery() + cand = ( + await session.execute( + sa.select(sub.c.owner, sub.c.peer, sub.c.ts).where(sub.c.rn == 1) + ) + ).all() + + total = len(cand) + page_rows = cand[(page - 1) * page_size : (page - 1) * page_size + page_size] + + items: list[dict] = [] + for owner, peer, ts in page_rows: + # 决赛圈点查:尾部消息类型+预览(走 ix_message_channel_owner_time) + last_msg = ( + await session.execute( + sa.select(Message).where( + Message.channel == channel, + Message.owner_account_id == owner, + Message.talker_id == peer, + sa.or_(Message.room_id.is_(None), Message.room_id == ""), + Message.timestamp.is_not(None), + ).order_by(Message.timestamp.desc()).limit(1) + ) + ).scalar_one_or_none() + if last_msg is None or last_msg.type_ not in _VALID_TAIL_TYPES: # 列名按实际 + continue + cust = await session.get(Customer, peer) + if cust is not None and ( + cust.delete_flag == 1 or cust.is_block is True + ): + continue + # 通话豁免:该客户手机号在员工名下有「晚于最后消息」的通话 → 已回应 + tail = ts + phone = _norm_phone(cust.phone if cust else None) + if phone: + amap = await account_map(session, channel) + uid = amap.get(owner) + call_ts = ( + await session.execute( + sa.select(sa.func.max(PhoneCall.start_call_time)).where( + PhoneCall.call_number == phone, + *(([PhoneCall.user_id == uid] if uid else [])), + ) + ) + ).scalar_one_or_none() + if call_ts and call_ts > ts: + continue # 通话已回应 + text = (last_msg.text or "").replace("\n", " ") + items.append({ + "peer_id": peer, + "channel": channel, + "owner_account_id": owner, + "last_ts": ts, + "wait_ms": now - ts, + "preview": text[:50], + "msg_id": last_msg.msg_id, + }) + return {"items": items, "total": total, "page": page, "page_size": page_size} +``` + +注意：`last_msg.type_` 按实际列名替换（见 Step 1 的说明）；`total` 语义 = 候选对话数（决赛圈类型过滤后可能少几行，v1 可接受，注释说明）。补一个 `test_unreplied_total_counts_candidates` 断言 total=5（t1/t3/t4/t5/t6 都是候选，过滤发生在 items）——**或者**把 total 改成 items 过滤后的口径？spec 说「候选集…决赛圈处理」，total=候选数会让前端页码有少量空页。**定案：total = 过滤后条数**——简单实现：先全量决赛圈过滤再 Python 分页（候选最多几千行，点查每行 3 次会慢…）。折中：候选集先点查过滤到完整列表再分页，一页 20 行时点查 60 次=180 个毫秒查询≈可接受，但几千候选就炸。**最终定案（写进代码注释）**：候选 ≤500 时全过滤后分页（total 精确）；>500 时按候选分页（total=候选数，注释口径）。用 `len(cand) <= 500` 分支。 + +- [ ] **Step 3: 端点**（analytics.py 追加） + +```python +@router.get("/unreplied") +async def unreplied_list( + session: SessionDep, + channel: Literal["wechat", "whatsapp"] = "wechat", + hours: int = Query(2, ge=1, le=168), + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), +) -> dict: + """B2 该回未回清单。""" + return await metrics_svc.unreplied( + session, channel=channel, hours=hours, page=page, page_size=page_size + ) +``` + +- [ ] **Step 4: 跑测试到绿** Run: `cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_analytics_churn_unreplied.py -x -q` → Expected: all passed（churn 2 + unreplied 1+） + +- [ ] **Step 5: Commit** `git commit -am "feat(analytics): B2 该回未回——尾部判定/有效类型/通话豁免/已删剔除"` + +### Task 8: C1 敏感词——词库 CRUD + 加词预览 + +**Files:** +- Create: `yunke-server/app/analytics/sensitive/__init__.py`（docstring 包说明） +- Create: `yunke-server/app/analytics/sensitive/words.py` +- Modify: `yunke-server/app/api/routers/analytics.py` +- Test: `yunke-server/tests/test_sensitive.py` + +已确认事实：Message 类型列名是 **`Message.type`**（String(32)，normalized label：text/image/…/unknown）。Task 7 的 `last_msg.type_` 全部写为 `last_msg.type`。 + +- [ ] **Step 1: `sensitive/__init__.py`** + +```python +"""C1 敏感词子系统(spec §3.6)——words(词库)/scanner(扫描)/hits(处置)。 + +内容源抽象:scanner 目前扫 message.text;预留 refer_msg 引文与 D4 ASR +转写文本接入位(二/三批)。日志禁打词内容与命中正文。 +""" +``` + +- [ ] **Step 2: 写失败测试** `tests/test_sensitive.py` + +```python +"""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, SensitiveWord + + +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 +``` + +- [ ] **Step 3: 实现 `sensitive/words.py`** + +```python +"""词库服务——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}") # 词内容不进异常文案之外的位置?词本身是用户输入,回显可 + 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(测试)。""" + 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) + ) + 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) +``` + +- [ ] **Step 4: 端点**（analytics.py 追加；`HTTPException` 已引入） + +```python +from pydantic import BaseModel + +from app.analytics.sensitive import words as words_svc + + +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]: + 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: + 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: + 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: + 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: + 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 +``` + +- [ ] **Step 5: 跑测试** Run: `cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_sensitive.py -x -q` → Expected: 3 passed + +- [ ] **Step 6: Commit** `git commit -am "feat(analytics): C1 敏感词库 CRUD+加词预览——防一加就淹没"` + +--- + +### Task 9: C1 scanner（增量水位扫描 + 全量回填） + +**Files:** +- Create: `yunke-server/app/analytics/sensitive/scanner.py` +- Test: `yunke-server/tests/test_sensitive.py`（追加） + +- [ ] **Step 1: 追加失败测试** + +```python +from app.analytics.sensitive import scanner +from app.db.models import SensitiveHit, SensitiveScanState + + +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) + f2 = dt.datetime(2026, 9, 4, 11, 0) + session.add(_msg("m1", "聊聊私单", f1)) + session.add(_msg("m2", "正常", f1)) + w = await words_svc.create_word(session, word="私单") + await session.commit() + r1 = await scanner.run_incremental(session) + await session.commit() + assert r1["hits"] == 1 and r1["new_watermark"] == f2 # 无行也有 MAX?—— + # ↑ m1/m2 fetched 10:00,水位应=10:00。断言修正: + assert r1["new_watermark"] == f1 + r2 = await scanner.run_incremental(session) # 再跑:水位之后无新行 + await session.commit() + assert r2["hits"] == 0 + n = (await session.execute( + __import__("sqlalchemy").select( + __import__("sqlalchemy").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)) + w = await words_svc.create_word(session, word="私单") + await session.commit() + r1 = await scanner.run_incremental(session) + await session.commit() + assert r1["hits"] == 1 + # 迟到行:timestamp 2026-01(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))) + w = await words_svc.create_word(session, word="坑") # 单字 + await session.commit() + r = await scanner.run_incremental(session) + await session.commit() + assert r["hits"] == 1 # 单字走 LIKE,MySQL ngram 也走 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() + hit = (await session.execute( + __import__("sqlalchemy").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() + r = await scanner.run_incremental(session) + await session.commit() + hit2 = (await session.execute( + __import__("sqlalchemy").select(SensitiveHit) + )).scalars().one() + assert hit2.disposition == "false_positive" + + await make_client(seed) +``` + +- [ ] **Step 2: 实现 `sensitive/scanner.py`** + +```python +"""敏感词扫描器——增量水位(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 增量仅数千);全量回填按词扫描、 +结果 partitions(500) 流式插入,防内存峰值。 +""" +from __future__ import annotations + +import datetime as dt +import logging + +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import Message, SensitiveHit, SensitiveScanState, SensitiveWord +from app.analytics.sensitive.words import _hit_exclude + +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: + return [ + sa.text("MATCH(message.text) AGAINST (:w IN BOOLEAN MODE)") + .bindparams(w=f'"{word}"') + ] + escaped = word.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + return [Message.text.like(f"%{escaped}%")] + + +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 run_incremental(session: AsyncSession) -> dict: + """增量扫描:fetched_at > 水位 的行 × 全部 enabled 词。""" + 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) + + 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) + base = 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) + + rows = (await session.execute(base)).all() + scanned = len(rows) + hits = 0 + if rows and words: + texts = {r[0]: r[6] for r in rows} + for w in words: + hit_ids = { + r[0] for r in rows + if w.word in (texts[r[0]] or "") + } if len(w.word) < 2 or dialect != "mysql" else None + if hit_ids is None: + # MySQL MATCH 路径:对候选行集合再过滤(增量行少,Python 侧 + # 子串判定即可——MATCH 只在全量回填用,增量统一 LIKE 语义 + # 精确)。此处直接子串: + hit_ids = {r[0] for r in rows if w.word in (texts[r[0]] or "")} + sel_rows = [r for r in rows if r[0] in hit_ids] + existing = set( + ( + await session.execute( + sa.select(SensitiveHit.msg_id).where( + SensitiveHit.word_id == w.id, + SensitiveHit.msg_id.in_([r[0] for r in sel_rows] or ["-"]), + ) + ) + ).scalars().all() + ) + hits += await _insert_hits(session, w, sel_rows, existing) + + 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 一次性):每词单独扫全库;水位推进同增量。""" + 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) + existing: set[str] = set() + batch: list = [] + result = await session.stream(stmt) + async for partition in result.partitions(500): + for r in partition: + batch.append(tuple(r)) + existing.update( + ( + await session.execute( + sa.select(SensitiveHit.msg_id).where( + SensitiveHit.word_id == w.id, + SensitiveHit.msg_id.in_( + [b[0] for b in batch] or ["-"] + ), + ) + ) + ).scalars().all() + ) + total_hits += await _insert_hits(session, w, batch, existing) + batch = [] + 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} +``` + +实现说明：① 增量路径统一 Python 子串判定（4h 增量仅数千行，语义精确=LIKE）；MATCH 只在全量回填的 MySQL 路径用（65 万行必须走索引）。上面 `run_incremental` 里那个 `if len(w.word) < 2 or dialect != "mysql"` 的分支两臂相同——落地时合并成一行 `{r[0] for r in rows if w.word in (texts[r[0]] or "")}`。② `session.stream` SQLite(aiosqlite) 支持需验证——测试若报错改 `session.execute` + `.partitions()`（Result 对象同样有 partitions）。③ 测试里 `new_watermark == f1` 修正版为准。 + +- [ ] **Step 3: 跑测试** Run: `cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_sensitive.py -x -q` → Expected: all passed（词库 3 + 扫描 4） + +- [ ] **Step 4: Commit** `git commit -am "feat(analytics): C1 扫描器——fetched_at 水位增量/迟到免疫/幂等/全量回填"` + +--- + +### Task 10: C1 命中列表 + 批量处置端点 + +**Files:** +- Create: `yunke-server/app/analytics/sensitive/hits.py` +- Modify: `yunke-server/app/api/routers/analytics.py` +- Test: `yunke-server/tests/test_sensitive.py`（追加） + +- [ ] **Step 1: 追加失败测试** + +```python +async def test_hits_list_default_filter_and_dispose(make_client): + async def seed(session): + w = await words_svc.create_word(session, word="私单") + session.add(SensitiveHit(word_id=w.id, msg_id="m1", channel="wechat", + ts=1756900000000, snippet="…私单…", + disposition="pending")) + session.add(SensitiveHit(word_id=w.id, msg_id="m2", channel="wechat", + ts=1756900001000, snippet="…私单…", + disposition="ignored")) + await session.commit() + + client = await make_client(seed) + r = await client.get("/api/analytics/sensitive/hits") + body = r.json() + assert body["total"] == 1 # 默认只 pending+confirmed + ids = [h["id"] for h in body["items"]] + r2 = await client.get("/api/analytics/sensitive/hits", + params={"disposition": "ignored"}) + assert r2.json()["total"] == 1 + r3 = await client.patch("/api/analytics/sensitive/hits", + json={"ids": ids, "disposition": "confirmed"}) + assert r3.json()["updated"] == 1 + r4 = await client.get("/api/analytics/sensitive/hits", + params={"disposition": "confirmed"}) + assert r4.json()["total"] == 1 +``` + +- [ ] **Step 2: 实现 `sensitive/hits.py`** + +```python +"""命中查询与处置——默认只显示 pending+confirmed(误报治理闭环)。""" +from __future__ import annotations + +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import SensitiveHit, SensitiveWord + +DISPOSITIONS = ("pending", "confirmed", "false_positive", "ignored") +DEFAULT_SHOW = ("pending", "confirmed") + + +async def list_hits( + session: AsyncSession, *, disposition: str | None = None, + word_id: int | None = None, page: int, page_size: int, +) -> dict: + conds = [] + if disposition: + conds.append(SensitiveHit.disposition == disposition) + else: + conds.append(SensitiveHit.disposition.in_(DEFAULT_SHOW)) + if word_id: + conds.append(SensitiveHit.word_id == word_id) + + total = ( + await session.execute( + sa.select(sa.func.count()).select_from(SensitiveHit).where(*conds) + ) + ).scalar_one() + rows = ( + await session.execute( + sa.select(SensitiveHit, SensitiveWord.word, SensitiveWord.category) + .join(SensitiveWord, SensitiveHit.word_id == SensitiveWord.id) + .where(*conds) + .order_by(SensitiveHit.ts.desc()) + .offset((page - 1) * page_size) + .limit(page_size) + ) + ).all() + items = [ + { + "id": h.id, "word_id": h.word_id, "word": word, "category": category, + "msg_id": h.msg_id, "channel": h.channel, + "owner_account_id": h.owner_account_id, "talker_id": h.talker_id, + "room_id": h.room_id, "ts": h.ts, "snippet": h.snippet, + "disposition": h.disposition, + } + for h, word, category in rows + ] + return {"items": items, "total": int(total), "page": page, "page_size": page_size} + + +async def dispose(session: AsyncSession, ids: list[int], disposition: str) -> int: + if disposition not in DISPOSITIONS: + raise ValueError(f"bad disposition: {disposition!r}") + if not ids: + return 0 + res = await session.execute( + sa.update(SensitiveHit) + .where(SensitiveHit.id.in_(ids)) + .values(disposition=disposition) + ) + return int(res.rowcount or 0) +``` + +- [ ] **Step 3: 端点**（analytics.py 追加） + +```python +from app.analytics.sensitive import hits as hits_svc + + +class DisposeIn(BaseModel): + ids: list[int] + disposition: str + + +@router.get("/sensitive/hits") +async def list_hits( + session: SessionDep, + disposition: str | None = None, + word_id: int | None = None, + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), +) -> dict: + return await hits_svc.list_hits( + session, disposition=disposition, word_id=word_id, + page=page, page_size=page_size, + ) + + +@router.patch("/sensitive/hits") +async def dispose_hits(session: SessionDep, body: DisposeIn) -> dict: + try: + n = await hits_svc.dispose(session, body.ids, body.disposition) + await session.commit() + return {"updated": n} + except ValueError as exc: + raise HTTPException(422, str(exc)) from exc +``` + +- [ ] **Step 4: 跑测试 + Commit** Run: `pytest tests/test_sensitive.py -x -q` → all passed；`git commit -am "feat(analytics): C1 命中列表默认过滤+批量处置"` + +--- + +### Task 11: 调度 4h job + CLI 回填 + full_sync 步骤 + +**Files:** +- Modify: `yunke-server/app/sync/scheduler.py`（start() 里加 job + 新方法 `_analytics_tick`） +- Modify: `yunke-server/app/sync/cli.py`（staff-daily-stats / sensitive-scan 两个子命令） +- Modify: `yunke-server/scripts/full_sync_all.sh`（daily-stats 步骤后追加一步） +- Test: `yunke-server/tests/test_scheduler_analytics_tick.py` + +- [ ] **Step 1: scheduler.py 加 job**（`start()` 内 weekly job 之后、`sched.start()` 之前） + +```python + # 4h 分析链:stat_daily_staff 滚动3天刷新 + 敏感词增量扫描(纯 DB, + # 无浏览器子进程;与消息增量同节奏,读滞后语义与「数据截至」一致) + sched.add_job( + self._analytics_tick, + IntervalTrigger(hours=self.message_interval_hours), + id="yunke_analytics_tick", + replace_existing=True, + coalesce=True, + max_instances=1, + ) +``` + +新方法（`_safe_page_run` 附近）： + +```python + async def _analytics_tick(self) -> None: + """4h 分析链(stat_daily_staff + 敏感词增量)——异常只记日志不断链。""" + try: + async with self._session_maker() as session: + from app.storage.staff_summary import recompute_days, rolling_days + from app.storage.summary import day_key + from app.analytics.sensitive import scanner + + today = day_key(int(dt.datetime.now().timestamp() * 1000)) + n = await recompute_days(session, rolling_days(today, 3)) + scan = await scanner.run_incremental(session) + await session.commit() + logger.info( + "[analytics-tick] staff_days=%s scanned=%s hits=%s", + n, scan.get("scanned"), scan.get("hits"), + ) + except Exception: # noqa: BLE001 + logger.exception("analytics tick failed") +``` + +（`dt`/`logging` 该文件已有 import；确认 `import datetime as dt` 存在，没有则补。） + +- [ ] **Step 2: cli.py 加两个子命令**（照 `daily-stats` 的三处模式：`_build_parser` 加 parser、`main()` 加分支、新增 `_cmd_*`） + +```python + # ``staff-daily-stats`` —— p_stats 后面 + p_staff_stats = sub.add_parser( + "staff-daily-stats", + help="Rebuild stat_daily_staff (staff x day rollup).", + description="Full rebuild of the staff daily rollup. Idempotent.", + ) + # ``sensitive-scan`` —— + p_scan = sub.add_parser( + "sensitive-scan", + help="Sensitive-word scan (incremental; --full for backfill).", + ) + p_scan.add_argument("--full", action="store_true", + help="Full backfill scan (first deploy).") +``` + +命令实现： + +```python +async def _cmd_staff_daily_stats(args: argparse.Namespace) -> int: + """全量重建 stat_daily_staff(夜间自愈/部署回填,幂等)。""" + settings = get_settings() + engine, session_maker = await _init_db(settings.db_url) + try: + from app.storage.staff_summary import recompute_all + + async with session_maker() as session: + days = await recompute_all(session) + await session.commit() + print(f"[staff-daily-stats] rebuilt {days} days") + return 0 + finally: + await engine.dispose() + + +async def _cmd_sensitive_scan(args: argparse.Namespace) -> int: + settings = get_settings() + engine, session_maker = await _init_db(settings.db_url) + try: + from app.analytics.sensitive import scanner + + async with session_maker() as session: + if args.full: + out = await scanner.run_full(session) + else: + out = await scanner.run_incremental(session) + await session.commit() + print(f"[sensitive-scan] {out}") + return 0 + finally: + await engine.dispose() +``` + +（`main()` 加两个 `if args.command == ...` 分支。） + +- [ ] **Step 3: full_sync_all.sh 追加步骤**（先 `grep -n "daily-stats" scripts/full_sync_all.sh` 找到该步骤的写法，紧随其后镜像加一行；形如） + +```bash +step_n "staff daily stats" "$PY -m app.sync.cli staff-daily-stats" +``` + +（变量名/函数名以脚本实际为准；这一步是夜间全量自愈的第 12 步。） + +- [ ] **Step 4: 写调度测试** `tests/test_scheduler_analytics_tick.py` + +```python +"""_analytics_tick:滚动刷新+增量扫描+异常吞掉。""" +from __future__ import annotations + +import pytest + +from app.db.models import MsgDailyStats, SensitiveWord +from app.sync.scheduler import SyncScheduler + + +@pytest.fixture +async def sched_engine(): + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + from sqlalchemy.pool import StaticPool + from app.db.base import Base + + engine = create_async_engine( + "sqlite+aiosqlite:///:memory:", poolclass=StaticPool, + connect_args={"check_same_thread": False}, + ) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + maker = async_sessionmaker(engine, expire_on_commit=False) + yield engine, maker + await engine.dispose() + + +async def test_analytics_tick_runs(sched_engine): + engine, maker = sched_engine + async with maker() as s: + s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t1", day=20260904, + send_cnt=1, recv_cnt=0, last_ts=1756950000000, + last_is_send=True)) + s.add(SensitiveWord(word="私单")) + await s.commit() + sch = SyncScheduler(engine=engine, session_maker=maker) + await sch._analytics_tick() # 不抛即过 + from sqlalchemy import func, select + from app.db.models import StatDailyStaff + + async with maker() as s: + n = (await s.execute( + select(func.count()).select_from(StatDailyStaff) + )).scalar_one() + assert n >= 1 +``` + +- [ ] **Step 5: 跑测试 + 全量回归** + +Run: `cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_scheduler_analytics_tick.py -x -q && .venv/Scripts/python.exe -m pytest tests/ -q` +Expected: all passed + +- [ ] **Step 6: Commit** `git commit -am "feat(analytics): 4h 调度分析链+CLI 回填+full_sync 夜间自愈步"` + +### Task 12: 前端骨架——echarts + useChart + 路由 + 侧栏组 + API 模块 + +**Files:** +- Modify: `yunkes/package.json`（+echarts） +- Create: `yunkes/src/composables/useChart.ts` +- Create: `yunkes/src/api/analytics.ts` +- Modify: `yunkes/src/router/index.ts` +- Modify: `yunkes/src/layouts/MainLayout.vue` +- Create: `yunkes/src/views/analytics/` 四个占位页（后续任务填实） + +- [ ] **Step 1: 安装 echarts** Run: `cd yunkes && npm install echarts` + +- [ ] **Step 2: `src/composables/useChart.ts`** + +```typescript +// ECharts 按需封装(2026-09-04):注册/主题/自适应/销毁一处管——后续所有 +// 图表页复用(二批 A2/A4/A5 直接 import 本 composable)。 +import { onBeforeUnmount, onMounted, ref, type Ref } from 'vue' +import * as echarts from 'echarts/core' +import { LineChart, BarChart } from 'echarts/charts' +import { GridComponent, TooltipComponent, LegendComponent } from 'echarts/components' +import { CanvasRenderer } from 'echarts/renderers' +import type { EChartsCoreOption } from 'echarts/core' + +echarts.use([LineChart, BarChart, GridComponent, TooltipComponent, LegendComponent, CanvasRenderer]) + +export function useChart(el: Ref) { + let chart: echarts.ECharts | null = null + let ro: ResizeObserver | null = null + + const render = (option: EChartsCoreOption) => { + if (!el.value) return + if (!chart) { + chart = echarts.init(el.value) + ro = new ResizeObserver(() => chart?.resize()) + ro.observe(el.value) + } + chart.setOption(option, true) // notMerge:指标切换时丢弃旧系列 + } + + onBeforeUnmount(() => { + ro?.disconnect() + chart?.dispose() + chart = null + }) + + return { render } +} + +export { echarts } +export const CHART_REF = ref // 防止 tree-shake 误删(无实际用途可去掉) +``` + +（`CHART_REF` 行删掉——多余。落地时不要。） + +- [ ] **Step 3: `src/api/analytics.ts`**（类型与后端返回一一对应） + +```typescript +// 数据分析板块 API(client.ts 的 http 已带 Bearer + baseURL=/api) +import { getJson, http } from './client' + +export type Channel = 'wechat' | 'whatsapp' +export type Scope = 'today' | 'week' | 'month' + +export interface OverviewCards { + msg_total: number + active_customers: number + active_staff: number + call_cnt: number + first_contact: number + prev_msg_total?: number + msg_total_delta?: number | null +} +export interface TrendPoint { day: number; msg: number; active_customers: number; call_cnt: number } +export interface AnomalyStaff { user_id: string; yesterday: number; prev7_mean: number } +export interface Anomaly { total_delta_pct: number; yesterday_total: number; prev7_mean: number; staff: AnomalyStaff[] } +export interface OverviewResp { + cards: OverviewCards + trend: TrendPoint[] + anomaly: Anomaly | null +} +export interface HealthResp { data_as_of: number | null; entities: { table: string; last_seen: string | null }[] } +export interface ChurnRow { + friend_id: string; channel: string; nickname: string; owner_account_id: string | null + user_id: string | null; last_chat_time: number | null; hist_msgs: number; active_days: number +} +export interface UnrepliedRow { + peer_id: string; channel: string; owner_account_id: string; last_ts: number + wait_ms: number; preview: string; msg_id: string +} +export interface Page { items: T[]; total: number; page: number; page_size: number } +export interface SensitiveWordRow { + id: number; word: string; category: string | null; enabled: boolean + min_len: number; exclude_words: string[] +} +export interface SensitiveHitRow { + id: number; word_id: number; word: string; category: string | null; msg_id: string + channel: string; owner_account_id: string | null; talker_id: string | null + room_id: string | null; ts: number | null; snippet: string | null; disposition: string +} + +export const getOverview = (channel: Channel, scope: Scope) => + getJson('/analytics/overview', { channel, scope }) +export const getAnalyticsHealth = () => getJson('/analytics/health') +export const getChurn = (params: Record) => + getJson>('/analytics/churn', params) +export const getUnreplied = (params: Record) => + getJson>('/analytics/unreplied', params) +export const getWords = () => getJson('/analytics/sensitive/words') +export const previewWord = (data: { word: string; exclude_words?: string[] }) => + http.post('/analytics/sensitive/words/preview', data) as unknown as Promise<{ count: number; warn: boolean }> +export const createWord = (data: { word: string; category?: string; exclude_words?: string[] }) => + http.post('/analytics/sensitive/words', data) as unknown as Promise<{ id: number; word: string }> +export const patchWord = (id: number, data: Record) => + http.patch(`/analytics/sensitive/words/${id}`, data) as unknown as Promise> +export const deleteWord = (id: number) => + http.delete(`/analytics/sensitive/words/${id}`) as unknown as Promise<{ ok: boolean }> +export const getHits = (params: Record) => + getJson>('/analytics/sensitive/hits', params) +export const disposeHits = (ids: number[], disposition: string) => + http.patch('/analytics/sensitive/hits', { ids, disposition }) as unknown as Promise<{ updated: number }> +``` + +（注意：`client.ts` 是否已导出 `http.patch/http.delete`？——`http` 是 axios 实例直接可用；若 client.ts 只导出了 getJson/http 两项则无需改。检查 `export` 再动手。） + +- [ ] **Step 4: 路由**（router/index.ts，`/search` 之后加） + +```typescript + { + // 数据分析板块(spec 2026-09-04):四页 + path: '/analytics/dashboard', + name: 'analytics-dashboard', + component: () => import('@/views/analytics/DashboardView.vue'), + meta: { title: '数据看板' }, + }, + { + path: '/analytics/churn', + name: 'analytics-churn', + component: () => import('@/views/analytics/ChurnView.vue'), + meta: { title: '流失预警' }, + }, + { + path: '/analytics/unreplied', + name: 'analytics-unreplied', + component: () => import('@/views/analytics/UnrepliedView.vue'), + meta: { title: '该回未回' }, + }, + { + path: '/analytics/sensitive', + name: 'analytics-sensitive', + component: () => import('@/views/analytics/SensitiveView.vue'), + meta: { title: '敏感词' }, + }, +``` + +- [ ] **Step 5: 侧栏组**（MainLayout.vue「通用功能」三行之后、`` 之前；照 wx/wa 组的样式惯例用平铺菜单项——分析页内部自带 channel 切换，无需挂 channel 参数） + +```html + + + + + 数据看板 + + + 流失预警 + + + 该回未回 + + + 敏感词 + + +``` + +- [ ] **Step 6: 四个占位页**（每页先放最小可编译内容，后续任务填实） + +```vue + + + +``` + +（四页同款，标题分别 流失预警/该回未回/敏感词。） + +- [ ] **Step 7: 构建验证** Run: `cd yunkes && npm run build` → Expected: 无类型错误 + +- [ ] **Step 8: Commit** + +```bash +git add yunkes/package.json yunkes/package-lock.json yunkes/src/composables/useChart.ts yunkes/src/api/analytics.ts yunkes/src/router/index.ts yunkes/src/layouts/MainLayout.vue yunkes/src/views/analytics/ +git commit -m "feat(analytics): 前端板块骨架——echarts按需/useChart/路由/侧栏组/API模块" +``` + +--- + +### Task 13: DashboardView（A1 总览页） + +**Files:** +- Create(填实): `yunkes/src/views/analytics/DashboardView.vue` + +- [ ] **Step 1: 实现页面**（结构=筛选栏 → 异常警示条 → 卡片行 → 趋势图；seq 守卫防 channel 切换竞态——项目惯例） + +```vue + + + + + +``` + +（若 `el-radio-button` 的 value prop 在当前 Element Plus 版本用 `label`，照项目其他页写法统一。） + +- [ ] **Step 2: 构建验证** Run: `cd yunkes && npm run build` → Expected: 通过 + +- [ ] **Step 3: 目测**（本地起后端+前端，或直接看构建产物）确认卡片/警示条/折线渲染、channel 切换无竞态。 + +- [ ] **Step 4: Commit** `git commit -am "feat(analytics): A1 数据看板页——卡片/异常警示/趋势折线/数据截至角标"` + +--- + +### Task 14: ChurnView（A3 流失预警页） + +**Files:** +- Create(填实): `yunkes/src/views/analytics/ChurnView.vue` + +- [ ] **Step 1: 实现页面**（三分法 tab + tier 筛选 + 表格 + CSV 导出 + 行点击跳聊天窗；CSV 用项目现成 `csvExport.ts` 全量翻页惯例） + +```vue + + + +``` + +（`exportRowsCsv` 的实际函数名/签名以 `yunkes/src/utils/csvExport.ts` 为准——先 grep 再用；导出行数上限遵守项目 cap 惯例。） + +- [ ] **Step 2: 构建验证 + Commit** Run: `npm run build` → 通过；`git commit -am "feat(analytics): A3 流失预警页——三分法/分层/CSV导出/跳聊天"` + +--- + +### Task 15: UnrepliedView（B2 该回未回页） + +**Files:** +- Create(填实): `yunkes/src/views/analytics/UnrepliedView.vue` + +- [ ] **Step 1: 实现页面**（时长筛选 1/2/4/24h + 表格 + 等待时长红橙着色 + 行点击跳聊天） + +```vue + + + +``` + +- [ ] **Step 2: 构建验证 + Commit** Run: `npm run build` → 通过；`git commit -am "feat(analytics): B2 该回未回页——时长筛选/等待着色/msgid定位跳转"` + +--- + +### Task 16: SensitiveView（C1 词库 + 命中处置页） + +**Files:** +- Create(填实): `yunkes/src/views/analytics/SensitiveView.vue` + +- [ ] **Step 1: 实现页面**（双 tab：词库管理（CRUD+预览）/ 命中列表（默认 pending+confirmed，多选批量处置，跳上下文）） + +```vue + + + +``` + +（词内容不在 console 打印；删除确认文案不回显词本身以外的命中内容。） + +- [ ] **Step 2: 构建验证 + Commit** Run: `npm run build` → 通过；`git commit -am "feat(analytics): C1 敏感词页——词库CRUD/预览警示/批量处置/上下文跳转"` + +--- + +### Task 17: 全量回归 + 部署收尾 + +**Files:** +- Modify: `docs/DEPLOY.md`（升级一节追加回填命令） + +- [ ] **Step 1: 后端全量测试** Run: `cd yunke-server && .venv/Scripts/python.exe -m pytest tests/ -q` → Expected: all passed（原 512+ 新增 ~20） + +- [ ] **Step 2: 前端构建** Run: `cd yunkes && npm run build` → Expected: 通过 + +- [ ] **Step 3: DEPLOY.md 追加升级注记**（升级流程一节，照现有条目格式加） + +```markdown +### 数据分析板块首部署附加步骤(一次性) + +迁移自动跑(4 新表+message.fetched_at 索引,86 万行建索引约十几秒)后, +夜间窗口执行两条回填: + +```bash +docker compose exec backend python -m app.sync.cli staff-daily-stats +docker compose exec backend python -m app.sync.cli sensitive-scan --full +``` + +> fetched_at 索引建立会锁表写入,避开抓取窗口(与 ngram 索引同注意事项)。 +> 之后 4h 调度自动增量,无需再手动跑。 +``` + +- [ ] **Step 4: Commit** `git commit -am "docs(deploy): 数据分析板块首部署回填注记"` + +- [ ] **Step 5: 服务器部署**（用户执行/确认后操作,照 DEPLOY.md 流程） + 1. `git push`（或 bundle：`git bundle create x.bundle <服务器HEAD>..master` + scp + 服务器 `git pull /tmp/x.bundle master`） + 2. 服务器 `sudo docker compose up -d --build`（迁移自动） + 3. 夜间窗口跑 Step 3 两条回填 + 4. `sudo docker compose restart frontend`（防 nginx DNS 缓存 502——9-03 踩过） + 5. 浏览器验收四页：`http://219.157.255.213:32002/analytics/dashboard` + +--- + +## 任务依赖图 + +``` +T1 模型迁移 ─→ T2 health ─→ T3 staff_summary ─→ T4 metrics.overview ─→ T5 /overview + │ │ + │ └─→ T6 churn / T7 unreplied(并行) + │ + └─→ T8 词库 ─→ T9 scanner ─→ T10 hits ─→ T11 调度/CLI(汇合 T3+T9) +T12 前端骨架 ─→ T13 Dashboard(依赖 T5) / T14 Churn(依赖 T6) / T15 Unreplied(依赖 T7) / T16 Sensitive(依赖 T8+T10) +T17 收尾(依赖全部) +``` + +## 验收清单（对照 spec §8 口径速查表） + +- [ ] A1 五卡片 + 周月环比 + 异常警示条 + 30 天折线 + 数据截至角标 +- [ ] A3 三分法名单 + tier 分层 + CSV + 跳聊天 +- [ ] B2 默认 2h 可调 + 有效类型过滤 + 通话豁免 + msgid 定位跳转 +- [ ] C1 加词预览警示 + 批量处置 + 默认过滤 + 词删除级联 +- [ ] stat_daily_staff 判别法测试(删 message 不丢数) + 空表回退 +- [ ] 生产无裸扫(全部查询骑汇总表/小表/索引点查) +- [ ] 敏感词内容不出现在任何日志 +