浅念
27f91e560d
五项核实→六表模型/migration→跨日精确fact算法→生命周期纯函数→ 通话小表→宽表(1:1口径/同源/幽灵过滤/纯生命周期日出行)→两条回填CLI→ 本地全量实测对账。判别测试覆盖v2 §9全部防线(跨夜串/friend口径/ 同源/超集/幂等)。批次B-F待批次A合入后另出计划
2072 lines
85 KiB
Markdown
2072 lines
85 KiB
Markdown
# 云客经营管理后台·批次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/<file> -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/<new>_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 "<SQL>"` 执行。
|
||
|
||
- [ ] **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='<UID>' AND m.day IN (20260901, 20260815, 20260801)
|
||
GROUP BY m.day;
|
||
SELECT day, msg_sent, msg_recv FROM employee_metric_daily
|
||
WHERE user_id='<UID>' 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='<UID>' AND f.day IN (...) GROUP BY f.day;
|
||
SELECT day, threads_answered FROM employee_metric_daily WHERE user_id='<UID>' AND ...;
|
||
-- 3) 串类列 vs 日切口径(允许 fact ≥ 日切——差集=跨夜串,记录差值)
|
||
SELECT day, cust_threads FROM stat_daily_staff WHERE user_id='<UID>' 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 <backend> 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 镜像重建、回测试运行。
|