diff --git a/docs/superpowers/plans/2026-09-08-management-console-batch-a-data-foundation.md b/docs/superpowers/plans/2026-09-08-management-console-batch-a-data-foundation.md new file mode 100644 index 0000000..507076d --- /dev/null +++ b/docs/superpowers/plans/2026-09-08-management-console-batch-a-data-foundation.md @@ -0,0 +1,2071 @@ +# 云客经营管理后台·批次A（数据底座）实施计划 + +> **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:** 落地经营管理后台的数据底座——6 张新表、首响事实跨日精确计算、员工指标宽表生成、两条回填 CLI，全部带判别性测试。 + +**Architecture:** 不动现有写钩子与两张汇总表；新增 `first_reply_fact`（跨日精确串）+ `employee_metric_daily`（1:1 口径宽表）+ `employee_call_daily`（通话小表）三张数据表与三张业务表（评分配置/快照/辅导，本批次只建模型不写业务逻辑）。依据 spec：[2026-09-08-yunke-management-console-implementation-v2.md](./2026-09-08-yunke-management-console-implementation-v2.md) §1-§4、§9（本计划 = v2 的 Task 0-2 展开；评分/异常/API/前端在批次 B-F 另出计划）。 + +**Tech Stack:** Python 3.12 + FastAPI + SQLAlchemy 2.0 async + Alembic；测试 pytest + 内存 SQLite（`StaticPool`）；生产 MySQL 8.4（本地 Docker，`DB_URL` 已在 `.env` 指向 `mysql+asyncmy://yunke:yunke@localhost:3307/yunke`）。 + +--- + +## 执行者必读：关键代码事实与口径 + +1. **日切 vs 跨日**：`app/storage/summary.py` `_thread_stats` 是日切近似——客户 23:00 发、员工次日 9:00 回的场景**整条样本丢失**（Day1 串不闭合、Day2 重起算无客户消息）。本计划的 `compute_conversation` 按会话全历史计算，跨夜串不丢（spec §1.1 #1）。 +2. **混合口径**：`app/storage/staff_summary.py` 聚合的 `sent/recv/threads/answered/fr` 无 kind 过滤 = friend+room 混合。宽表消息计数列/串类列一律 **friend-only**（spec §1.1 #2）。 +3. **同源原则**：宽表串类三列与 fact 行出自**同一个** `compute_conversation` 调用——回复率分子分母一致（spec §1.1 #5）。 +4. **通话无渠道**：`PhoneCall.channel` 是通话类型（`phone_call`/`voice_call`）非消息渠道（models.py:512 docstring），通话进独立小表 `employee_call_daily(day, user_id)`（spec §1.2 #9）。 +5. **员工主数据** = `staff` 表（`user_id` PK + `is_on_job`，models.py:126）。幽灵员工（account_map 映射不上或不在 staff）过滤不落宽表（spec §1.3 #11）。 +6. **day 口径**：Asia/Shanghai 自然日 yyyymmdd，一律 Python 计算（复用 `app.storage.summary.day_key`），不用 SQL 日期函数。 +7. **生命周期口径**（spec §1.1 #3 + v1 §21 默认）：new=首次沟通≤7 天；reactivated=当日回流且前次间隔>30 天；active=最近互动≤6 天；cooling=7~29 天；dormant=≥30 天且历史消息≥5 条（对齐 churn real 的 min_msgs=5）。`customer` 归属用 `owner_account_id`→`account_map`→`user_id`（当前归属口径，换人后归新员工）。 +8. **测试命令**（Windows venv，在 `yunke-server/` 下）：`.venv/Scripts/python.exe -m pytest tests/ -v`。提交信息用中文 conventional 风格（参考 `git log`）。**仓库 master 单主干，直接在 master 提交**（项目惯例，不用 worktree）。 +9. **本地起服务前先查 Docker**：yunke-backend-1/frontend-1 容器占 8000/5173，需保持 stopped（`docker stop yunke-backend-1 yunke-frontend-1`；mysql/scheduler 常驻）。 + +## 文件地图 + +| 文件 | 职责 | 动作 | +|---|---|---| +| `yunke-server/app/db/models.py` | 6 张新表 ORM 定义 | 追加 | +| `yunke-server/migrations/versions/_management_phase1_tables.py` | 建表迁移（幂等守卫） | 新建 | +| `yunke-server/app/storage/first_reply_facts.py` | 跨日精确串算法 + fact 落库 | 新建 | +| `yunke-server/app/analytics/lifecycle.py` | 生命周期判定纯函数（阈值单一定义处） | 新建 | +| `yunke-server/app/storage/employee_metrics.py` | 宽表 + 通话小表生成 | 新建 | +| `yunke-server/app/sync/cli.py` | 两条回填命令 | 修改 | +| `yunke-server/tests/conftest.py` | `make_session` fixture | 修改 | +| `yunke-server/tests/test_management_models.py` 等 5 个测试文件 | 判别性测试 | 新建 | +| `docs/superpowers/plans/2026-09-08-management-batch-a-verification.md` | Task 0 核实记录 | 新建 | +| `docs/DEPLOY.md` | 两条回填命令部署注记 | 修改 | + +--- + +### Task 0: 五项核实 + 回测名单（不改代码） + +**Files:** +- Create: `docs/superpowers/plans/2026-09-08-management-batch-a-verification.md` + +- [ ] **Step 1: 确认本地 MySQL 容器名与连通** + +```bash +docker ps --format '{{.Names}}' # 期望看到 yunke-mysql-1（常驻） +``` + +以下 SQL 用 `docker exec -it yunke-mysql-1 mysql -uyunke -pyunke yunke -e ""` 执行。 + +- [ ] **Step 2: 核实①员工主数据（staff 表）** + +```sql +SELECT COUNT(*) AS total, SUM(is_on_job=1) AS on_job FROM staff; +SELECT COUNT(DISTINCT user_id) FROM wechat_account WHERE user_id IS NOT NULL; +SELECT COUNT(DISTINCT user_id) FROM whatsapp_account WHERE user_id IS NOT NULL; +``` + +期望：staff 总数≈41；账号表 user_id 全部能落在 staff（`SELECT COUNT(*) FROM wechat_account w LEFT JOIN staff s ON w.user_id=s.user_id WHERE w.user_id IS NOT NULL AND s.user_id IS NULL` 应为 0 或极少——不为 0 时记录哪些幽灵账号）。 + +- [ ] **Step 3: 核实②phone_call 列与单位** + +```sql +SHOW COLUMNS FROM phone_call; +SELECT channel, COUNT(*) FROM phone_call GROUP BY channel; +SELECT call_id, channel, user_id, call_number, start_call_time, call_duration, incoming_call +FROM phone_call ORDER BY start_call_time DESC LIMIT 10; +SELECT COUNT(*) AS total, SUM(call_duration) AS dur_sum, AVG(call_duration) FROM phone_call WHERE call_duration IS NOT NULL; +``` + +判定并记录：`call_duration` 单位（秒/毫秒——对照已知总量约 73 小时/3,190 条 ⇒ 均值≈82 秒）；`call_status`/`incoming_call` 取值分布（后续 Batch B 接通率可能用）；`start_call_time` 是否毫秒 epoch。 + +- [ ] **Step 4: 核实③customer 归属字段填充率** + +```sql +SELECT COUNT(*) AS total, + SUM(user_id IS NOT NULL) AS uid_fill, + SUM(owner_account_id IS NOT NULL) AS owner_fill, + SUM(delete_flag=1) AS deleted +FROM customer; +SELECT channel, COUNT(*), SUM(owner_account_id IS NOT NULL) FROM customer GROUP BY channel; +``` + +记录：owner_fill 比例（决定生命周期归属覆盖面）；`user_id` 列与 `owner_account_id→account_map` 两条路径是否一致（抽 5 行比对）。默认走 owner_account_id 路径。 + +- [ ] **Step 5: 核实④群占比（量化混合口径偏差）** + +```sql +SELECT channel, peer_kind, COUNT(*) AS rows_cnt, + SUM(send_cnt+recv_cnt) AS msgs, SUM(cust_threads) AS threads +FROM msg_daily_stats GROUP BY channel, peer_kind; +``` + +记录 friend vs room 的 threads 占比——这就是现有看板首响/回复率被群污染的程度（写进核实文档，Batch C 上线时口径说明引用）。 + +- [ ] **Step 6: 核实⑤call_followup join 键（voice_call 客户号能否对上消息 talker）** + +```sql +SELECT COUNT(*) FROM phone_call p +WHERE p.channel='voice_call' AND p.call_number IS NOT NULL + AND EXISTS(SELECT 1 FROM message m WHERE m.talker_id = p.call_number + AND (m.room_id IS NULL OR m.room_id='')); +``` + +记录命中率。低于 50% 时：call_followup_count 一期标记低置信（列照建，口径注明）。 + +- [ ] **Step 7: 选定 3-5 名回测员工** + +```sql +-- 消息量 top（高负载） +SELECT s.user_id, s.name, SUM(m.msg_sent+m.msg_recv) AS msgs +FROM stat_daily_staff s JOIN staff st ON s.user_id=st.user_id +JOIN staff st2 ON st2.user_id=s.user_id, stat_daily_staff m -- (示意;实际写法见下) +``` + +实际用（更直接）： + +```sql +SELECT user_id, SUM(msg_sent+msg_recv) AS msgs FROM stat_daily_staff GROUP BY user_id ORDER BY msgs DESC LIMIT 5; +SELECT user_id, SUM(msg_sent+msg_recv) AS msgs FROM stat_daily_staff WHERE channel='whatsapp' GROUP BY user_id ORDER BY msgs DESC LIMIT 3; +SELECT user_id, SUM(active_rooms) AS rooms FROM stat_daily_staff GROUP BY user_id ORDER BY rooms DESC LIMIT 3; +SELECT user_id, SUM(msg_sent+msg_recv) FROM stat_daily_staff GROUP BY user_id ORDER BY 2 ASC LIMIT 5; -- 低样本 +``` + +覆盖：微信为主 1 名、WA 为主 1 名、群运营为主 1 名、高负载 1 名、低样本 1 名（可合并）。 + +- [ ] **Step 8: 写核实文档并提交** + +把 Step 2-7 结果写入 `docs/superpowers/plans/2026-09-08-management-batch-a-verification.md`（含每项结论 + 与本计划默认假设的偏差；**偏差影响后续任务实现细节的，在对应任务执行时按核实结论调整并在提交信息注明**）。 + +```bash +git add docs/superpowers/plans/2026-09-08-management-batch-a-verification.md +git commit -m "docs(management): 批次A Task0核实记录——phone_call/customer/staff/群占比/回测名单" +``` + +--- + +### Task 1: 6 张表模型 + make_session fixture + 冒烟测试 + +**Files:** +- Modify: `yunke-server/tests/conftest.py`（追加 `make_session` fixture） +- Modify: `yunke-server/app/db/models.py`（文件末尾 SensitiveScanState 之后追加 6 个类；`__all__` 按字母序插入 6 个名字） +- Test: `yunke-server/tests/test_management_models.py` + +- [ ] **Step 1: conftest 追加存储层测试 fixture** + +在 `tests/conftest.py` 末尾追加（import 区已有 `async_sessionmaker/create_async_engine/StaticPool`）： + +```python +@pytest.fixture +async def make_session() -> AsyncIterator[AsyncSession]: + """独立内存 SQLite AsyncSession(不挂 API 依赖)——存储层/管理域测试用。 + + 与 make_client 共用 StaticPool 单连接模式;每个测试独享全新 schema。 + """ + 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) + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + async with SessionLocal() as s: + yield s + await engine.dispose() +``` + +- [ ] **Step 2: 写模型冒烟测试（先失败）** + +新建 `tests/test_management_models.py`： + +```python +"""管理域 6 张新表的模型冒烟测试(批次A Task1)。 + +只验证:建表可用、主键/唯一约束生效、默认值正确。业务逻辑在 +first_reply_facts / employee_metrics 的测试里。 +""" +import datetime as dt + +import pytest +import sqlalchemy as sa +from sqlalchemy.exc import IntegrityError + +from app.db.models import ( + EmployeeCallDaily, + EmployeeMetricDaily, + EmployeeScoreConfig, + EmployeeScoreSnapshot, + FirstReplyFact, + ManagementFollowup, +) + + +async def test_first_reply_fact_roundtrip(make_session): + s = make_session + s.add(FirstReplyFact(channel="wechat", owner_account_id="wx1", peer_id="p1", + thread_end_ts=1_700_000_000_000, + reply_ts=1_700_000_600_000, + day=20260901, fr_ms=600_000)) + await s.commit() + val = (await s.execute(sa.select(FirstReplyFact.fr_ms))).scalar_one() + assert val == 600_000 + + +async def test_first_reply_fact_unique_sample(make_session): + s = make_session + kwargs = dict(channel="wechat", owner_account_id="wx1", peer_id="p1", + thread_end_ts=1, reply_ts=2, day=20260901, fr_ms=1) + s.add(FirstReplyFact(**kwargs)) + await s.commit() + s.add(FirstReplyFact(**kwargs)) + with pytest.raises(IntegrityError): + await s.commit() + + +async def test_metric_daily_versioning_coexists(make_session): + s = make_session + for v in ("v1", "v2"): + s.add(EmployeeMetricDaily(day=20260901, channel="wechat", + user_id="u1", metric_version=v)) + await s.commit() + n = (await s.execute( + sa.select(sa.func.count()).select_from(EmployeeMetricDaily))).scalar_one() + assert n == 2 + + +async def test_metric_daily_defaults_zero_not_none(make_session): + s = make_session + s.add(EmployeeMetricDaily(day=20260901, channel="wechat", user_id="u1")) + await s.commit() + row = (await s.execute(sa.select(EmployeeMetricDaily))).scalar_one() + assert row.msg_sent == 0 and row.cust_threads == 0 and row.fr_sum_ms == 0 + assert row.metric_version == "v1" + assert row.data_complete is False or row.data_complete == 0 # SQLite 0/1 + + +async def test_call_daily_roundtrip(make_session): + s = make_session + s.add(EmployeeCallDaily(day=20260901, user_id="u1", call_count=2, + call_seconds=164, answered_call_count=1, + call_followup_count=1)) + await s.commit() + row = (await s.execute(sa.select(EmployeeCallDaily))).scalar_one() + assert row.call_seconds == 164 + + +async def test_score_config_version_unique(make_session): + s = make_session + s.add(EmployeeScoreConfig(version="v1.0", name="一期试算", + config_json={"dimensions": {}})) + await s.commit() + s.add(EmployeeScoreConfig(version="v1.0", name="重复", + config_json={})) + with pytest.raises(IntegrityError): + await s.commit() + + +async def test_snapshot_batch_user_unique(make_session): + s = make_session + common = dict(snapshot_batch_id="b1", period_type="week", + period_start=20260901, period_end=20260907, + score_config_version="v1.0", metric_version="v1", + data_as_of=20260907) + s.add(EmployeeScoreSnapshot(user_id="u1", **common)) + await s.commit() + s.add(EmployeeScoreSnapshot(user_id="u1", **common)) + with pytest.raises(IntegrityError): + await s.commit() + + +async def test_followup_defaults(make_session): + s = make_session + s.add(ManagementFollowup(employee_user_id="u1", source_type="manual")) + await s.commit() + row = (await s.execute(sa.select(ManagementFollowup))).scalar_one() + assert row.status == "open" + assert row.created_at is not None +``` + +- [ ] **Step 3: 跑测试确认失败** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_management_models.py -v +``` + +期望：collection 报 `ImportError: cannot import name 'FirstReplyFact'`。 + +- [ ] **Step 4: models.py 追加 6 个类** + +在 `SensitiveScanState` 类之后、`__all__` 之前追加（文件头部已有所需 import：`String/Integer/BigInteger/Boolean/JSON/Text/Index/UniqueConstraint/mapped_column/Mapped/sa/dt/_updated_at`——若缺 `Text` 则补）： + +```python +# ========================================================================== +# 经营管理后台(2026-09-08 v2 spec,批次A)——fact/宽表/通话小表 + 三张业务表 +# ========================================================================== + + +class FirstReplyFact(Base): + """首响事实表(管理域)——一行 = 一个被回的 1:1 客户消息串(v2 §3.1)。 + + 跨日精确串口径:按会话**全历史**计算,fr_ms = 员工首条回复 ts − 串末条 + 客户消息 ts,day = 串末条所在上海自然日。区别于 msg_daily_stats 首响 + 三列的日切近似(跨夜串整条丢失,v2 §1.1 #1)。只写 1:1(friend)会话 + ——群串语义不成立。由回填 CLI / 4h 滚动 / 夜间全量重算,**不挂写钩子** + (写钩子零改动)。用途:精确 P50/P90、2h/24h 回复率分子、下钻证据。 + """ + + __tablename__ = "first_reply_fact" + + channel: Mapped[str] = mapped_column(String(16), primary_key=True) + owner_account_id: Mapped[str] = mapped_column(String(128), primary_key=True) + peer_id: Mapped[str] = mapped_column(String(128), primary_key=True) + thread_end_ts: Mapped[int] = mapped_column(BigInteger, primary_key=True) + reply_ts: Mapped[int] = mapped_column(BigInteger, primary_key=True) + # 串末条所在上海自然日 yyyymmdd(冗余物化,窗口/趋势索引) + day: Mapped[int] = mapped_column(Integer, nullable=False) + fr_ms: Mapped[int] = mapped_column(BigInteger, nullable=False) + + __table_args__ = (Index("ix_fr_fact_day_channel", "day", "channel"),) + + +class EmployeeMetricDaily(Base): + """员工×日管理指标宽表(v2 §3.2)——1:1 口径,管理域专用。 + + 列来源(v2 §1.1 #2/#5):消息计数/负载列自 msg_daily_stats + (peer_kind='friend' 过滤 + account_map 归并);串类三列与首响日值由 + first_reply_facts.compute_conversation 同源输出(串末条归属日)—— + 回复率分子分母同口径。metric_version 只影响派生列(生命周期), + 默认 'v1'。account_map 映射不上或不在 staff 的 owner(幽灵员工) + 过滤不落表(v2 §1.3 #11)。通话在 employee_call_daily(无渠道维)。 + """ + + __tablename__ = "employee_metric_daily" + + 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) + metric_version: Mapped[str] = mapped_column( + String(16), primary_key=True, default="v1", server_default="v1" + ) + # 消息计数列(1:1 口径) + 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") + ) + # 负载列(active_rooms 为群负载,唯一计入宽表的群维度) + 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") + ) + # 串类列(与 first_reply_fact 同源,按串末条归属日) + cust_threads: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default=sa.text("0") + ) + threads_answered: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default=sa.text("0") + ) + fr_sum_ms: Mapped[int] = mapped_column( + BigInteger, nullable=False, default=0, server_default=sa.text("0") + ) + # 首响日值(趋势图用;评分用周期值从 fact 表窗口函数精确计算) + first_reply_p50_ms: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + first_reply_p90_ms: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + # 生命周期 4 列(距今天数口径,当日持有客户状态分布,当前归属) + new_customer_count: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default=sa.text("0") + ) + cooling_customer_count: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default=sa.text("0") + ) + dormant_customer_count: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default=sa.text("0") + ) + reactivated_customer_count: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default=sa.text("0") + ) + # fact 是否已覆盖当日(宽表生成前置条件;False=串类列可能滞后) + data_complete: Mapped[bool] = mapped_column( + Boolean, nullable=False, default=False, server_default=sa.text("0") + ) + computed_at: Mapped[dt.datetime] = mapped_column( + sa.DateTime, nullable=False, server_default=sa.func.now() + ) + + +class EmployeeCallDaily(Base): + """员工×日通话小表(v2 §1.2 #9)——通话无消息渠道维度,独立于宽表。 + + phone_call.channel 是通话类型(phone_call/voice_call)非消息渠道, + 塞进 (day,channel,user) 宽表会在 channel=all 聚合时双计。接通 = + call_duration > 0;followup = 通话后 24h 内该员工任一账号对该客户 + 有消息(Task 0 核实 join 键,默认双渠道 talker_id 匹配)。 + """ + + __tablename__ = "employee_call_daily" + + day: Mapped[int] = mapped_column(Integer, primary_key=True) + user_id: Mapped[str] = mapped_column(String(64), primary_key=True) + call_count: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default=sa.text("0") + ) + call_seconds: Mapped[int] = mapped_column( + BigInteger, nullable=False, default=0, server_default=sa.text("0") + ) + answered_call_count: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default=sa.text("0") + ) + call_followup_count: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default=sa.text("0") + ) + computed_at: Mapped[dt.datetime] = mapped_column( + sa.DateTime, nullable=False, server_default=sa.func.now() + ) + + +class EmployeeScoreConfig(Base): + """评分配置版本(v1 §10.4/v2 §3.3)——config_json 存维度/指标/权重/ + 方向/阈值/样本门槛;激活前须过权重总和与阈值校验(批次B scoring 实现)。 + MVP 无可视化编辑器,受控 CLI 种子维护。""" + + __tablename__ = "employee_score_config" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + version: Mapped[str] = mapped_column(String(32), unique=True, nullable=False) + name: Mapped[str] = mapped_column(String(128), nullable=False) + # draft | active | retired + status: Mapped[str] = mapped_column( + String(16), nullable=False, default="draft", server_default="draft" + ) + effective_from: Mapped[int | None] = mapped_column(Integer, nullable=True) + effective_to: Mapped[int | None] = mapped_column(Integer, nullable=True) + config_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=False) + created_by: Mapped[str | None] = mapped_column(String(64), nullable=True) + approved_by: Mapped[str | None] = mapped_column(String(64), nullable=True) + created_at: Mapped[dt.datetime] = mapped_column( + sa.DateTime, nullable=False, server_default=sa.func.now() + ) + activated_at: Mapped[dt.datetime | None] = mapped_column(sa.DateTime, nullable=True) + + +class EmployeeScoreSnapshot(Base): + """周/月考核快照(v1 §8/v2 §3.3)——不可变:冻结后业务层禁改评分字段; + 重算走 supersedes_id 替代链。同批次×员工唯一。""" + + __tablename__ = "employee_score_snapshot" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + snapshot_batch_id: Mapped[str] = mapped_column(String(64), nullable=False) + # week | month | custom + period_type: Mapped[str] = mapped_column(String(8), nullable=False) + period_start: Mapped[int] = mapped_column(Integer, nullable=False) + period_end: Mapped[int] = mapped_column(Integer, nullable=False) + user_id: Mapped[str] = mapped_column(String(64), nullable=False) + score_config_version: Mapped[str] = mapped_column(String(32), nullable=False) + metric_version: Mapped[str] = mapped_column(String(16), nullable=False) + total_score: Mapped[int | None] = mapped_column(Integer, nullable=True) + grade: Mapped[str | None] = mapped_column(String(2), nullable=True) + # scored | insufficient_sample | incomplete_data + score_status: Mapped[str] = mapped_column( + String(24), nullable=False, default="scored", server_default="scored" + ) + dimension_scores_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + metric_details_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + data_as_of: Mapped[int] = mapped_column(Integer, nullable=False) + generated_at: Mapped[dt.datetime] = mapped_column( + sa.DateTime, nullable=False, server_default=sa.func.now() + ) + frozen_at: Mapped[dt.datetime | None] = mapped_column(sa.DateTime, nullable=True) + frozen_by: Mapped[str | None] = mapped_column(String(64), nullable=True) + supersedes_id: Mapped[int | None] = mapped_column(Integer, nullable=True) + + __table_args__ = ( + sa.UniqueConstraint("snapshot_batch_id", "user_id", + name="uq_score_snapshot_batch_user"), + Index("ix_score_snapshot_period", "period_type", "period_start"), + ) + + +class ManagementFollowup(Base): + """管理辅导闭环(v1 §7.3/v2 §3.3)——来源指标→问题→管理动作→结果。""" + + __tablename__ = "management_followup" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + employee_user_id: Mapped[str] = mapped_column(String(64), nullable=False) + # metric | anomaly | review | manual + source_type: Mapped[str] = mapped_column(String(16), nullable=False) + source_code: Mapped[str | None] = mapped_column(String(64), nullable=True) + source_period_start: Mapped[int | None] = mapped_column(Integer, nullable=True) + source_period_end: Mapped[int | None] = mapped_column(Integer, nullable=True) + source_value_json: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + reason: Mapped[str | None] = mapped_column(sa.Text, nullable=True) + action_type: Mapped[str | None] = mapped_column(String(32), nullable=True) + owner_manager_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + # open | in_progress | resolved | cancelled + status: Mapped[str] = mapped_column( + String(16), nullable=False, default="open", server_default="open" + ) + due_at: Mapped[dt.datetime | None] = mapped_column(sa.DateTime, nullable=True) + result: Mapped[str | None] = mapped_column(sa.Text, nullable=True) + employee_feedback: Mapped[str | None] = mapped_column(sa.Text, nullable=True) + created_by: Mapped[str | None] = mapped_column(String(64), nullable=True) + created_at: Mapped[dt.datetime] = mapped_column( + sa.DateTime, nullable=False, server_default=sa.func.now() + ) + updated_at: Mapped[dt.datetime] = _updated_at() + closed_at: Mapped[dt.datetime | None] = mapped_column(sa.DateTime, nullable=True) + + __table_args__ = ( + Index("ix_followup_emp_status_due", "employee_user_id", "status", "due_at"), + Index("ix_followup_owner_status", "owner_manager_id", "status"), + ) +``` + +`__all__` 按字母序插入：`"EmployeeCallDaily"`, `"EmployeeMetricDaily"`, `"EmployeeScoreConfig"`, `"EmployeeScoreSnapshot"`, `"FirstReplyFact"`, `"ManagementFollowup"`。 + +- [ ] **Step 5: 跑测试确认通过** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_management_models.py -v +``` + +期望：8 passed。 + +- [ ] **Step 6: 全量回归** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m pytest tests/ -q +``` + +期望：全绿（现有 ~575 + 新 8）。 + +- [ ] **Step 7: 提交** + +```bash +git add yunke-server/app/db/models.py yunke-server/tests/conftest.py yunke-server/tests/test_management_models.py +git commit -m "feat(management): 批次A六表模型——fact/宽表/通话小表+评分配置/快照/辅导" +``` + +--- + +### Task 2: Alembic migration + +**Files:** +- Create: `yunke-server/migrations/versions/b7c8d9e0f1a2_management_phase1_tables.py` + +- [ ] **Step 1: 确认当前 head** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m alembic heads +``` + +期望 `e6f7a8b9c0d1`（若不同，把迁移文件的 `down_revision` 改成实际 head）。 + +- [ ] **Step 2: 写迁移文件** + +照 `f0a1b2c3d4e5_analytics_module_tables.py` 的幂等守卫模式（`sa.inspect` 先查后建；**不在 migration 里碰 message/msg_daily_stats 数据**）： + +```python +"""经营管理后台批次A:6 张新表(fact/宽表/通话小表/评分配置/快照/辅导) + +Revision ID: b7c8d9e0f1a2 +Revises: e6f7a8b9c0d1 +Create Date: 2026-09-08 + +spec: docs/superpowers/plans/2026-09-08-yunke-management-console-implementation-v2.md §3。 +幂等:sa.inspect 先查,存在即跳过;SQLite/MySQL 通用。回填走 CLI +(first-reply-facts / employee-metrics),migration 不扫数据。 +""" +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "b7c8d9e0f1a2" +down_revision: str | Sequence[str] | None = "e6f7a8b9c0d1" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_TABLES = [ + ("first_reply_fact", lambda: op.create_table( + "first_reply_fact", + sa.Column("channel", sa.String(length=16), primary_key=True), + sa.Column("owner_account_id", sa.String(length=128), primary_key=True), + sa.Column("peer_id", sa.String(length=128), primary_key=True), + sa.Column("thread_end_ts", sa.BigInteger(), primary_key=True), + sa.Column("reply_ts", sa.BigInteger(), primary_key=True), + sa.Column("day", sa.Integer(), nullable=False), + sa.Column("fr_ms", sa.BigInteger(), nullable=False), + )), + ("employee_metric_daily", lambda: op.create_table( + "employee_metric_daily", + 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("metric_version", sa.String(length=16), primary_key=True, + server_default="v1"), + 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("cust_threads", sa.Integer(), nullable=False, server_default=sa.text("0")), + sa.Column("threads_answered", sa.Integer(), nullable=False, server_default=sa.text("0")), + sa.Column("fr_sum_ms", sa.BigInteger(), nullable=False, server_default=sa.text("0")), + sa.Column("first_reply_p50_ms", sa.BigInteger(), nullable=True), + sa.Column("first_reply_p90_ms", sa.BigInteger(), nullable=True), + sa.Column("new_customer_count", sa.Integer(), nullable=False, server_default=sa.text("0")), + sa.Column("cooling_customer_count", sa.Integer(), nullable=False, server_default=sa.text("0")), + sa.Column("dormant_customer_count", sa.Integer(), nullable=False, server_default=sa.text("0")), + sa.Column("reactivated_customer_count", sa.Integer(), nullable=False, server_default=sa.text("0")), + sa.Column("data_complete", sa.Boolean(), nullable=False, server_default=sa.text("0")), + sa.Column("computed_at", sa.DateTime(), nullable=False, server_default=sa.func.now()), + )), + ("employee_call_daily", lambda: op.create_table( + "employee_call_daily", + sa.Column("day", sa.Integer(), primary_key=True), + sa.Column("user_id", sa.String(length=64), primary_key=True), + sa.Column("call_count", sa.Integer(), nullable=False, server_default=sa.text("0")), + sa.Column("call_seconds", sa.BigInteger(), nullable=False, server_default=sa.text("0")), + sa.Column("answered_call_count", sa.Integer(), nullable=False, server_default=sa.text("0")), + sa.Column("call_followup_count", sa.Integer(), nullable=False, server_default=sa.text("0")), + sa.Column("computed_at", sa.DateTime(), nullable=False, server_default=sa.func.now()), + )), + ("employee_score_config", lambda: op.create_table( + "employee_score_config", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("version", sa.String(length=32), nullable=False, unique=True), + sa.Column("name", sa.String(length=128), nullable=False), + sa.Column("status", sa.String(length=16), nullable=False, server_default="draft"), + sa.Column("effective_from", sa.Integer(), nullable=True), + sa.Column("effective_to", sa.Integer(), nullable=True), + sa.Column("config_json", sa.JSON(), nullable=False), + sa.Column("created_by", sa.String(length=64), nullable=True), + sa.Column("approved_by", sa.String(length=64), nullable=True), + sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.func.now()), + sa.Column("activated_at", sa.DateTime(), nullable=True), + )), + ("employee_score_snapshot", lambda: op.create_table( + "employee_score_snapshot", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("snapshot_batch_id", sa.String(length=64), nullable=False), + sa.Column("period_type", sa.String(length=8), nullable=False), + sa.Column("period_start", sa.Integer(), nullable=False), + sa.Column("period_end", sa.Integer(), nullable=False), + sa.Column("user_id", sa.String(length=64), nullable=False), + sa.Column("score_config_version", sa.String(length=32), nullable=False), + sa.Column("metric_version", sa.String(length=16), nullable=False), + sa.Column("total_score", sa.Integer(), nullable=True), + sa.Column("grade", sa.String(length=2), nullable=True), + sa.Column("score_status", sa.String(length=24), nullable=False, server_default="scored"), + sa.Column("dimension_scores_json", sa.JSON(), nullable=True), + sa.Column("metric_details_json", sa.JSON(), nullable=True), + sa.Column("data_as_of", sa.Integer(), nullable=False), + sa.Column("generated_at", sa.DateTime(), nullable=False, server_default=sa.func.now()), + sa.Column("frozen_at", sa.DateTime(), nullable=True), + sa.Column("frozen_by", sa.String(length=64), nullable=True), + sa.Column("supersedes_id", sa.Integer(), nullable=True), + sa.UniqueConstraint("snapshot_batch_id", "user_id", + name="uq_score_snapshot_batch_user"), + )), + ("management_followup", lambda: op.create_table( + "management_followup", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("employee_user_id", sa.String(length=64), nullable=False), + sa.Column("source_type", sa.String(length=16), nullable=False), + sa.Column("source_code", sa.String(length=64), nullable=True), + sa.Column("source_period_start", sa.Integer(), nullable=True), + sa.Column("source_period_end", sa.Integer(), nullable=True), + sa.Column("source_value_json", sa.JSON(), nullable=True), + sa.Column("reason", sa.Text(), nullable=True), + sa.Column("action_type", sa.String(length=32), nullable=True), + sa.Column("owner_manager_id", sa.String(length=64), nullable=True), + sa.Column("status", sa.String(length=16), nullable=False, server_default="open"), + sa.Column("due_at", sa.DateTime(), nullable=True), + sa.Column("result", sa.Text(), nullable=True), + sa.Column("employee_feedback", sa.Text(), nullable=True), + sa.Column("created_by", sa.String(length=64), 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()), + sa.Column("closed_at", sa.DateTime(), nullable=True), + )), +] + +_INDEXES = [ + ("first_reply_fact", "ix_fr_fact_day_channel", ["day", "channel"]), + ("employee_score_snapshot", "ix_score_snapshot_period", ["period_type", "period_start"]), + ("management_followup", "ix_followup_emp_status_due", + ["employee_user_id", "status", "due_at"]), + ("management_followup", "ix_followup_owner_status", ["owner_manager_id", "status"]), +] + + +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) + for name, create in _TABLES: + if not insp.has_table(name): + create() + # 索引守卫独立于建表(MySQL DDL 逐句自动提交;建表成功但索引失败的 + # 中断重跑要能补上索引) + for table, index, cols in _INDEXES: + if not _has_index(bind, table, index): + op.create_index(index, table, cols) + + +def downgrade() -> None: + bind = op.get_bind() + insp = sa.inspect(bind) + for table in ("management_followup", "employee_score_snapshot", + "employee_score_config", "employee_call_daily", + "employee_metric_daily", "first_reply_fact"): + if insp.has_table(table): + op.drop_table(table) +``` + +注意：`updated_at` 无 alembic 服务端 onupdate（模型 `_updated_at()` 是 Python 侧 onupdate，迁移里静态 now() 即可——与 f0a1b2c3d4e5 同做法）。 + +- [ ] **Step 3: 本地库升级实测** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m alembic upgrade head +``` + +期望：输出包含 `Running upgrade ... -> b7c8d9e0f1a2`。再跑一次（幂等）应无操作。 + +- [ ] **Step 4: 验证表存在 + downgrade 冒烟** + +```bash +docker exec -it yunke-mysql-1 mysql -uyunke -pyunke yunke -e "SHOW TABLES LIKE '%employee%'; SHOW TABLES LIKE '%first_reply%'; SHOW TABLES LIKE '%management%';" +cd yunke-server && .venv/Scripts/python.exe -m alembic downgrade -1 && .venv/Scripts/python.exe -m alembic upgrade head +``` + +期望：6 张表存在；downgrade 后消失、再 upgrade 恢复。 + +- [ ] **Step 5: 全量测试 + 提交** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m pytest tests/ -q +git add yunke-server/migrations/versions/b7c8d9e0f1a2_management_phase1_tables.py +git commit -m "feat(management): 批次A建表迁移——六表幂等守卫/索引独立补齐/downgrade只删新表" +``` + +--- + +### Task 3: first_reply_facts 跨日精确算法 + 落库 + +**Files:** +- Create: `yunke-server/app/storage/first_reply_facts.py` +- Test: `yunke-server/tests/test_first_reply_facts.py` + +- [ ] **Step 1: 写判别性测试（先失败）** + +新建 `tests/test_first_reply_facts.py`： + +```python +"""first_reply_fact 跨日精确串算法判别测试(批次A Task3)。 + +四个核心判据(v2 §1.1 #1/#5、§9): +1. 跨夜串必须产生样本(日切口径丢失的场景); +2. 多日未回串只计一次、归属末条日(日切会按天重复计——分母膨胀); +3. 同源:threads_answered ≡ fact 行数(按归属日); +4. is_send=None 不参与串判定(与 _thread_stats 一致)。 +""" +import datetime as dt +from zoneinfo import ZoneInfo + +import sqlalchemy as sa + +from app.db.models import FirstReplyFact, Message, MsgDailyStats +from app.storage.first_reply_facts import ( + all_friend_pairs, + compute_conversation, + recompute_pairs, +) + +_SH = ZoneInfo("Asia/Shanghai") + + +def _ts(*args) -> int: + return int(dt.datetime(*args, tzinfo=_SH).timestamp() * 1000) + + +# ---------------------------- 纯函数 ---------------------------- + +def test_overnight_thread_recovered(): + """客户 23:00 发→员工次日 9:00 回:fr_ms≈10h,归属 D1(v2 §1.1 #1)。""" + facts, agg = compute_conversation( + [(_ts(2026, 9, 1, 23, 0), False), (_ts(2026, 9, 2, 9, 0), True)] + ) + assert len(facts) == 1 + assert facts[0].fr_ms == 10 * 3600 * 1000 + assert facts[0].day == 20260901 + assert agg[20260901].answered == 1 + + +def test_multiday_open_thread_counts_once_at_end_day(): + """客户消息跨 D1/D2 无回复:1 个串归属 D2;日切口径会 D1、D2 各计 1。""" + facts, agg = compute_conversation( + [(_ts(2026, 9, 1, 10, 0), False), (_ts(2026, 9, 1, 20, 0), False), + (_ts(2026, 9, 2, 9, 0), False)] + ) + assert facts == [] + assert 20260901 not in agg + assert agg[20260902].threads == 1 and agg[20260902].answered == 0 + + +def test_reply_closes_thread_and_opens_new(): + """员工回复后客户再发=新串;串内多条客户消息取末条为 end。""" + facts, agg = compute_conversation( + [(_ts(2026, 9, 1, 10, 0), False), (_ts(2026, 9, 1, 10, 5), False), + (_ts(2026, 9, 1, 10, 30), True), # 回复1:fr=30-10:05=25min + (_ts(2026, 9, 1, 11, 0), False), (_ts(2026, 9, 1, 12, 0), True)] + ) + assert len(facts) == 2 + assert facts[0].fr_ms == 25 * 60 * 1000 + assert agg[20260901].threads == 2 and agg[20260901].answered == 2 + + +def test_unknown_direction_ignored(): + facts, agg = compute_conversation( + [(_ts(2026, 9, 1, 10, 0), False), (_ts(2026, 9, 1, 10, 10), None), + (_ts(2026, 9, 1, 10, 20), True)] + ) + assert len(facts) == 1 # None 不打断串:仍视为对 10:00 串的回复 + + +def test_staff_message_outside_thread_is_noop(): + facts, agg = compute_conversation( + [(_ts(2026, 9, 1, 9, 0), True), (_ts(2026, 9, 1, 10, 0), True)] + ) + assert facts == [] and agg == {} + + +def test_unsorted_rows_raise_or_documented(): + """调用方契约:rows 须升序。乱序输入行为未定义——文档断言 + 防御排序 + 在 load_conversation 里做(与 summary.py 惯例一致)。""" + # 纯函数不做防御(性能);落库路径负责排序。此处仅锁定契约注释。 + import app.storage.first_reply_facts as m + assert "升序" in m.compute_conversation.__doc__ + + +# ---------------------------- 落库 ---------------------------- + +async def _seed_conversation(session, *, channel="wechat", owner="wx1", + peer="p1", rows): + for i, (ts, is_send) in enumerate(rows): + session.add(Message(msg_id=f"{peer}-{i}", channel=channel, + owner_account_id=owner, talker_id=peer, + timestamp=ts, is_send=is_send, type="text", + text="x")) + session.add(MsgDailyStats(channel=channel, owner_account_id=owner, + peer_kind="friend", peer_id=peer, + day=20260901, send_cnt=1, recv_cnt=1)) + await session.commit() + + +async def test_recompute_pairs_persists_and_idempotent(make_session): + s = make_session + await _seed_conversation( + s, rows=[(_ts(2026, 9, 1, 23, 0), False), (_ts(2026, 9, 2, 9, 0), True)]) + n = await recompute_pairs(s, [("wechat", "wx1", "p1")]) + await s.commit() + assert n == 1 + rows = (await s.execute(sa.select(FirstReplyFact))).scalars().all() + assert len(rows) == 1 and rows[0].fr_ms == 10 * 3600 * 1000 + # 幂等:重跑不重复 + await recompute_pairs(s, [("wechat", "wx1", "p1")]) + await s.commit() + assert len((await s.execute(sa.select(FirstReplyFact))).scalars().all()) == 1 + + +async def test_recompute_pairs_replaces_stale(make_session): + """会话追加消息后重算,旧样本被替换(delete+insert 语义)。""" + s = make_session + await _seed_conversation( + s, rows=[(_ts(2026, 9, 1, 10, 0), False), (_ts(2026, 9, 1, 11, 0), True)]) + await recompute_pairs(s, [("wechat", "wx1", "p1")]) + await s.commit() + s.add(Message(msg_id="p1-new", channel="wechat", owner_account_id="wx1", + talker_id="p1", timestamp=_ts(2026, 9, 1, 12, 0), + is_send=False, type="text", text="y")) + await s.commit() + await recompute_pairs(s, [("wechat", "wx1", "p1")]) + await s.commit() + rows = (await s.execute(sa.select(FirstReplyFact))).scalars().all() + assert len(rows) == 1 # 新串未回,只剩旧样本 + + +async def test_all_friend_pairs_only_friend(make_session): + s = make_session + s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="room", peer_id="r1", day=20260901)) + s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="p1", day=20260901)) + await s.commit() + assert await all_friend_pairs(s) == [("wechat", "wx1", "p1")] +``` + +- [ ] **Step 2: 跑测试确认失败** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_first_reply_facts.py -v +``` + +期望：`ModuleNotFoundError: app.storage.first_reply_facts`。 + +- [ ] **Step 3: 实现 first_reply_facts.py** + +```python +"""first_reply_fact 维护层(管理域,批次A)——跨日精确串算法。 + +区别于 app.storage.summary._thread_stats 的日切近似(v2 §1.1 #1): +按会话**全历史**计算,跨夜串(客户 23:00 → 员工次日 9:00 回)不再丢失。 +**不挂写钩子**(v2 §3.1):由回填 CLI / 4h 滚动 / 夜间全量重算,写钩子 +零改动。只处理 1:1(friend)会话——群串语义不成立。 + +compute_conversation 同时输出每日串聚合(threads/answered/fr_sum_ms, +按串末条归属日)——employee_metric_daily 的串类列与 fact 行**同源** +(回复率分子分母一致,v2 §1.1 #5)。 +""" +from __future__ import annotations + +import logging +from collections import namedtuple + +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import FirstReplyFact, Message, MsgDailyStats +from app.storage.summary import day_key + +logger = logging.getLogger(__name__) + +# 一个被回串的首响样本;day = 串末条客户消息所在上海自然日 +FactSample = namedtuple("FactSample", "thread_end_ts reply_ts fr_ms day") +# 某会话某日的串聚合(分子分母同源) +DayAgg = namedtuple("DayAgg", "threads answered fr_sum_ms") + + +class _DayAcc: + __slots__ = ("threads", "answered", "fr_sum_ms") + + def __init__(self) -> None: + self.threads = 0 + self.answered = 0 + self.fr_sum_ms = 0 + + +def compute_conversation(rows: list) -> tuple[list[FactSample], dict[int, DayAgg]]: + """跨日精确串算法。rows = [(ts, is_send)] 该会话全历史,**须按 ts 升序**。 + + - 串 = 客户连续消息(is_send=False)不被员工消息(is_send=True)打断的 + 段;is_send=None 跳过(不打断也不起串)——与 _thread_stats 判定一致。 + - 每个被回串一个样本:fr_ms = 员工首条回复 ts − 串**末条**客户消息 ts, + 归属 day = 串末条所在上海自然日(跨夜串不再丢失)。 + - 会话末尾仍未闭合的串:计入当日 threads(回复率分母),不产生样本。 + """ + facts: list[FactSample] = [] + acc: dict[int, _DayAcc] = {} + in_thread = False + end_ts: int | None = None + for ts, is_send in rows: + if is_send is False: + in_thread = True + end_ts = int(ts) + elif is_send is True: + if in_thread and end_ts is not None: + fr = int(ts) - end_ts + d = day_key(end_ts) + facts.append(FactSample(end_ts, int(ts), fr, d)) + a = acc.setdefault(d, _DayAcc()) + a.threads += 1 + a.answered += 1 + a.fr_sum_ms += fr + in_thread = False + end_ts = None + # is_send=None:跳过 + if in_thread and end_ts is not None: # 会话末尾未闭合串 + a = acc.setdefault(day_key(end_ts), _DayAcc()) + a.threads += 1 + return facts, {d: DayAgg(a.threads, a.answered, a.fr_sum_ms) + for d, a in acc.items()} + + +async def load_conversation( + session: AsyncSession, channel: str, owner: str, peer: str +) -> list: + """拉取该 1:1 会话全历史 (timestamp, is_send),Python 侧排序(串算法 + 需严格时序——与 summary.recompute_days 同惯例,避免依赖 SQL 排序)。 + + 公有命名:employee_metrics 宽表生成复用同一加载口径(同源原则)。""" + rows = ( + await session.execute( + sa.select(Message.timestamp, Message.is_send).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), + ) + ) + ).all() + return sorted(rows, key=lambda r: r[0]) + + +async def recompute_pairs( + session: AsyncSession, pairs: list[tuple[str, str, str]] +) -> int: + """重算给定 friend 会话的 fact 行:按会话 delete+insert,幂等。 + + pairs = [(channel, owner_account_id, peer_id)]。返回会话数。 + """ + for channel, owner, peer in sorted(pairs): + rows = await load_conversation(session, channel, owner, peer) + facts, _agg = compute_conversation(rows) + await session.execute( + sa.delete(FirstReplyFact).where( + FirstReplyFact.channel == channel, + FirstReplyFact.owner_account_id == owner, + FirstReplyFact.peer_id == peer, + ) + ) + for f in facts: + session.add( + FirstReplyFact( + channel=channel, + owner_account_id=owner, + peer_id=peer, + thread_end_ts=f.thread_end_ts, + reply_ts=f.reply_ts, + day=f.day, + fr_ms=f.fr_ms, + ) + ) + return len(pairs) + + +async def all_friend_pairs(session: AsyncSession) -> list[tuple[str, str, str]]: + """全部 1:1 会话(friend 行 distinct)——全量回填用。""" + rows = ( + await session.execute( + sa.select( + MsgDailyStats.channel, + MsgDailyStats.owner_account_id, + MsgDailyStats.peer_id, + ).where(MsgDailyStats.peer_kind == "friend") + ) + ).all() + return sorted({(r[0], r[1], r[2]) for r in rows}) + + +async def pairs_active_in_days( + session: AsyncSession, days +) -> list[tuple[str, str, str]]: + """窗口内有消息的 friend 会话(4h 滚动重算用;批次F 接调度)。""" + rows = ( + await session.execute( + sa.select( + MsgDailyStats.channel, + MsgDailyStats.owner_account_id, + MsgDailyStats.peer_id, + ).where( + MsgDailyStats.peer_kind == "friend", + MsgDailyStats.day.in_(list(days)), + ) + ) + ).all() + return sorted({(r[0], r[1], r[2]) for r in rows}) + + +async def recompute_all(session: AsyncSession) -> int: + """全量重建(回填/夜间自愈,幂等)。""" + return await recompute_pairs(session, await all_friend_pairs(session)) +``` + +- [ ] **Step 4: 跑测试确认通过 + 全量回归** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_first_reply_facts.py -v +cd yunke-server && .venv/Scripts/python.exe -m pytest tests/ -q +``` + +期望：10 passed；全量绿。 + +- [ ] **Step 5: 提交** + +```bash +git add yunke-server/app/storage/first_reply_facts.py yunke-server/tests/test_first_reply_facts.py +git commit -m "feat(management): 首响事实表跨日精确算法——跨夜串不丢/串末条归属日/同源日聚合" +``` + +--- + +### Task 4: 生命周期判定纯函数 + +**Files:** +- Create: `yunke-server/app/analytics/lifecycle.py` +- Test: `yunke-server/tests/test_lifecycle.py` + +- [ ] **Step 1: 写测试（先失败）** + +新建 `tests/test_lifecycle.py`： + +```python +"""客户生命周期判定纯函数测试(批次A Task4)。 + +口径(v2 §1.1 #3 + v1 §21 默认,已用户确认):新客 7 天、沉睡 30 天、 +历史≥5 条才算沉睡(对齐 churn real min_msgs=5)。analytics 现有页 +(churn/客户资产)暂不切换(口径差异 v2 §5 标注),不动 575 既有测试。 +""" +from app.analytics.lifecycle import NEW_DAYS, DORMANT_DAYS, MIN_HISTORY_MSGS, classify, day_add + + +def test_constants_locked_to_confirmed_defaults(): + assert NEW_DAYS == 7 and DORMANT_DAYS == 30 and MIN_HISTORY_MSGS == 5 + + +def test_new_customer(): + # 首次沟通在 7 天窗口内 + assert classify(first_day=20260901, last_day=20260903, prev_day=None, + as_of=20260905, hist_msgs=2) == "new" + + +def test_reactivated_requires_gap_over_30d(): + # 当日回流且前次间隔>30 天 + assert classify(first_day=20260101, last_day=20260905, prev_day=20260801, + as_of=20260905, hist_msgs=50) == "reactivated" + # 间隔仅 10 天 → 只是 active + assert classify(first_day=20260101, last_day=20260905, prev_day=20260827, + as_of=20260905, hist_msgs=50) == "active" + + +def test_active_cooling_dormant_boundaries(): + f, m = 20260101, 100 + assert classify(f, 20260830, None, 20260905, m) == "active" # 6 天 + assert classify(f, 20260829, None, 20260905, m) == "cooling" # 7 天 + assert classify(f, 20260807, None, 20260905, m) == "cooling" # 29 天 + assert classify(f, 20260806, None, 20260905, m) == "dormant" # 30 天 + + +def test_dormant_requires_history(): + # ≥30 天无互动但历史<5 条:不计入任何状态(None) + assert classify(20260101, 20260806, None, 20260905, 4) is None + assert classify(20260101, 20260806, None, 20260905, 5) == "dormant" + + +def test_no_interaction_returns_none(): + assert classify(None, None, None, 20260905, 0) is None + + +def test_day_add(): + assert day_add(20260831, 1) == 20260901 + assert day_add(20260901, -7) == 20260825 +``` + +- [ ] **Step 2: 跑测试确认失败** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_lifecycle.py -v +``` + +期望：`ModuleNotFoundError: app.analytics.lifecycle`。 + +- [ ] **Step 3: 实现 lifecycle.py** + +```python +"""客户生命周期判定(管理域,批次A)——距今天数口径(v2 §1.1 #3 决议)。 + +阈值取已确认默认(v1 §21,用户拍板):新客 7 天、沉睡 30 天;历史互动 +门槛 MIN_HISTORY_MSGS=5 对齐 analytics churn real 的 min_msgs=5。 +analytics 现有页(churn/客户资产)**暂不切换**到此模块——两套数字的 +差异在 metric-definitions 标注(v2 §5);后续统一时再接,避免动 575 +个既有测试。 +""" +from __future__ import annotations + +import datetime as dt + +NEW_DAYS = 7 +DORMANT_DAYS = 30 +MIN_HISTORY_MSGS = 5 + +STATES = ("new", "active", "cooling", "dormant", "reactivated") + + +def _to_date(day: int) -> dt.date: + return dt.date(day // 10000, (day // 100) % 100, day % 100) + + +def day_add(day: int, delta_days: int) -> int: + """yyyymmdd 键加减天数。""" + d = _to_date(day) + dt.timedelta(days=delta_days) + return d.year * 10000 + d.month * 100 + d.day + + +def days_between(day_a: int, day_b: int) -> int: + """b − a 的天数(输入 yyyymmdd 键)。""" + return (_to_date(day_b) - _to_date(day_a)).days + + +def classify( + first_day: int | None, + last_day: int | None, + prev_day: int | None, + as_of: int, + hist_msgs: int, +) -> str | None: + """as_of 日的客户状态;输入均为互动日键(yyyymmdd,上海自然日)。 + + - first_day: 该客户全历史首次互动日;last_day: ≤as_of 的最近互动日; + prev_day: last_day 之前的一次互动日(无则 None);hist_msgs: 历史总消息数。 + - 返回 None = 未激活(first/last 缺)或历史不足的长期静默(不计入任何列)。 + + 判定优先级:new → reactivated → active → cooling → dormant。 + """ + if first_day is None or last_day is None: + return None + if first_day >= day_add(as_of, -(NEW_DAYS - 1)): + return "new" + if ( + last_day == as_of + and prev_day is not None + and days_between(prev_day, last_day) > DORMANT_DAYS + ): + return "reactivated" + gap = days_between(last_day, as_of) + if gap < NEW_DAYS: + return "active" + if gap < DORMANT_DAYS: + return "cooling" + if hist_msgs >= MIN_HISTORY_MSGS: + return "dormant" + return None +``` + +- [ ] **Step 4: 跑测试通过 + 提交** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_lifecycle.py -v +git add yunke-server/app/analytics/lifecycle.py yunke-server/tests/test_lifecycle.py +git commit -m "feat(management): 生命周期判定纯函数——距今天数口径/7天新客/30天沉睡/历史门槛5条" +``` + +--- + +### Task 5: employee_call_daily 通话小表生成 + +**Files:** +- Create: `yunke-server/app/storage/employee_metrics.py`（本任务先写通话部分） +- Test: `yunke-server/tests/test_employee_metrics.py` + +- [ ] **Step 1: 写测试（先失败）** + +新建 `tests/test_employee_metrics.py`： + +```python +"""employee_call_daily / employee_metric_daily 生成判别测试(批次A Task5/6)。 + +判据(v2 §1.1 #2/#5、§1.2 #9、§1.3 #11): +- 通话无渠道维,独立小表;接通=duration>0;followup=24h 内任一账号有消息; +- 幽灵员工(owner 映射不上/不在 staff)过滤; +- 宽表消息计数 friend-only;串类与 fact 同源。 +""" +import datetime as dt +from zoneinfo import ZoneInfo + +import sqlalchemy as sa + +from app.db.models import ( + EmployeeCallDaily, + EmployeeMetricDaily, + FirstReplyFact, + Message, + MsgDailyStats, + PhoneCall, + Staff, + WechatAccount, +) +from app.storage.employee_metrics import recompute_call_days + +_SH = ZoneInfo("Asia/Shanghai") + + +def _ts(*args) -> int: + return int(dt.datetime(*args, tzinfo=_SH).timestamp() * 1000) + + +async def _seed_org(session): + session.add(Staff(user_id="u1", name="张三", is_on_job=True)) + session.add(WechatAccount(wx_id="wx1", user_id="u1")) + await session.commit() + + +async def test_call_daily_aggregates(make_session): + s = make_session + await _seed_org(s) + s.add(PhoneCall(call_id="c1", channel="voice_call", user_id="u1", + call_number="p1", start_call_time=_ts(2026, 9, 1, 10, 0), + call_duration=120)) + s.add(PhoneCall(call_id="c2", channel="voice_call", user_id="u1", + call_number="p2", start_call_time=_ts(2026, 9, 1, 15, 0), + call_duration=0)) # 未接通 + # c1 后 2h 员工有跟进消息(多账号语义:u1 只有 wx1) + s.add(Message(msg_id="f1", channel="wechat", owner_account_id="wx1", + talker_id="p1", timestamp=_ts(2026, 9, 1, 12, 0), + is_send=True, type="text", text="跟进")) + await s.commit() + n = await recompute_call_days(s, [20260901]) + await s.commit() + assert n == 1 + row = (await s.execute(sa.select(EmployeeCallDaily))).scalar_one() + assert row.call_count == 2 + assert row.call_seconds == 120 # duration 单位按 Task0 核实,默认秒 + assert row.answered_call_count == 1 + assert row.call_followup_count == 1 + + +async def test_call_daily_ignores_unknown_staff(make_session): + s = make_session + await _seed_org(s) + s.add(PhoneCall(call_id="c9", channel="phone_call", user_id="ghost", + call_number="p9", start_call_time=_ts(2026, 9, 1, 10, 0), + call_duration=60)) + await s.commit() + await recompute_call_days(s, [20260901]) + await s.commit() + assert (await s.execute( + sa.select(sa.func.count()).select_from(EmployeeCallDaily))).scalar_one() == 0 +``` + +- [ ] **Step 2: 跑测试确认失败** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_employee_metrics.py -v +``` + +期望：`ImportError: recompute_call_days`。 + +- [ ] **Step 3: 实现通话小表生成（employee_metrics.py 第一部分）** + +```python +"""employee_call_daily / employee_metric_daily 维护层(管理域,批次A)。 + +数据流(v2 §4): + msg_daily_stats(friend) + first_reply_facts 同源扫描 + customer 生命周期 + → employee_metric_daily (day, channel, user, metric_version) + phone_call(按 user_id+start_call_time 日聚合) + → employee_call_daily (day, user)——通话无渠道维(v2 §1.2 #9) + +幽灵员工过滤(v2 §1.3 #11):owner 映射不上或不在 staff 表的不落表。 +重算语义:按 day delete+insert,幂等;夜间全量 + (批次F)4h 滚动。 +""" +from __future__ import annotations + +import logging + +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import ( + EmployeeCallDaily, + Message, + PhoneCall, + Staff, +) +from app.storage.staff_summary import account_map +from app.storage.summary import day_bounds_ms + +logger = logging.getLogger(__name__) + + +async def known_user_ids(session: AsyncSession) -> set[str]: + """员工主数据 user_id 集合(staff 表;Task0 已核实 ≈41 行)。""" + rows = (await session.execute(sa.select(Staff.user_id))).scalars().all() + return {u for u in rows if u} + + +async def _followup_exists( + session: AsyncSession, + owner_accounts: list[str], + peer: str, + call_ts_ms: int, + window_ms: int = 86_400_000, +) -> bool: + """通话后 24h 内员工任一账号对该客户有 1:1 消息(双渠道匹配—— + voice_call 行无法从列判别微信/WA,Task0 核实后可收窄)。""" + if not owner_accounts or not peer: + return False + n = ( + await session.execute( + sa.select(sa.func.count()).select_from(Message).where( + Message.owner_account_id.in_(owner_accounts), + Message.talker_id == peer, + sa.or_(Message.room_id.is_(None), Message.room_id == ""), + Message.timestamp.is_not(None), + Message.timestamp > call_ts_ms, + Message.timestamp <= call_ts_ms + window_ms, + ) + ) + ).scalar_one() + return bool(n) + + +async def recompute_call_days(session: AsyncSession, days) -> int: + """重算给定日的通话小表:按 day delete+insert,幂等。返回写入行数。""" + # 员工→全部账号(双渠道),followup 判定用 + acct = {ch: await account_map(session, ch) for ch in ("wechat", "whatsapp")} + rev: dict[str, list[str]] = {} + for amap in acct.values(): + for owner, uid in amap.items(): + rev.setdefault(uid, []).append(owner) + + written = 0 + for day in sorted(set(days)): + lo, hi = day_bounds_ms(day) + calls = ( + await session.execute( + sa.select( + PhoneCall.user_id, + PhoneCall.call_number, + PhoneCall.start_call_time, + PhoneCall.call_duration, + ).where( + PhoneCall.start_call_time.is_not(None), + PhoneCall.start_call_time >= lo, + PhoneCall.start_call_time < hi, + ) + ) + ).all() + await session.execute( + sa.delete(EmployeeCallDaily).where(EmployeeCallDaily.day == day) + ) + # duration 单位按 Task0 核实结论调整(默认秒;若毫秒在此换算) + agg: dict[str, dict] = {} + for uid, number, ts, dur in calls: + if not uid: + continue + a = agg.setdefault( + uid, {"cnt": 0, "sec": 0, "answered": 0, "followup": 0} + ) + a["cnt"] += 1 + a["sec"] += int(dur or 0) + if (dur or 0) > 0: + a["answered"] += 1 + for uid, a in sorted(agg.items()): + followup = 0 + for number, ts, dur in ( + (c[1], c[2], c[3]) + for c in calls + if c[0] == uid and (c[3] or 0) > 0 + ): + if await _followup_exists(session, rev.get(uid, []), number, ts): + followup += 1 + a["followup"] = followup + session.add( + EmployeeCallDaily( + day=day, + user_id=uid, + call_count=a["cnt"], + call_seconds=a["sec"], + answered_call_count=a["answered"], + call_followup_count=a["followup"], + ) + ) + written += 1 + return written +``` + + +- [ ] **Step 4: 跑测试通过** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_employee_metrics.py -v +``` + +期望：2 passed。 + +- [ ] **Step 5: 提交** + +```bash +git add yunke-server/app/storage/employee_metrics.py yunke-server/tests/test_employee_metrics.py +git commit -m "feat(management): 通话日聚合小表——无渠道维/接通=duration>0/24h任一账号跟进" +``` + +--- + +### Task 6: employee_metric_daily 宽表生成 + +**Files:** +- Modify: `yunke-server/app/storage/employee_metrics.py`（追加宽表部分） +- Test: `yunke-server/tests/test_employee_metrics.py`（追加） + +- [ ] **Step 1: 追加宽表判别测试（先失败）** + +在 `tests/test_employee_metrics.py` 追加： + +```python +from app.storage.employee_metrics import recompute_metric_days # noqa: E402 + + +async def _seed_msg_rows(s, *, day=20260901, channel="wechat", owner="wx1", + kind="friend", peer="p1", send=3, recv=2): + s.add(MsgDailyStats(channel=channel, owner_account_id=owner, peer_kind=kind, + peer_id=peer, day=day, send_cnt=send, recv_cnt=recv)) + await s.commit() + + +async def test_metric_daily_friend_only_caliber(make_session): + """口径判别(v2 §1.1 #2 防线):room 行不进消息计数/串列,只进 active_rooms。""" + s = make_session + await _seed_org(s) + await _seed_msg_rows(s, send=3, recv=2, kind="friend", peer="p1") + await _seed_msg_rows(s, send=10, recv=0, kind="room", peer="r1") + # friend 会话给一条消息供串扫描(无消息则无串) + s.add(Message(msg_id="p1-0", channel="wechat", owner_account_id="wx1", + talker_id="p1", timestamp=_ts(2026, 9, 1, 10, 0), + is_send=False, type="text", text="q")) + await s.commit() + await recompute_metric_days(s, [20260901]) + await s.commit() + row = (await s.execute(sa.select(EmployeeMetricDaily))).scalar_one() + assert row.msg_sent == 3 and row.msg_recv == 2 # room 的 10 不进 + assert row.active_rooms == 1 + assert row.cust_threads == 1 and row.threads_answered == 0 # 未回复串 + + +async def test_metric_daily_ghost_filtered(make_session): + s = make_session + await _seed_org(s) + await _seed_msg_rows(s, owner="wx-ghost", peer="pg") + await recompute_metric_days(s, [20260901]) + await s.commit() + assert (await s.execute( + sa.select(sa.func.count()).select_from(EmployeeMetricDaily))).scalar_one() == 0 + + +async def test_metric_daily_same_source_as_fact(make_session): + """同源判别(v2 §1.1 #5 防线): + 1) threads_answered ≡ fact 行数(同归属日); + 2) 多日未回串宽表 threads < 日切 cust_threads 之和(分母不再膨胀); + 3) 幂等。""" + from app.storage.first_reply_facts import recompute_pairs + s = make_session + await _seed_org(s) + # 客户 D1 10:00、D1 20:00、D2 09:00 三条消息,始终未回 → 1 个跨日串 + rows = [(_ts(2026, 9, 1, 10, 0), False), (_ts(2026, 9, 1, 20, 0), False), + (_ts(2026, 9, 2, 9, 0), False)] + for i, (ts, is_send) in enumerate(rows): + s.add(Message(msg_id=f"g{i}", channel="wechat", owner_account_id="wx1", + talker_id="pg", timestamp=ts, is_send=is_send, + type="text", text="x")) + # 日切口径的 msg_daily_stats:D1、D2 各计一个未回串(供对照) + await _seed_msg_rows(s, day=20260901, peer="pg", send=0, recv=2) + await _seed_msg_rows(s, day=20260902, peer="pg", send=0, recv=1) + await s.commit() + await recompute_pairs(s, [("wechat", "wx1", "pg")]) + await s.commit() + await recompute_metric_days(s, [20260901, 20260902]) + await s.commit() + got = {r.day: r for r in ( + await s.execute(sa.select(EmployeeMetricDaily))).scalars().all()} + # 1) fact 行数=0(未回),宽表 threads_answered 也=0 且同源 + assert (await s.execute( + sa.select(sa.func.count()).select_from(FirstReplyFact))).scalar_one() == 0 + # 2) 跨日串只归属 D2;日切 D1+D2=2,宽表合计=1 + assert got[20260901].cust_threads == 0 + assert got[20260902].cust_threads == 1 + # 3) 幂等 + await recompute_metric_days(s, [20260901, 20260902]) + await s.commit() + assert len((await s.execute( + sa.select(EmployeeMetricDaily))).scalars().all()) == 2 + + +async def test_metric_daily_lifecycle_counts(make_session): + """生命周期列:cooling/dormant/reactivated 判定进宽表。""" + s = make_session + await _seed_org(s) + from app.db.models import Customer + # 客户 pc:D1 有互动后静默(历史≥5 条) → as_of=D1+40 为 dormant; + # D41 回流 → 该日 reactivated + s.add(Customer(friend_id="pc", channel="wechat", owner_account_id="wx1", + delete_flag=0)) + hist = [] + for h in range(9, 17): # D1 8 条消息(≥5) + hist.append((_ts(2026, 9, 1, h, 0), h % 2 == 0)) + for i, (ts, is_send) in enumerate(hist): + s.add(Message(msg_id=f"lc{i}", channel="wechat", owner_account_id="wx1", + talker_id="pc", timestamp=ts, is_send=is_send, + type="text", text="x")) + await _seed_msg_rows(s, day=20260901, peer="pc", send=4, recv=4) + await s.commit() + # D41(10-11)客户回流一条 + s.add(Message(msg_id="lc-b", channel="wechat", owner_account_id="wx1", + talker_id="pc", timestamp=_ts(2026, 10, 11, 9, 0), + is_send=False, type="text", text="back")) + await _seed_msg_rows(s, day=20261011, peer="pc", send=0, recv=1) + await s.commit() + from app.storage.first_reply_facts import recompute_pairs + await recompute_pairs(s, [("wechat", "wx1", "pc")]) + await recompute_metric_days(s, [20261010, 20261011]) + await s.commit() + got = {r.day: r for r in ( + await s.execute(sa.select(EmployeeMetricDaily))).scalars().all()} + assert got[20261010].dormant_customer_count == 1 # 40 天静默+历史 8 条 + assert got[20261011].reactivated_customer_count == 1 # 当日回流且 gap>30 +``` + +- [ ] **Step 2: 跑测试确认失败** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_employee_metrics.py -v +``` + +期望：`ImportError: recompute_metric_days`。 + +- [ ] **Step 3: 实现宽表生成（追加到 employee_metrics.py）** + +```python +import math # (文件头补充) +from bisect import bisect_right + +from app.analytics.lifecycle import classify +from app.db.models import Customer, EmployeeMetricDaily, MsgDailyStats +from app.storage.first_reply_facts import ( + compute_conversation, + load_conversation, + pairs_active_in_days, +) + +METRIC_VERSION = "v1" + + +def _percentile(vals: list[int], q: float) -> int | None: + """最近邻秩分位 ceil(q·n);空列表→None(与 0 严格区分)。""" + if not vals: + return None + s = sorted(vals) + idx = min(len(s) - 1, max(0, math.ceil(q * len(s)) - 1)) + return s[idx] + + +async def _lifecycle_daily( + session: AsyncSession, + days: list[int], + channel: str, + amap: dict[str, str], + valid_users: set[str], +) -> dict[int, dict[str, dict[str, int]]]: + """逐日 {day: {user: {state: count}}}——customer 当前归属 + 互动日 sweep。 + + 数据源:msg_daily_stats friend 行全历史(互动日/条数)。16k 客户 × 窗口 + 天数 bisect,夜间全量约秒级;4h tick 只算当日+3 天。 + """ + rows = ( + await session.execute( + sa.select( + MsgDailyStats.peer_id, MsgDailyStats.day, + MsgDailyStats.send_cnt, MsgDailyStats.recv_cnt, + ).where( + MsgDailyStats.channel == channel, + MsgDailyStats.peer_kind == "friend", + ) + ) + ).all() + days_map: dict[str, set[int]] = {} + cnt: dict[str, int] = {} + for peer, day, snd, rcv in rows: + days_map.setdefault(peer, set()).add(day) + cnt[peer] = cnt.get(peer, 0) + int(snd or 0) + int(rcv or 0) + interaction = {p: sorted(ds) for p, ds in days_map.items()} + + custs = ( + await session.execute( + sa.select(Customer.owner_account_id, Customer.friend_id).where( + Customer.channel == channel, + sa.or_(Customer.delete_flag.is_(None), Customer.delete_flag == 0), + Customer.owner_account_id.is_not(None), + ) + ) + ).all() + peer_to_user: dict[str, str] = {} + for owner, peer in custs: + uid = amap.get(owner or "") + if uid and uid in valid_users and peer: + peer_to_user[str(peer)] = uid + + out: dict[int, dict[str, dict[str, int]]] = {d: {} for d in days} + for peer, days_list in interaction.items(): + uid = peer_to_user.get(peer) + if uid is None: + continue + first = days_list[0] + hist = cnt.get(peer, 0) + for d in days: + idx = bisect_right(days_list, d) - 1 + if idx < 0: + continue + last = days_list[idx] + prev = days_list[idx - 1] if idx > 0 else None + state = classify(first, last, prev, d, hist) + if state: + out[d].setdefault(uid, {}).setdefault(state, 0) + out[d][uid][state] += 1 + return out + + +async def recompute_metric_days( + session: AsyncSession, + days, + channels: tuple[str, ...] = ("wechat", "whatsapp"), +) -> int: + """重算给定日的宽表:按 (day, channel) delete+insert,幂等。返回写入行数。 + + 串类列与首响日值:扫窗口内活跃 friend 会话的**全历史**(每会话一次), + 与 first_reply_fact 同一 compute_conversation——分子分母同源(v2 §1.1 #5)。 + """ + days = sorted(set(days)) + valid_users = await known_user_ids(session) + pairs = await pairs_active_in_days(session, days) + pairs_by_channel: dict[str, list] = {} + for p in pairs: + pairs_by_channel.setdefault(p[0], []).append((p[1], p[2])) + + written = 0 + for channel in channels: + amap = await account_map(session, channel) + # ---- 串类+分位:逐会话全历史一次,聚合到 (day, user) ---- + thread_cols: dict[tuple[int, str], dict] = {} + frs: dict[tuple[int, str], list[int]] = {} + day_set = set(days) + for owner, peer in pairs_by_channel.get(channel, []): + uid = amap.get(owner) + if uid is None or uid not in valid_users: + continue + rows = await load_conversation(session, channel, owner, peer) + facts, agg = compute_conversation(rows) + for d, a in agg.items(): + if d not in day_set: + continue + key = (d, uid) + t = thread_cols.setdefault( + key, {"threads": 0, "answered": 0, "fr": 0} + ) + t["threads"] += a.threads + t["answered"] += a.answered + t["fr"] += a.fr_sum_ms + frs.setdefault(key, []).extend( + f.fr_ms for f in facts if f.day == d + ) + + # ---- 生命周期 ---- + life = await _lifecycle_daily(session, days, channel, amap, valid_users) + + # ---- 消息计数+负载:msg_daily_stats(算法无关列) ---- + for day in days: + mds = ( + 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() + await session.execute( + sa.delete(EmployeeMetricDaily).where( + EmployeeMetricDaily.day == day, + EmployeeMetricDaily.channel == channel, + EmployeeMetricDaily.metric_version == METRIC_VERSION, + ) + ) + agg: dict[str, dict] = {} + for owner, kind, peer, snd, rcv in mds: + uid = amap.get(owner) + if uid is None or uid not in valid_users: + continue # 幽灵过滤 + a = agg.setdefault( + uid, + {"sent": 0, "recv": 0, "friends": set(), "rooms": set()}, + ) + # 消息计数列 = 1:1 口径(v2 §1.1 #2):room 只进 active_rooms + if kind == "friend": + a["sent"] += int(snd or 0) + a["recv"] += int(rcv or 0) + a["friends"].add(peer) + elif kind == "room": + a["rooms"].add(peer) + day_life = life.get(day, {}) + # 行并集:消息源 ∪ 生命周期源——纯生命周期日(员工当日无消息 + # 但持有降温/沉睡客户)也要出宽表行,否则客户经营维度漏天 + all_uids = sorted(set(agg) | set(day_life)) + for uid in all_uids: + a = agg.get( + uid, {"sent": 0, "recv": 0, "friends": set(), "rooms": set()} + ) + key = (day, uid) + t = thread_cols.get(key, {"threads": 0, "answered": 0, "fr": 0}) + lv = day_life.get(uid, {}) + session.add( + EmployeeMetricDaily( + day=day, + channel=channel, + user_id=uid, + metric_version=METRIC_VERSION, + msg_sent=a["sent"], + msg_recv=a["recv"], + active_friends=len(a["friends"]), + active_rooms=len(a["rooms"]), + cust_threads=t["threads"], + threads_answered=t["answered"], + fr_sum_ms=t["fr"], + first_reply_p50_ms=_percentile(frs.get(key, []), 0.5), + first_reply_p90_ms=_percentile(frs.get(key, []), 0.9), + new_customer_count=lv.get("new", 0), + cooling_customer_count=lv.get("cooling", 0), + dormant_customer_count=lv.get("dormant", 0), + reactivated_customer_count=lv.get("reactivated", 0), + data_complete=True, + ) + ) + written += 1 + return written +``` + +注意实现要点（执行者核对）： +- `load_conversation` 在 Task 3 已公有定义——宽表生成 import 复用，保证串类列与 fact 行同源同加载口径； +- `_percentile` 的最近邻秩公式与测试对齐（若写 `math.ceil` 版本更清晰可替换，测试只断言行为）； +- 幂等删除条件带 `metric_version == METRIC_VERSION`，其他版本行不受影响。 + +- [ ] **Step 4: 跑测试通过 + 全量回归** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_employee_metrics.py -v +cd yunke-server && .venv/Scripts/python.exe -m pytest tests/ -q +``` + +期望：6 passed（通话 2 + 宽表 4）；全量绿。 + +- [ ] **Step 5: 提交** + +```bash +git add yunke-server/app/storage/employee_metrics.py yunke-server/tests/test_employee_metrics.py +git commit -m "feat(management): 员工指标宽表——1:1口径/串类与fact同源/生命周期列/幽灵过滤" +``` + +--- + +### Task 7: 两条回填 CLI + +**Files:** +- Modify: `yunke-server/app/sync/cli.py` + +- [ ] **Step 1: 注册子命令（`_build_parser` 的 sensitive-scan 之后）** + +```python + # ``first-reply-facts`` ---------------------------------------------- + p_fr = sub.add_parser( + "first-reply-facts", + help="Rebuild first_reply_fact (cross-day exact threads).", + description=( + "Full rebuild of the first-reply fact table from the message " + "table (cross-day exact thread algorithm — overnight samples " + "not lost, unlike msg_daily_stats day-cut). Deploy backfill + " + "nightly self-heal step. Idempotent." + ), + ) + + # ``employee-metrics`` ------------------------------------------------ + p_emp = sub.add_parser( + "employee-metrics", + help="Rebuild employee_metric_daily + employee_call_daily.", + description=( + "Full rebuild of the management wide table (per-day x channel " + "x staff, 1:1 caliber) and the call daily table. Must run " + "AFTER first-reply-facts (thread columns share the same " + "source). Deploy backfill + nightly self-heal. Idempotent." + ), + ) +``` + +- [ ] **Step 2: 实现命令函数（`_cmd_sensitive_scan` 之后）** + +```python +async def _cmd_first_reply_facts(args: argparse.Namespace) -> int: + """全量重建 first_reply_fact(跨日精确串;部署回填/夜间自愈,幂等)。""" + settings = get_settings() + engine, session_maker = await _init_db(settings.db_url) + try: + from app.storage.first_reply_facts import recompute_all + + async with session_maker() as session: + pairs = await recompute_all(session) + await session.commit() + print(f"[first-reply-facts] rebuilt {pairs} conversations") + return 0 + finally: + await engine.dispose() + + +async def _cmd_employee_metrics(args: argparse.Namespace) -> int: + """全量重建 employee_metric_daily + employee_call_daily(幂等)。""" + settings = get_settings() + engine, session_maker = await _init_db(settings.db_url) + try: + import sqlalchemy as sa + + from app.db.models import MsgDailyStats + from app.storage.employee_metrics import ( + recompute_call_days, + recompute_metric_days, + ) + + async with session_maker() as session: + days = ( + await session.execute(sa.select(MsgDailyStats.day).distinct()) + ).scalars().all() + rows = await recompute_metric_days(session, list(days)) + calls = await recompute_call_days(session, list(days)) + await session.commit() + print( + f"[employee-metrics] rebuilt {rows} wide rows, {calls} call rows" + ) + return 0 + finally: + await engine.dispose() +``` + +`main()` 分发表追加两行（`sensitive-scan` 之后）： + +```python + if args.command == "first-reply-facts": + return asyncio.run(_cmd_first_reply_facts(args)) + if args.command == "employee-metrics": + return asyncio.run(_cmd_employee_metrics(args)) +``` + +- [ ] **Step 3: 冒烟验证（对本地 Docker MySQL 真跑）** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m app.sync.cli first-reply-facts +cd yunke-server && .venv/Scripts/python.exe -m app.sync.cli employee-metrics +``` + +期望：退出码 0，输出会话数（预期数千）与行数（宽表预期 = 活跃 day×channel×员工组合，数千~数万）。 + +- [ ] **Step 4: 幂等复跑 + 抽查** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m app.sync.cli employee-metrics # 复跑行数应一致 +docker exec -it yunke-mysql-1 mysql -uyunke -pyunke yunke -e " +SELECT COUNT(*) AS fact_rows FROM first_reply_fact; +SELECT COUNT(*) AS wide_rows, SUM(threads_answered) AS answered FROM employee_metric_daily; +SELECT COUNT(*) FROM first_reply_fact f +WHERE NOT EXISTS (SELECT 1 FROM msg_daily_stats m + WHERE m.peer_kind='friend' AND m.channel=f.channel + AND m.owner_account_id=f.owner_account_id AND m.peer_id=f.peer_id AND m.day=f.day); +" +``` + +期望：第三个查询返回 0（每个 fact 样本所在会话×日都有 msg_daily_stats 行——落库链路自洽）。 + +- [ ] **Step 5: 全量测试 + 提交** + +```bash +cd yunke-server && .venv/Scripts/python.exe -m pytest tests/ -q +git add yunke-server/app/sync/cli.py +git commit -m "feat(management): 两条回填CLI——first-reply-facts/employee-metrics全量重建幂等" +``` + +--- + +### Task 8: 本地全量回填实测 + 对账 + DEPLOY.md 注记 + +**Files:** +- Modify: `docs/DEPLOY.md`（§7 回填清单追加两条命令） +- Modify: `docs/superpowers/plans/2026-09-08-management-batch-a-verification.md`（追加实测数据） + +- [ ] **Step 1: 计时全量回填** + +```bash +cd yunke-server && time .venv/Scripts/python.exe -m app.sync.cli first-reply-facts +cd yunke-server && time .venv/Scripts/python.exe -m app.sync.cli employee-metrics +``` + +记录耗时到核实文档（预期：fact 分钟级、宽表分钟级；超过 10 分钟时记下瓶颈会话，评估尾部增量优化——**不阻塞批次验收**）。 + +- [ ] **Step 2: 对账抽查（v2 §9 数据核验的批次A子集）** + +用 Task 0 回测名单中的 2 名员工 × 3 个日期： + +```sql +-- 1) 消息计数列 ↔ msg_daily_stats friend 聚合(必须一致) +SELECT m.day, SUM(m.send_cnt) sent, SUM(m.recv_cnt) recv +FROM msg_daily_stats m JOIN wechat_account w ON m.owner_account_id=w.wx_id +WHERE m.peer_kind='friend' AND w.user_id='' AND m.day IN (20260901, 20260815, 20260801) +GROUP BY m.day; +SELECT day, msg_sent, msg_recv FROM employee_metric_daily +WHERE user_id='' AND channel='wechat' AND day IN (20260901, 20260815, 20260801); +-- 2) 串类列 ↔ fact 行数(必须一致——同源) +SELECT f.day, COUNT(*) FROM first_reply_fact f JOIN wechat_account w ON f.owner_account_id=w.wx_id +WHERE w.user_id='' AND f.day IN (...) GROUP BY f.day; +SELECT day, threads_answered FROM employee_metric_daily WHERE user_id='' AND ...; +-- 3) 串类列 vs 日切口径(允许 fact ≥ 日切——差集=跨夜串,记录差值) +SELECT day, cust_threads FROM stat_daily_staff WHERE user_id='' AND ...; +``` + +把对账结果（一致/差异+解释）写入核实文档。**发现不一致且无法用口径差异解释时停下排查，不得带病进入批次B**。 + +- [ ] **Step 3: DEPLOY.md §7 追加回填注记** + +在现有 daily-stats/staff-daily-stats 回填注记之后追加（格式对齐现有条目）： + +```markdown +- 管理后台批次A新增（部署带迁移 b7c8d9e0f1a2 后各跑一次，幂等）： + - `python -m app.sync.cli first-reply-facts`（跨日精确首响事实；本地实测耗时见批次A核实文档） + - `python -m app.sync.cli employee-metrics`（员工宽表+通话小表；**必须在 first-reply-facts 之后跑**） + - 服务器注意：backend 容器不自动跑 alembic，须手动 `sudo docker exec alembic upgrade head` +``` + +- [ ] **Step 4: 提交 + 批次收尾** + +```bash +git add docs/DEPLOY.md docs/superpowers/plans/2026-09-08-management-batch-a-verification.md +git commit -m "docs(management): 批次A回填实测+对账记录+DEPLOY部署注记" +``` + +--- + +## 批次A验收清单（全部勾完才算完成） + +- [ ] 6 张表在本地 MySQL 存在，migration 幂等 + downgrade 干净； +- [ ] `first-reply-facts` 全量回填成功且复跑幂等； +- [ ] `employee-metrics` 全量回填成功且复跑幂等； +- [ ] 跨夜串判别测试通过（23:00→次日 9:00 产生 ~10h 样本）； +- [ ] friend-only 口径判别测试通过（room 不进消息计数/串列）； +- [ ] 同源判别测试通过（threads_answered ≡ fact 行数；多日未回串分母不膨胀）； +- [ ] 幽灵员工过滤测试通过； +- [ ] 对账抽查：消息计数列与 msg_daily_stats friend 聚合一致；串类列差异全部可用跨夜串解释； +- [ ] 全量测试套件绿（现有 ~575 + 新增 ~25）； +- [ ] 核实文档完整（五项核实 + 回测名单 + 回填耗时 + 对账记录）； +- [ ] 写钩子零改动（`git diff` 确认 summary.py/repos.py 未动）。 + +## 批次B-F 的后续计划 + +批次A合入后另出计划（避免远期细节空转）： +- **批次B**（v2 Task 3-5）：scoring/anomalies/管理 API + 配置种子 CLI； +- **批次C**（Task 6-7）：前端导航+管理驾驶舱； +- **批次D**（Task 8-9）：员工总览+详情； +- **批次E**（Task 10-11）：辅导闭环+考核快照； +- **批次F**（Task 12-13）：analytics_tick 第三步、full_sync 第 13/14 步、scheduler 镜像重建、回测试运行。