Compare commits
13 Commits
d42895337d
...
99773f69b9
| Author | SHA1 | Date | |
|---|---|---|---|
 浅念
|
99773f69b9 | ||
|
浅念
|
9dc22cd1e6 | ||
|
浅念
|
58d274096d | ||
|
浅念
|
0b0dd0808b | ||
|
浅念
|
373fb266a6 | ||
|
浅念
|
2b1f91762b | ||
|
浅念
|
e0c764997f | ||
|
浅念
|
399ecc680f | ||
|
浅念
|
2b516362cd | ||
|
浅念
|
b8c2bef6df | ||
|
浅念
|
5f6bfe1ff6 | ||
|
浅念
|
27f91e560d | ||
|
浅念
|
a866ea8abd |
@@ -0,0 +1,172 @@
|
||||
# 批次A Task 0 核实记录：五项核实 + 回测名单
|
||||
|
||||
- **日期**：2026-09-09
|
||||
- **方式**：`docker exec -i yunke-mysql-1 mysql -uyunke -pyunke yunke -e "<SQL>"`（容器名以 `docker ps` 实测为准，确认为 `yunke-mysql-1`）
|
||||
- **用途**：本记录的全部结论直接约束批次A（数据底座）后续任务的实现细节；偏差处显式标注「影响 Task N」。任务编号引用 [2026-09-08-management-console-batch-a-data-foundation.md](./2026-09-08-management-console-batch-a-data-foundation.md)。
|
||||
- **数据快照范围**：`stat_daily_staff` / `msg_daily_stats` 覆盖 20251024 ~ 20260907；`phone_call.start_call_time` 覆盖 2025-10-18 ~ 2026-09-05。
|
||||
|
||||
---
|
||||
|
||||
## ① 员工主数据（staff 表）
|
||||
|
||||
| 指标 | 实测 |
|
||||
|---|---|
|
||||
| staff 总数 | **41**（与计划假设 ≈41 一致） |
|
||||
| `is_on_job=1` | 41（全部在职） |
|
||||
| `wechat_account` DISTINCT user_id | 41 |
|
||||
| `whatsapp_account` DISTINCT user_id | 20 |
|
||||
| 微信幽灵账号（user_id 不在 staff） | **0** |
|
||||
| WA 幽灵账号（补充核实） | **0** |
|
||||
| `stat_daily_staff` DISTINCT user_id | 40（其 user_id 全部在 staff 内；员工 `FD685A5620184156BE75A8D73FB72F64` 无任何消息统计行） |
|
||||
|
||||
**结论**：账号表 → staff 无幽灵映射，`known_user_ids` 过滤（v2 §1.3 #11）零损耗，不会意外丢行。两个渠道的 account_map 产出 user_id 均能落到 staff。**无偏差，Task 5/6 按计划实现即可。**
|
||||
|
||||
---
|
||||
|
||||
## ② phone_call 列与单位
|
||||
|
||||
表结构关键列：`call_id`(PK)、`channel`、`start_call_time`(bigint)、`call_duration`(int)、`call_status`(int)、`incoming_call`(int)、`user_id`、`call_number`、`fetched_at`、`updated_at`。
|
||||
|
||||
### channel 分布
|
||||
|
||||
| channel | 行数 |
|
||||
|---|---|
|
||||
| `voice_call` | 2,504 |
|
||||
| `wa_voice_call` | 686 |
|
||||
| 合计 | **3,190**（与已知 3,190 条一致） |
|
||||
|
||||
### call_duration 单位：**秒**（确认）
|
||||
|
||||
| 指标 | 实测 |
|
||||
|---|---|
|
||||
| 总和 | 263,651 秒 = **73.24 小时**（对照已知约 73 小时，吻合） |
|
||||
| 均值 | 82.65 秒（对照预期均值 ≈82 秒，吻合） |
|
||||
|
||||
### call_status × duration 交叉
|
||||
|
||||
| call_status | 行数 | duration=0 | duration>0 | 语义判定 |
|
||||
|---|---|---|---|---|
|
||||
| 1 | 1,508 | 31 | 1,477 | **接通**（含 31 条 0 秒即挂） |
|
||||
| 2 | 1,682 | 1,682 | 0 | **未接通**（未接/拒接/取消，与方向无关） |
|
||||
|
||||
### incoming_call 分布
|
||||
|
||||
| incoming_call | 行数 | 语义 |
|
||||
|---|---|---|
|
||||
| 1 | 1,764 | 呼入 |
|
||||
| 0 | 1,426 | 呼出 |
|
||||
|
||||
### start_call_time：**毫秒 epoch**（确认）
|
||||
|
||||
13 位整数；`FROM_UNIXTIME(x/1000)` 解出 max = 2026-09-05 16:00:55（合理近期），按秒解释则超范围。min = 2025-10-18。与 `message.timestamp` 毫秒同口径。
|
||||
|
||||
### call_number 形态（按 channel 完全可分）
|
||||
|
||||
- `voice_call` → 微信 ID：`wxid_*` 或自定义微信号（`csf7169708`、`yuanye751380`、`Maple_1988` 等）
|
||||
- `wa_voice_call` → WA JID：`2348182388888@s.whatsapp.net` 等
|
||||
|
||||
### 与计划默认假设的偏差（重要）
|
||||
|
||||
1. **计划假设 #4「PhoneCall.channel 是通话类型（phone_call/voice_call）非消息渠道」不完全成立**：实测无 `phone_call`（蜂窝）值，且 `wa_voice_call` **能**从 channel 列判别 WA。计划 Task 5 注释「voice_call 行无法从列判别微信/WA」不成立。**影响 Task 5**：`_followup_exists` 可按渠道收窄——`voice_call` 只查该员工 wechat 账号的 1:1 消息、`wa_voice_call` 只查 whatsapp 账号（当前双渠道任一匹配的写法也不会错——wxid 不会撞 WA JID，仅多一次无效查询；收窄属优化非必需）。
|
||||
2. `call_duration` 单位=秒、`start_call_time`=毫秒 epoch：**与计划默认一致，Task 5 无需换算**（`day_bounds_ms` 直接用）。
|
||||
3. 接通判定：计划默认 `duration>0`。实测 `call_status=1 AND duration>0` = 1,477 行，`duration>0` 口径与之差 31 行（31 条接通但 0 秒，占 1.0%）。**影响 Task 5**：`answered=dur>0` 口径可用；若追求语义精确可改 `call_status=1`（含 0 秒接通），差异 ≤1%，实现时任选其一并在表注释写明。
|
||||
|
||||
---
|
||||
|
||||
## ③ customer 归属字段填充率
|
||||
|
||||
| 指标 | 实测 |
|
||||
|---|---|
|
||||
| 总行数 | 16,339 |
|
||||
| `user_id` 填充 | 16,339（**100%**） |
|
||||
| `owner_account_id` 填充 | 16,339（**100%**） |
|
||||
| `delete_flag=1` | 53 |
|
||||
|
||||
分渠道：wechat 7,040（owner 填充 7,040）/ whatsapp 9,299（owner 填充 9,299）。
|
||||
|
||||
### 两条归属路径一致性（全表校验，非抽样）
|
||||
|
||||
`customer.user_id` vs `owner_account_id → wechat_account.wx_id / whatsapp_account.uid → user_id`：
|
||||
|
||||
| 渠道 | 校验行数 | 一致 |
|
||||
|---|---|---|
|
||||
| whatsapp | 9,299 | 9,299 |
|
||||
| wechat | 7,040 | 7,040 |
|
||||
|
||||
**结论**：两条路径 100% 等价、零冲突。生命周期归属覆盖面完整。**影响 Task 6**：默认 `owner_account_id → account_map → user_id` 路径确认可用；`customer.user_id` 列可直接当冗余等价路径使用（如需省一次 join，实测安全）。
|
||||
|
||||
补充观察（**影响 Task 3/6 的 peer 对齐**）：**微信侧主体是无后缀裸 ID，后缀形态是少数**。`customer.friend_id` 实测形态分布（全表）：
|
||||
|
||||
| 渠道 | 形态 | 行数 | 占比 |
|
||||
|---|---|---|---|
|
||||
| wechat | **无后缀裸 ID**（`a_515855826`、`a053136a` 等） | 6,779 | **96.3%** |
|
||||
| wechat | `@openim` 后缀 | 261 | 3.7% |
|
||||
| whatsapp | `@s.whatsapp.net` | 9,266 | 99.6% |
|
||||
| whatsapp | `@lid` 后缀 | 33 | 0.4% |
|
||||
|
||||
`message.talker_id` 侧同查（行数口径）：wechat 裸 ID 844,287 / `@openim` 16,873 / `@chatroom` 17,003（群）；whatsapp `@s.whatsapp.net` 72,357 / `@lid` 4,284 / `@g.us` 9,439（群）。即：微信 peer 主键以裸 ID 为准、`@openim` 为少数共存；WA 侧基本全带后缀。Task 3 实现 peer 匹配时按实测形态对齐（两边原值相等才命中），不做猜测性截断/补后缀。
|
||||
|
||||
---
|
||||
|
||||
## ④ 群占比（msg_daily_stats，量化混合口径偏差）
|
||||
|
||||
| channel | peer_kind | 行数 | 消息量(send+recv) | 会话线程(cust_threads) |
|
||||
|---|---|---|---|---|
|
||||
| wechat | friend | 38,343 | 269,074 | 82,879 |
|
||||
| wechat | room | 16,234 | 609,089 | 26,220 |
|
||||
| whatsapp | friend | 11,673 | 77,994 | 14,178 |
|
||||
| whatsapp | room | 866 | 8,086 | 1,402 |
|
||||
| **合计** | | **67,116** | **964,243** | **124,679** |
|
||||
|
||||
占比汇总：
|
||||
|
||||
| 维度 | room 占比 | 说明 |
|
||||
|---|---|---|
|
||||
| 微信消息量 | **69.4%**（609,089/878,163） | 微信消息近七成来自群 |
|
||||
| WA 消息量 | 9.4%（8,086/86,080） | WA 几乎纯 1:1 |
|
||||
| 总消息量 | **64.0%**（617,175/964,243） | |
|
||||
| 总会话线程 | **22.2%**（27,622/124,679） | |
|
||||
|
||||
**结论**：若现有看板首响/回复率不分 `peer_kind`，消息量维度被群污染 64%、线程维度 22%——污染主要来自微信侧（微信消息 69% 是群消息）。批次A `first_reply_fact` / `employee_metric_daily` 强制 `peer_kind='friend'` 过滤是必要的，不能沿用混合口径。
|
||||
|
||||
---
|
||||
|
||||
## ⑤ call_followup join 键（call_number ↔ message.talker_id，1:1 消息）
|
||||
|
||||
| channel | 分母（call_number 非空） | 命中（talker_id 匹配且非群消息） | 命中率 |
|
||||
|---|---|---|---|
|
||||
| `voice_call` | 2,504 | 2,378 | **95.0%** |
|
||||
| `wa_voice_call`（补充核实） | 686 | 681 | **99.3%** |
|
||||
|
||||
**结论**：双渠道命中率均远超 50% 阈值，`call_followup_count` 一期**高置信**，无需降级标记、无需在口径中注记低置信。**影响 Task 5**：24h 跟进判定按计划实现。未命中成因已精确解释（复核补充）：`voice_call` 的 126 条未命中**恰好等于** `call_number` 带 `@chatroom` 后缀的 126 条群通话（实测这 126 条对 1:1 消息全部 0 命中——join 限定 `room_id` 为空，群通话本就不该匹配）；剔除群通话后 1:1 可匹配通话命中率为 **2,378/2,378 = 100%**。`wa_voice_call` 侧后缀分布为 `@s.whatsapp.net` 676 / `@g.us`（群）9 / `@lid` 1，未命中 5 条**全部为** `@g.us` 群通话（逐条列出验证）。
|
||||
|
||||
---
|
||||
|
||||
## 回测员工名单
|
||||
|
||||
选取依据：消息量/WA 量/活跃群排名 + 低样本端（见文末各候选分渠道数据）。覆盖：微信为主、WA 为主、群运营为主、高负载、低样本（前三类与高负载有合并）。
|
||||
|
||||
| # | user_id | 名字 | 覆盖类型 | 关键数据（20251024~20260907） |
|
||||
|---|---|---|---|---|
|
||||
| 1 | `369D257EEC9A4357935FDE19DDD99CAA` | 王云青 | **高负载 + 群运营为主 + 微信为主**（合并） | 消息量 #1（162,740：微信 162,371 / WA 369）；活跃群 #1（4,036）；通话 431+9 |
|
||||
| 2 | `9C0B195A2C234C829FF507E011B4B460` | 田瑞志 | **WA 为主** | WA 消息 9,986 / 总 11,597 = 86%；WA 通话 28 |
|
||||
| 3 | `F99F6557C2904A7E8F4C5A3BD8A597A1` | 张士蓓 | **双通道混合**（WA 量 #1，群 #2） | 微信 91,936 + WA 12,529；活跃群 2,273；通话最多（471 微信 + 97 WA） |
|
||||
| 4 | `DC48798A5EA24DBB8B31A89CB858E08B` | 15639270965 | **纯微信无 WA 行**（验证 WA 缺失路径） | 微信 78,680，`stat_daily_staff` 无 whatsapp 行；通话 3 |
|
||||
| 5 | `946C08639BF84DFEBBF5EC2A607AE1FA` | 18539430257 | **低样本**（验证空数据路径） | 消息 8 条，无群、无通话、无 WA |
|
||||
|
||||
> 名单中的 #1/#3 建议作为 Task 8「本地全量回填实测」的 2 名回测员工（数据最厚、双通道+通话+群全齐）。
|
||||
|
||||
---
|
||||
|
||||
## 偏差汇总（对后续任务的约束清单）
|
||||
|
||||
| # | 偏差/确认 | 影响 |
|
||||
|---|---|---|
|
||||
| 1 | `phone_call.channel` 实测仅 `voice_call`（微信）/`wa_voice_call`（WA）两值，**无蜂窝 `phone_call` 值**；channel 可判别渠道，计划「无法从列判别微信/WA」结论修正 | Task 5（followup 可按渠道收窄，属优化非必需） |
|
||||
| 2 | `call_duration` 单位=**秒**（263,651s=73.24h≈已知 73h）；`start_call_time`=**毫秒 epoch** | Task 5（默认口径正确，无需换算） |
|
||||
| 3 | 接通判定 `dur>0` vs `call_status=1` 差 31 行（1.0%），前者少计 31 条 0 秒接通 | Task 5（任选其一，注释写明） |
|
||||
| 4 | `customer.owner_account_id` 填充 **100%**，与 `user_id` 列全表 100% 等价 | Task 6（默认 owner 路径确认；可直接用 user_id 列省 join） |
|
||||
| 5 | 微信侧 peer ID **主体为无后缀裸 ID**（customer.friend_id 6,779/7,040=96.3%；message.talker_id 行数口径 844,287 裸 vs 16,873 `@openim`），后缀形态（`@openim` 等）为少数；WA 侧基本全带后缀（`@s.whatsapp.net`/`@lid`） | Task 3/6（peer 对齐按实测形态原值匹配，勿截断/勿补后缀） |
|
||||
| 6 | 群消息占总消息量 64%（微信侧 69%） | Task 3/6（friend 过滤必须，佐证 spec 口径） |
|
||||
| 7 | call_number↔talker_id 命中率 voice_call 95.0% / wa_voice_call 99.3% | Task 5（call_followup_count 高置信，不降级） |
|
||||
| 8 | staff 41 行无幽灵账号；`stat_daily_staff` 仅 40 名员工有行 | Task 5/6（known_user_ids 过滤零损耗） |
|
||||
File diff suppressed because it is too large
Load Diff
@@ -39,16 +39,17 @@
|
||||
| 6 | 周期分位数若 Python 拉全窗口 fact 行（月窗最坏 ~8 万行 ≈10MB 传输）不可取 | 改 SQL 窗口函数 `PERCENT_RANK() OVER (PARTITION BY user_id ORDER BY fr_ms)`（MySQL 8.4 / SQLite 3.25+ 双方言支持），传输降为员工数×2 行 |
|
||||
| 7 | fact 4h 增量成本 | 滚动 3 天=近 3 天活跃会话**全历史重算**（最大 friend 会话 5.2 万条 ≈1-2s × 几十活跃会话，4h 一次可接受）；一期取简单正确，预留「尾部增量」优化（历史已闭合串不变） |
|
||||
| 8 | full_sync_all.sh 步数变化 | 12→**14** 步：第 13 步 fact 全量重建、第 14 步宽表全量重建（宽表必须在 fact 之后） |
|
||||
| 9 | **通话列渠道错位**：v2 初稿把通话四列放进 (day, channel, user) 宽表，但 `phone_call.channel` 是**通话类型**（phone_call/voice_call）非消息渠道（models.py docstring 自证「calls are not tied to a messaging channel」）——channel=all 聚合时通话四列双计 | `app/db/models.py` PhoneCall 定义（批次A计划定案） | 通话拆出独立小表 `employee_call_daily`（day, user_id 主键，无渠道维），宽表删去通话四列 |
|
||||
|
||||
### 1.3 P3 级（补充明确/简化）
|
||||
|
||||
| # | 项 | 结论 |
|
||||
|---|---|---|
|
||||
| 9 | 员工主数据来源未核实 | 列入 Task 0 核实（organization 路由背后是独立员工表还是账号表派生） |
|
||||
| 10 | 幽灵员工 | account_map 映射不上的 owner（离职/异常账号），宽表生成时直接**过滤**，不进管理域员工列表（评分场景不适用 analytics 的「读侧标注」策略） |
|
||||
| 11 | 功能开关降级 | 现状单 admin 账号无角色可分，v1 Task 13 的「功能开关」是过度设计；改为**上线顺序控制**（本地先行→服务器后上），RBAC 二/三期（与 v1 §15 一致） |
|
||||
| 12 | fr_ms 边界 | 按 ts 排序保证非负；超长样本（>30 天）保留在 fact，评分按异常裁剪原则处理（上游整合计划 §6.5「极端值不直接放大分数」）：>30 天样本参与分布展示，但不作为 0 分依据，单独列入超时明细 |
|
||||
| 13 | 五维分布图 | 一期只画 3 个可用维度（客户经营/服务质量/有效负载）+ 2 个「待数据」占位（行动执行/风险合规），不暗示完整五维 |
|
||||
| 10 | 员工主数据来源未核实 | 列入 Task 0 核实（默认 = `staff` 表，user_id 主键 + is_on_job，models.py 已有） |
|
||||
| 11 | 幽灵员工 | account_map 映射不上的 owner（离职/异常账号），宽表生成时直接**过滤**，不进管理域员工列表（评分场景不适用 analytics 的「读侧标注」策略） |
|
||||
| 12 | 功能开关降级 | 现状单 admin 账号无角色可分，v1 Task 13 的「功能开关」是过度设计；改为**上线顺序控制**（本地先行→服务器后上），RBAC 二/三期（与 v1 §15 一致） |
|
||||
| 13 | fr_ms 边界 | 按 ts 排序保证非负；超长样本（>30 天）保留在 fact，评分按异常裁剪原则处理（上游整合计划 §6.5「极端值不直接放大分数」）：>30 天样本参与分布展示，但不作为 0 分依据，单独列入超时明细 |
|
||||
| 14 | 五维分布图 | 一期只画 3 个可用维度（客户经营/服务质量/有效负载）+ 2 个「待数据」占位（行动执行/风险合规），不暗示完整五维 |
|
||||
|
||||
---
|
||||
|
||||
@@ -71,7 +72,7 @@
|
||||
|
||||
---
|
||||
|
||||
## 3. 数据库设计：5 张新表
|
||||
## 3. 数据库设计：6 张新表
|
||||
|
||||
### 3.1 `first_reply_fact` 首响事实表（本设计新增）
|
||||
|
||||
@@ -122,12 +123,19 @@ PK: (day, channel, user_id, metric_version)
|
||||
生命周期 4 列（距今天数口径，当日持有客户状态分布，按 customer 当前归属）:
|
||||
new_customer_count, cooling_customer_count,
|
||||
dormant_customer_count, reactivated_customer_count
|
||||
通话 4 列（phone_call 按 user_id+day 直聚，user_id 已是员工 ID 无需转换）:
|
||||
call_count, call_seconds, answered_call_count, call_followup_count
|
||||
元数据:
|
||||
data_complete, computed_at
|
||||
```
|
||||
|
||||
**通话小表 `employee_call_daily`**（§1.2 #9 拆出，无渠道维）：
|
||||
|
||||
```text
|
||||
PK: (day, user_id)
|
||||
列: call_count, call_seconds, answered_call_count, call_followup_count, computed_at
|
||||
数据源: phone_call 按 user_id+start_call_time 日聚合；接通=call_duration>0；
|
||||
followup=通话后 24h 内该员工任一账号对该客户有消息（Task 0 核实 join 键）
|
||||
```
|
||||
|
||||
**相对 v1 的删减与更名**：
|
||||
|
||||
- 6 列可派生不落：`reply_2h/24h_count`、`unreplied_2h/24h_count`（fact 表 fr_ms≤阈值计数 + cust_threads 派生）、`first_reply_count/first_reply_sum_ms`（≡ threads_answered/fr_sum_ms）；
|
||||
@@ -170,7 +178,8 @@ message（按会话全历史，跨日精确算法） [新 storage/first_r
|
||||
│
|
||||
└─→ employee_metric_daily [新 storage/employee_metrics.py：
|
||||
msg 计数(msg_daily_stats,friend) + 串类/分位(fact 同源)
|
||||
+ customer 生命周期 + phone_call]
|
||||
+ customer 生命周期]
|
||||
phone_call → employee_call_daily [通话小表，无渠道维，§1.2 #9]
|
||||
→ 期聚合/四对比（上期/个人基线4周/团队中位数）
|
||||
→ 试算评分 + 异常规则 [management/]
|
||||
→ 周/月快照（不可变） [management/snapshots]
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
"""客户生命周期判定(管理域,批次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);后续统一时再接,避免动既有
|
||||
测试。
|
||||
|
||||
本模块为纯叶子节点,禁止 import app.storage(防止 storage→analytics
|
||||
消费时成环)。
|
||||
"""
|
||||
|
||||
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
|
||||
@@ -19,7 +19,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Response
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api.deps import get_session
|
||||
from app.db.models import Customer, Message
|
||||
from app.db.models import Customer, CustomerAccountLink, Message
|
||||
|
||||
router = APIRouter(prefix="/api/export", tags=["export"])
|
||||
|
||||
@@ -209,22 +209,40 @@ async def export_chat_html_zip(
|
||||
status_code=422, detail=f"一次最多导出 {ZIP_MAX_CONVS} 个会话"
|
||||
)
|
||||
|
||||
# 文件名显示名:批量取一次 Customer(备注 > 昵称 > friend_id)
|
||||
cust_rows = (await session.execute(
|
||||
# 文件名显示名:批量取一次(备注 > 昵称 > friend_id)。优先
|
||||
# customer_account_link per-owner 快照——共享客户的档案行可能挂在别的
|
||||
# 员工名下,按 owner 查档案会查空、zip 文件名回退成 wxid(2026-09-09,
|
||||
# 同 sessions._meta 修复);link 没覆盖的回退档案。
|
||||
link_rows = (await session.execute(
|
||||
sa.select(
|
||||
Customer.friend_id,
|
||||
Customer.friend_nickname,
|
||||
Customer.friend_remark,
|
||||
CustomerAccountLink.friend_id,
|
||||
CustomerAccountLink.friend_nickname,
|
||||
CustomerAccountLink.friend_remark,
|
||||
).where(
|
||||
Customer.channel == channel,
|
||||
Customer.owner_account_id == owner_account_id,
|
||||
Customer.friend_id.in_(ids),
|
||||
CustomerAccountLink.channel == channel,
|
||||
CustomerAccountLink.owner_account_id == owner_account_id,
|
||||
CustomerAccountLink.friend_id.in_(ids),
|
||||
)
|
||||
)).all()
|
||||
names = {
|
||||
r.friend_id: (r.friend_remark or r.friend_nickname or r.friend_id)
|
||||
for r in cust_rows
|
||||
for r in link_rows
|
||||
}
|
||||
missing = [t for t in ids if t not in names]
|
||||
if missing:
|
||||
cust_rows = (await session.execute(
|
||||
sa.select(
|
||||
Customer.friend_id,
|
||||
Customer.friend_nickname,
|
||||
Customer.friend_remark,
|
||||
).where(
|
||||
Customer.channel == channel,
|
||||
Customer.owner_account_id == owner_account_id,
|
||||
Customer.friend_id.in_(missing),
|
||||
)
|
||||
)).all()
|
||||
for r in cust_rows:
|
||||
names[r.friend_id] = (r.friend_remark or r.friend_nickname or r.friend_id)
|
||||
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
|
||||
|
||||
@@ -22,7 +22,13 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api.deps import get_session
|
||||
from app.api.schemas import SessionListOut, SessionOut
|
||||
from app.db.models import ChatGroup, Customer, Message, MsgDailyStats
|
||||
from app.db.models import (
|
||||
ChatGroup,
|
||||
Customer,
|
||||
CustomerAccountLink,
|
||||
Message,
|
||||
MsgDailyStats,
|
||||
)
|
||||
from app.storage.summary import day_key, has_rows as _summary_has_rows
|
||||
|
||||
router = APIRouter(prefix="/api", tags=["sessions"])
|
||||
@@ -117,7 +123,13 @@ async def _summary_rows(
|
||||
|
||||
|
||||
async def _meta(session: AsyncSession, owner: str, channel: str, friend_ids: list[str], room_ids: list[str]):
|
||||
"""查 customer/chat_group 表合并昵称/头像;好友行额外带 delete_flag(is_del 过滤用)。
|
||||
"""查 customer_account_link(主)/customer(兜底)/chat_group 合并昵称/头像。
|
||||
|
||||
好友名优先读 ``customer_account_link`` 的 per-owner 快照——共享客户的
|
||||
档案行(customer 全局单行,last-wins)可能挂在别的员工名下,按 owner 查
|
||||
档案会查空,会话列表就回退显示 wxid(2026-09-09 用户报「有昵称却显示
|
||||
微信号」);link 表每账号一行,回填后都有昵称。link 没覆盖的(空表过渡/
|
||||
老数据)回退档案路径。
|
||||
|
||||
friends 值为 ``(name, avatar, delete_flag)``;群无删除标记,groups 值仍为
|
||||
``(name, avatar)``。
|
||||
@@ -126,12 +138,29 @@ async def _meta(session: AsyncSession, owner: str, channel: str, friend_ids: lis
|
||||
if friend_ids:
|
||||
for fid, remark, nick, real, avatar, delete_flag in (await session.execute(
|
||||
sa.select(
|
||||
Customer.friend_id, Customer.friend_remark, Customer.friend_nickname,
|
||||
Customer.real_name, Customer.head_image_url, Customer.delete_flag,
|
||||
CustomerAccountLink.friend_id, CustomerAccountLink.friend_remark,
|
||||
CustomerAccountLink.friend_nickname, CustomerAccountLink.real_name,
|
||||
CustomerAccountLink.head_image_url, CustomerAccountLink.delete_flag,
|
||||
)
|
||||
.where(
|
||||
CustomerAccountLink.channel == channel,
|
||||
CustomerAccountLink.owner_account_id == owner,
|
||||
CustomerAccountLink.friend_id.in_(friend_ids),
|
||||
)
|
||||
.where(Customer.channel == channel, Customer.owner_account_id == owner, Customer.friend_id.in_(friend_ids))
|
||||
)).all():
|
||||
friends[fid] = (remark or nick or real or fid, avatar, delete_flag)
|
||||
# link 未覆盖的好友回退档案(按 owner 过滤;共享好友在此仍可能查空,
|
||||
# 上层以 wxid 兜底显示)
|
||||
missing = [fid for fid in friend_ids if fid not in friends]
|
||||
if missing:
|
||||
for fid, remark, nick, real, avatar, delete_flag in (await session.execute(
|
||||
sa.select(
|
||||
Customer.friend_id, Customer.friend_remark, Customer.friend_nickname,
|
||||
Customer.real_name, Customer.head_image_url, Customer.delete_flag,
|
||||
)
|
||||
.where(Customer.channel == channel, Customer.owner_account_id == owner, Customer.friend_id.in_(missing))
|
||||
)).all():
|
||||
friends[fid] = (remark or nick or real or fid, avatar, delete_flag)
|
||||
groups: dict[str, tuple[str, str | None]] = {}
|
||||
if room_ids:
|
||||
for rid, name, avatar in (await session.execute(
|
||||
|
||||
@@ -20,7 +20,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api.deps import get_session
|
||||
from app.api.schemas import StaffChatCustomerOut, StaffChatStatsOut, StaffChatStatsSummary
|
||||
from app.db.models import Customer, Message, MsgDailyStats
|
||||
from app.db.models import Customer, CustomerAccountLink, Message, MsgDailyStats
|
||||
from app.storage.summary import day_key, has_rows as _summary_has_rows
|
||||
|
||||
router = APIRouter(prefix="/api", tags=["staff-chat-stats"])
|
||||
@@ -144,22 +144,43 @@ async def _aggregate_summary(
|
||||
|
||||
|
||||
async def _customer_names(session: AsyncSession, owner: str, talker_ids: list[str]) -> dict[str, tuple[str, str, str]]:
|
||||
"""客户名/头像解析:优先 customer_account_link per-owner 快照,档案兜底。
|
||||
|
||||
共享客户的档案行(customer 全局单行,last-wins)可能挂在别的员工名下,
|
||||
按 owner 查档案会查空、明细列表回退显示 wxid(2026-09-09,同 sessions
|
||||
._meta 修复)。
|
||||
"""
|
||||
if not talker_ids:
|
||||
return {}
|
||||
stmt = sa.select(
|
||||
Customer.friend_id,
|
||||
Customer.friend_remark,
|
||||
Customer.friend_nickname,
|
||||
Customer.real_name,
|
||||
Customer.head_image_url,
|
||||
).where(
|
||||
Customer.channel == "wechat",
|
||||
Customer.owner_account_id == owner,
|
||||
Customer.friend_id.in_(talker_ids),
|
||||
)
|
||||
info: dict[str, tuple[str, str, str]] = {}
|
||||
for fid, remark, nick, real, head in (await session.execute(stmt)).all():
|
||||
link_stmt = sa.select(
|
||||
CustomerAccountLink.friend_id,
|
||||
CustomerAccountLink.friend_remark,
|
||||
CustomerAccountLink.friend_nickname,
|
||||
CustomerAccountLink.real_name,
|
||||
CustomerAccountLink.head_image_url,
|
||||
).where(
|
||||
CustomerAccountLink.channel == "wechat",
|
||||
CustomerAccountLink.owner_account_id == owner,
|
||||
CustomerAccountLink.friend_id.in_(talker_ids),
|
||||
)
|
||||
for fid, remark, nick, real, head in (await session.execute(link_stmt)).all():
|
||||
info[fid] = (remark or nick or real or fid, real or "", head or "")
|
||||
missing = [t for t in talker_ids if t not in info]
|
||||
if missing:
|
||||
stmt = sa.select(
|
||||
Customer.friend_id,
|
||||
Customer.friend_remark,
|
||||
Customer.friend_nickname,
|
||||
Customer.real_name,
|
||||
Customer.head_image_url,
|
||||
).where(
|
||||
Customer.channel == "wechat",
|
||||
Customer.owner_account_id == owner,
|
||||
Customer.friend_id.in_(missing),
|
||||
)
|
||||
for fid, remark, nick, real, head in (await session.execute(stmt)).all():
|
||||
info[fid] = (remark or nick or real or fid, real or "", head or "")
|
||||
return info
|
||||
|
||||
|
||||
|
||||
@@ -40,7 +40,14 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api.deps import get_session
|
||||
from app.api.schemas import StaffStatsCountsOut, StaffStatsDetailOut
|
||||
from app.db.models import ChatGroup, Customer, Message, PhoneCall, WechatAccount
|
||||
from app.db.models import (
|
||||
ChatGroup,
|
||||
Customer,
|
||||
CustomerAccountLink,
|
||||
Message,
|
||||
PhoneCall,
|
||||
WechatAccount,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/api", tags=["staff-stats"])
|
||||
|
||||
@@ -96,50 +103,57 @@ async def _new_customers(
|
||||
) -> tuple[int, list[dict[str, Any]]]:
|
||||
"""新客户:add_time 窗内;add_time IS NULL 用首条消息时间兜底。
|
||||
|
||||
读 ``customer_account_link``(2026-09-09):customer 档案全局单行、
|
||||
归属 last-wins,共享客户在其他员工名下会漏计;link 每账号一行,与
|
||||
云客按员工统计的口径一致(数字会比档案口径大)。
|
||||
|
||||
无 ``detail`` 分支——计数模式有意复用明细物化(窗内新增客户通常
|
||||
几十行,``len()`` 即可;COUNT 需 direct+fallback 双查询,不值)。
|
||||
"""
|
||||
base = [Customer.channel == "wechat", Customer.owner_account_id == wxid]
|
||||
base = [
|
||||
CustomerAccountLink.channel == "wechat",
|
||||
CustomerAccountLink.owner_account_id == wxid,
|
||||
]
|
||||
# 兜底口径:每个 add_time 为空的客户,取其与该员工首条消息 min(timestamp)
|
||||
# (LEFT JOIN 走 ix_message_channel_owner_talker_time,HAVING 在 SQL 侧过滤窗)。
|
||||
fallback = (
|
||||
sa.select(
|
||||
Customer.friend_id,
|
||||
Customer.friend_nickname,
|
||||
Customer.friend_remark,
|
||||
Customer.head_image_url,
|
||||
CustomerAccountLink.friend_id,
|
||||
CustomerAccountLink.friend_nickname,
|
||||
CustomerAccountLink.friend_remark,
|
||||
CustomerAccountLink.head_image_url,
|
||||
sa.func.min(Message.timestamp).label("first_ts"),
|
||||
)
|
||||
.outerjoin(
|
||||
Message,
|
||||
sa.and_(
|
||||
Message.channel == "wechat",
|
||||
Message.owner_account_id == Customer.owner_account_id,
|
||||
Message.talker_id == Customer.friend_id,
|
||||
Message.owner_account_id == CustomerAccountLink.owner_account_id,
|
||||
Message.talker_id == CustomerAccountLink.friend_id,
|
||||
Message.timestamp.is_not(None),
|
||||
),
|
||||
)
|
||||
.where(*base, Customer.add_time.is_(None))
|
||||
# GROUP BY 显式列出全部裸选列:friend_id 是 Customer 唯一 PK,其余列
|
||||
# 函数依赖于它→分组结果不变;MySQL 8 FD 推断可容忍裸列,但显式列表
|
||||
# 在 ONLY_FULL_GROUP_BY 下部署安全(SQLite 同样兼容)。
|
||||
.where(*base, CustomerAccountLink.add_time.is_(None))
|
||||
# GROUP BY 显式列出全部裸选列:friend_id 是 link 复合 PK 的尾列,
|
||||
# (owner,friend) 已由 WHERE 钉死,其余列函数依赖于它→分组结果不变;
|
||||
# MySQL ONLY_FULL_GROUP_BY 部署安全(SQLite 同样兼容)。
|
||||
.group_by(
|
||||
Customer.friend_id,
|
||||
Customer.friend_nickname,
|
||||
Customer.friend_remark,
|
||||
Customer.head_image_url,
|
||||
CustomerAccountLink.friend_id,
|
||||
CustomerAccountLink.friend_nickname,
|
||||
CustomerAccountLink.friend_remark,
|
||||
CustomerAccountLink.head_image_url,
|
||||
)
|
||||
.having(sa.func.min(Message.timestamp) >= start)
|
||||
.having(sa.func.min(Message.timestamp) <= end)
|
||||
)
|
||||
# 直接口径:add_time 本身在窗内。
|
||||
direct = sa.select(
|
||||
Customer.friend_id,
|
||||
Customer.friend_nickname,
|
||||
Customer.friend_remark,
|
||||
Customer.head_image_url,
|
||||
Customer.add_time,
|
||||
).where(*base, Customer.add_time >= start, Customer.add_time <= end)
|
||||
CustomerAccountLink.friend_id,
|
||||
CustomerAccountLink.friend_nickname,
|
||||
CustomerAccountLink.friend_remark,
|
||||
CustomerAccountLink.head_image_url,
|
||||
CustomerAccountLink.add_time,
|
||||
).where(*base, CustomerAccountLink.add_time >= start, CustomerAccountLink.add_time <= end)
|
||||
|
||||
rows: list[dict[str, Any]] = [
|
||||
{
|
||||
@@ -237,19 +251,43 @@ async def _red_transfer(
|
||||
talker_ids = [m.talker_id for m in msgs if m.talker_id]
|
||||
nick_map: dict[str, str] = {}
|
||||
if talker_ids:
|
||||
# 昵称优先 link per-owner 快照(共享好友档案行可能挂别的员工名下,
|
||||
# 2026-09-09 同 sessions._meta),档案兜底
|
||||
nick_map = {
|
||||
fid: nick
|
||||
for fid, nick in (
|
||||
await session.execute(
|
||||
sa.select(Customer.friend_id, Customer.friend_nickname).where(
|
||||
Customer.channel == "wechat",
|
||||
Customer.owner_account_id == wxid,
|
||||
Customer.friend_id.in_(talker_ids),
|
||||
sa.select(
|
||||
CustomerAccountLink.friend_id,
|
||||
CustomerAccountLink.friend_nickname,
|
||||
).where(
|
||||
CustomerAccountLink.channel == "wechat",
|
||||
CustomerAccountLink.owner_account_id == wxid,
|
||||
CustomerAccountLink.friend_id.in_(talker_ids),
|
||||
)
|
||||
)
|
||||
).all()
|
||||
if nick
|
||||
}
|
||||
missing = [t for t in talker_ids if t not in nick_map]
|
||||
if missing:
|
||||
nick_map.update(
|
||||
{
|
||||
fid: nick
|
||||
for fid, nick in (
|
||||
await session.execute(
|
||||
sa.select(
|
||||
Customer.friend_id, Customer.friend_nickname
|
||||
).where(
|
||||
Customer.channel == "wechat",
|
||||
Customer.owner_account_id == wxid,
|
||||
Customer.friend_id.in_(missing),
|
||||
)
|
||||
)
|
||||
).all()
|
||||
if nick
|
||||
}
|
||||
)
|
||||
rows: list[dict[str, Any]] = []
|
||||
for m in msgs:
|
||||
raw_name = (m.raw_json or {}).get("name") if m.raw_json else None
|
||||
@@ -273,15 +311,20 @@ async def _flag_customers(
|
||||
flag_cond: sa.ColumnElement,
|
||||
*,
|
||||
detail: bool,
|
||||
model: type,
|
||||
) -> tuple[int, list[dict[str, Any]]]:
|
||||
"""del_customers / blocked_customers 共用:现存量口径,忽略时间窗。
|
||||
|
||||
``model`` 由调用方决定(两者列名同形):del 走 ``CustomerAccountLink``
|
||||
(delete_flag per-relation,共享好友不漏),blocked 走 ``Customer``
|
||||
档案(is_block 未镜像进 link,2026-09-09)。
|
||||
|
||||
无时间字段→明细行 time 用 updated_at 毫秒(仅展示;update 触发即变)。
|
||||
"""
|
||||
conds = [Customer.channel == "wechat", Customer.owner_account_id == wxid, flag_cond]
|
||||
conds = [model.channel == "wechat", model.owner_account_id == wxid, flag_cond]
|
||||
count = (
|
||||
await session.execute(
|
||||
sa.select(sa.func.count()).select_from(Customer).where(*conds)
|
||||
sa.select(sa.func.count()).select_from(model).where(*conds)
|
||||
)
|
||||
).scalar_one()
|
||||
if not detail:
|
||||
@@ -289,13 +332,13 @@ async def _flag_customers(
|
||||
rows = (
|
||||
await session.execute(
|
||||
sa.select(
|
||||
Customer.friend_id,
|
||||
Customer.friend_nickname,
|
||||
Customer.friend_remark,
|
||||
Customer.updated_at,
|
||||
model.friend_id,
|
||||
model.friend_nickname,
|
||||
model.friend_remark,
|
||||
model.updated_at,
|
||||
)
|
||||
.where(*conds)
|
||||
.order_by(Customer.updated_at.desc(), Customer.friend_id.asc())
|
||||
.order_by(model.updated_at.desc(), model.friend_id.asc())
|
||||
)
|
||||
).all()
|
||||
return count, [
|
||||
@@ -311,18 +354,26 @@ async def _flag_customers(
|
||||
async def _del_customers(
|
||||
session: AsyncSession, wxid: str, start: int, end: int, detail: bool
|
||||
) -> tuple[int, list[dict[str, Any]]]:
|
||||
"""删除客户:delete_flag=1 现存量(start/end 按口径忽略)。"""
|
||||
"""删除客户:delete_flag=1 现存量(start/end 按口径忽略)。
|
||||
|
||||
读 link(per-relation delete_flag,共享好友不漏计,2026-09-09)。
|
||||
"""
|
||||
return await _flag_customers(
|
||||
session, wxid, Customer.delete_flag == 1, detail=detail
|
||||
session, wxid, CustomerAccountLink.delete_flag == 1,
|
||||
detail=detail, model=CustomerAccountLink,
|
||||
)
|
||||
|
||||
|
||||
async def _blocked_customers(
|
||||
session: AsyncSession, wxid: str, start: int, end: int, detail: bool
|
||||
) -> tuple[int, list[dict[str, Any]]]:
|
||||
"""流失(拉黑)客户:is_block=1 现存量(start/end 按口径忽略)。"""
|
||||
"""流失(拉黑)客户:is_block=1 现存量(start/end 按口径忽略)。
|
||||
|
||||
保持档案口径:is_block 未镜像进 link 表,切了会漏;已知偏差=共享好友
|
||||
在其他员工名下的拉黑不计入(待 link 补列后切换)。
|
||||
"""
|
||||
return await _flag_customers(
|
||||
session, wxid, Customer.is_block.is_(True), detail=detail
|
||||
session, wxid, Customer.is_block.is_(True), detail=detail, model=Customer
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -999,11 +999,247 @@ class SensitiveScanState(Base):
|
||||
updated_at: Mapped[dt.datetime] = _updated_at()
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 经营管理后台(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)
|
||||
# 回复消息 msg_id:同毫秒 F,T,F,T 交换的 (thread_end_ts, reply_ts) 并列
|
||||
# 去重(一条回复恰好闭合一个串);兼作下钻证据定位回复消息
|
||||
reply_msg_id: Mapped[str] = mapped_column(String(64), 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 已核实:1:1 通话可匹配率 100%,@chatroom 群通话除外)。
|
||||
"""
|
||||
|
||||
__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 种子维护。
|
||||
JSON 列(config_json 等)修改须整列重赋值,原地修改不会提交。"""
|
||||
|
||||
__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]] = 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__ = [
|
||||
"ChatGroup",
|
||||
"Customer",
|
||||
"Department",
|
||||
"EmployeeCallDaily",
|
||||
"EmployeeMetricDaily",
|
||||
"EmployeeScoreConfig",
|
||||
"EmployeeScoreSnapshot",
|
||||
"FirstReplyFact",
|
||||
"GroupMember",
|
||||
"ManagementFollowup",
|
||||
"Media",
|
||||
"Message",
|
||||
"MomentsPost",
|
||||
|
||||
@@ -0,0 +1,195 @@
|
||||
"""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)。
|
||||
|
||||
时序确定性:同毫秒并列(ts 相同的 F/T 消息)按 msg_id 定序(wechat
|
||||
msgSvrId 服务端单调,近似真实到达序)——SQL 等键返回顺序未定义,不做
|
||||
定序会导致串归属随查询抖动;fact PK 含 reply_msg_id,同毫秒 F,T,F,T
|
||||
交换的两个样本不撞主键。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from collections import namedtuple
|
||||
from collections.abc import Iterable
|
||||
|
||||
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 = 串末条客户消息所在上海自然日,
|
||||
# reply_msg_id = 闭合该串的员工回复消息 msg_id(PK 并列去重 + 下钻证据)
|
||||
FactSample = namedtuple("FactSample", "thread_end_ts reply_ts fr_ms day reply_msg_id")
|
||||
# 某会话某日的串聚合(分子分母同源)
|
||||
DayAgg = namedtuple("DayAgg", "threads answered fr_sum_ms")
|
||||
|
||||
|
||||
class _DayAcc:
|
||||
__slots__ = ("answered", "fr_sum_ms", "threads")
|
||||
|
||||
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, msg_id)] 该会话全历史,
|
||||
**须按 (ts, msg_id) 升序**(load_conversation 负责排序;同毫秒并列以
|
||||
msg_id 定序,见模块 docstring)。
|
||||
|
||||
- 串 = 客户连续消息(is_send=False)不被员工消息(is_send=True)打断的
|
||||
段;is_send=None 跳过(不打断也不起串)——与 _thread_stats 判定一致。
|
||||
- 每个被回串一个样本:fr_ms = 员工首条回复 ts − 串**末条**客户消息 ts,
|
||||
归属 day = 串末条所在上海自然日(跨夜串不再丢失);reply_msg_id =
|
||||
该回复消息 msg_id——一条回复恰好闭合一个串,同毫秒 F,T,F,T 交换的
|
||||
(thread_end_ts, reply_ts) 并列靠它去重,兼作下钻证据。
|
||||
- 会话末尾仍未闭合的串:计入当日 threads(回复率分母),不产生样本。
|
||||
"""
|
||||
facts: list[FactSample] = []
|
||||
acc: dict[int, _DayAcc] = {}
|
||||
in_thread = False
|
||||
end_ts: int | None = None
|
||||
for ts, is_send, msg_id 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, msg_id))
|
||||
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, msg_id),Python 侧按
|
||||
(ts, msg_id) 排序——串算法需严格时序且同毫秒并列须定序(SQL 等键顺序
|
||||
未定义,不排序会随查询抖动;与 summary.recompute_days 同惯例,避免
|
||||
依赖 SQL 排序)。friend 过滤口径与 summary._peer_filter 的 friend 分支
|
||||
同源:talker_id 原值精确匹配 + room_id 为空(peer_id 与 talker_id 同源
|
||||
同形态,不做截断/归一化)。
|
||||
|
||||
公有命名:employee_metrics 宽表生成复用同一加载口径(同源原则)。"""
|
||||
rows = (
|
||||
await session.execute(
|
||||
sa.select(
|
||||
Message.timestamp, Message.is_send, Message.msg_id
|
||||
).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], r[2]))
|
||||
|
||||
|
||||
async def recompute_pairs(
|
||||
session: AsyncSession, pairs: list[tuple[str, str, str]]
|
||||
) -> int:
|
||||
"""重算给定 friend 会话的 fact 行:按会话 delete+insert,幂等。
|
||||
|
||||
pairs = [(channel, owner_account_id, peer_id)]。返回会话数。每 500 个
|
||||
会话 flush 一次约束 session 内存(与 summary.recompute_all 同惯例——
|
||||
全量回填百万级 fact 行不能攒到最后),同时打进度日志(回填分钟级,
|
||||
可观测)。commit 由调用方负责。
|
||||
"""
|
||||
n = 0
|
||||
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,
|
||||
reply_msg_id=f.reply_msg_id,
|
||||
day=f.day,
|
||||
fr_ms=f.fr_ms,
|
||||
)
|
||||
)
|
||||
n += 1
|
||||
if n % 500 == 0:
|
||||
await session.flush()
|
||||
logger.info("first_reply_facts: 已重算 %d 个会话", n)
|
||||
return n
|
||||
|
||||
|
||||
async def all_friend_pairs(session: AsyncSession) -> list[tuple[str, str, str]]:
|
||||
"""全部 1:1 会话(friend 行 distinct)——全量回填用。
|
||||
|
||||
依赖 msg_daily_stats 已回填(先 stats 后 facts):friend 会话清单取自
|
||||
stats 表而非 message 全表 distinct——写路径钩子保证 stats 常新,代价
|
||||
远低于扫 message。
|
||||
"""
|
||||
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: Iterable[int]
|
||||
) -> 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))
|
||||
@@ -0,0 +1,164 @@
|
||||
"""经营管理后台批次A:6 张新表(fact/宽表/通话小表/评分配置/快照/辅导)
|
||||
|
||||
Revision ID: b7c8d9e0f1a2
|
||||
Revises: f7a8b9c0d1e2
|
||||
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 = "f7a8b9c0d1e2"
|
||||
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),
|
||||
# 同毫秒 F,T,F,T 交换的 (thread_end_ts, reply_ts) 并列去重(一条回复
|
||||
# 恰好闭合一个串,reply_msg_id 唯一定位);兼作下钻证据定位回复消息
|
||||
sa.Column("reply_msg_id", sa.String(length=64), 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)
|
||||
@@ -112,3 +112,22 @@ async def make_client() -> AsyncIterator[
|
||||
await client.aclose()
|
||||
await engine.dispose()
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
@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()
|
||||
|
||||
@@ -5,7 +5,7 @@ import time
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.db.models import ChatGroup, Customer, Message
|
||||
from app.db.models import ChatGroup, Customer, CustomerAccountLink, Message
|
||||
|
||||
OWNER = "wx_owner_s"
|
||||
CH = "wechat"
|
||||
@@ -45,6 +45,40 @@ async def test_sessions_friend_and_group(make_client):
|
||||
assert g["name"] == "群一号" and g["today_count"] == 1
|
||||
|
||||
|
||||
async def test_sessions_shared_friend_name_from_link(make_client):
|
||||
"""共享好友昵称读 link 快照(2026-09-09):档案行挂在别的员工名下时,
|
||||
会话列表此前回退显示 wxid;link 表每账号一行,回填后都有昵称。
|
||||
"""
|
||||
|
||||
async def seed(session: AsyncSession) -> None:
|
||||
session.add_all([
|
||||
# OWNER 与共享好友 t_shared 有会话
|
||||
Message(msg_id="sh1", channel=CH, owner_account_id=OWNER, talker_id="t_shared",
|
||||
room_id=None, is_send=True, timestamp=_NOW, type="text", type_raw=1, text="x"),
|
||||
# 档案行(last-wins)挂在别的员工名下——按 owner 查档案必然查空
|
||||
Customer(friend_id="t_shared", channel=CH,
|
||||
owner_account_id="wx_other", friend_nickname="A**郑"),
|
||||
# OWNER 名下的 per-owner link 快照
|
||||
CustomerAccountLink(channel=CH, owner_account_id=OWNER, friend_id="t_shared",
|
||||
friend_nickname="A**郑", friend_remark=None),
|
||||
])
|
||||
await session.commit()
|
||||
|
||||
client = await make_client(seed)
|
||||
r = await client.get("/api/sessions", params={"owner_account_id": OWNER, "channel": CH})
|
||||
assert r.status_code == 200
|
||||
by = {s.get("talker_id"): s for s in r.json()["sessions"]}
|
||||
assert by["t_shared"]["name"] == "A**郑"
|
||||
|
||||
|
||||
async def test_sessions_name_falls_back_to_archive_without_link(make_client):
|
||||
"""link 未覆盖的好友回退档案路径(空表过渡安全)。"""
|
||||
client = await make_client(_seed)
|
||||
r = await client.get("/api/sessions", params={"owner_account_id": OWNER, "channel": CH})
|
||||
by = {s.get("talker_id"): s for s in r.json()["sessions"]}
|
||||
assert by["t1"]["name"] == "Alice" # 档案行在 OWNER 名下
|
||||
|
||||
|
||||
async def test_sessions_sorted_by_last_chat_time_desc(make_client):
|
||||
client = await make_client(_seed)
|
||||
body = (await client.get("/api/sessions", params={"owner_account_id": OWNER})).json()
|
||||
|
||||
@@ -5,7 +5,14 @@ import time
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.db.models import ChatGroup, Customer, Message, PhoneCall, WechatAccount
|
||||
from app.db.models import (
|
||||
ChatGroup,
|
||||
Customer,
|
||||
CustomerAccountLink,
|
||||
Message,
|
||||
PhoneCall,
|
||||
WechatAccount,
|
||||
)
|
||||
|
||||
OWNER = "wxid_o"
|
||||
UID = "U32HEX" # 员工 userId(32位HEX)——phone_call.user_id 存的是它,不是 wxId!
|
||||
@@ -39,6 +46,11 @@ async def _seed_counts(s: AsyncSession) -> None:
|
||||
# 删除/拉黑(现存量口径,忽略窗口)
|
||||
s.add(Customer(friend_id="c9", channel="wechat", owner_account_id=OWNER, delete_flag=1))
|
||||
s.add(Customer(friend_id="c8", channel="wechat", owner_account_id=OWNER, is_block=True))
|
||||
# link 行(生产抓取经 upsert_customer 成对写;新客户/删除统计读 link)
|
||||
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c1", add_time=IN))
|
||||
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c2"))
|
||||
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c3", add_time=1_000))
|
||||
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c9", delete_flag=1))
|
||||
# 语音通话: user_id 是 32HEX 非 wxId!另 seed 一条直存 wxId 的(不计,锁死 join 口径)
|
||||
s.add(PhoneCall(call_id="pc1", channel="voice_call", user_id=UID,
|
||||
start_call_time=IN + 4, call_number="c1"))
|
||||
@@ -78,6 +90,9 @@ async def test_counts_default_window_last_7_days(make_client):
|
||||
s.add(Customer(friend_id="c_now", channel="wechat", owner_account_id=OWNER, add_time=now))
|
||||
s.add(Customer(friend_id="c_old", channel="wechat", owner_account_id=OWNER,
|
||||
add_time=now - 8 * DAY))
|
||||
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c_now", add_time=now))
|
||||
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c_old",
|
||||
add_time=now - 8 * DAY))
|
||||
s.add(PhoneCall(call_id="pc_now", channel="voice_call", user_id=UID,
|
||||
start_call_time=now, call_number="c_now"))
|
||||
s.add(PhoneCall(call_id="pc_old", channel="voice_call", user_id=UID,
|
||||
@@ -127,6 +142,12 @@ async def test_detail_new_customers_fields(make_client):
|
||||
s.add(Customer(friend_id="c2", channel="wechat", owner_account_id=OWNER, add_time=None,
|
||||
friend_nickname="兜底客"))
|
||||
s.add(Customer(friend_id="c3", channel="wechat", owner_account_id=OWNER, add_time=1_000))
|
||||
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c1", add_time=IN,
|
||||
friend_nickname="牛牛", friend_remark="大客户",
|
||||
head_image_url="http://a/1.png"))
|
||||
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c2",
|
||||
friend_nickname="兜底客"))
|
||||
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c3", add_time=1_000))
|
||||
s.add(Message(msg_id="m1", channel="wechat", owner_account_id=OWNER, talker_id="c2",
|
||||
timestamp=IN - 5))
|
||||
await s.commit()
|
||||
@@ -238,6 +259,9 @@ async def test_detail_del_and_blocked_fields(make_client):
|
||||
s.add(Customer(friend_id="c_ok", channel="wechat", owner_account_id=OWNER))
|
||||
s.add(Customer(friend_id="c_other", channel="wechat", owner_account_id="wxid_other",
|
||||
delete_flag=1))
|
||||
# 删除读 link:c_del 补 link 行;is_block 保持档案(c_blk 不补)
|
||||
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c_del",
|
||||
delete_flag=1, friend_nickname="已删客"))
|
||||
s.add(Customer(friend_id="c_fine", channel="wechat", owner_account_id=OWNER,
|
||||
delete_flag=0, is_block=False))
|
||||
await s.commit()
|
||||
@@ -394,3 +418,32 @@ async def test_new_customers_fallback_first_msg_before_window(make_client):
|
||||
assert r_count.status_code == r_detail.status_code == 200
|
||||
assert r_count.json()["new_customers"] == 0
|
||||
assert r_detail.json()["rows"] == []
|
||||
|
||||
|
||||
async def test_new_customers_counts_shared_friend_via_link(make_client):
|
||||
"""共享好友(档案挂别的员工名下)凭 link 行计入新客户(2026-09-09 口径修复)。
|
||||
|
||||
旧档案口径下 c_shared 对 OWNER 不可见(add_time 在窗内也不计);切 link
|
||||
后按 per-owner 关系计数,与云客按员工统计一致。
|
||||
"""
|
||||
|
||||
async def seed(s: AsyncSession) -> None:
|
||||
s.add(WechatAccount(wx_id=OWNER, user_id=UID))
|
||||
# 档案行(last-wins)挂在别的员工名下
|
||||
s.add(Customer(friend_id="c_shared", channel="wechat",
|
||||
owner_account_id="wxid_other", add_time=IN))
|
||||
# OWNER 名下的 link 关系
|
||||
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER,
|
||||
friend_id="c_shared", add_time=IN))
|
||||
await s.commit()
|
||||
|
||||
client = await make_client(seed)
|
||||
r = await client.get(
|
||||
"/api/staff-stats",
|
||||
params={"wechat_id": OWNER, "kind": "new_customers",
|
||||
"start": WIN_START, "end": WIN_END},
|
||||
)
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert [row["talker_id"] for row in body["rows"]] == ["c_shared"]
|
||||
assert len(body["rows"]) == 1
|
||||
|
||||
@@ -0,0 +1,202 @@
|
||||
"""first_reply_fact 跨日精确串算法判别测试(批次A Task3)。
|
||||
|
||||
四个核心判据(v2 §1.1 #1/#5、§9):
|
||||
1. 跨夜串必须产生样本(日切口径丢失的场景);
|
||||
2. 多日未回串只计一次、归属末条日(日切会按天重复计——分母膨胀);
|
||||
3. 同源:threads_answered ≡ fact 行数(按归属日);
|
||||
4. is_send=None 不参与串判定(与 _thread_stats 一致)。
|
||||
|
||||
评审修正补判据:
|
||||
5. 同毫秒 F,T,F,T 交换 → reply_msg_id 进 PK 去重,不撞主键(原崩溃回归);
|
||||
6. 同毫秒并列按 (ts, msg_id) 定序,结果确定不随 SQL 等键序抖动。
|
||||
"""
|
||||
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, "f1"),
|
||||
(_ts(2026, 9, 2, 9, 0), True, "t1")]
|
||||
)
|
||||
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, "f1"), (_ts(2026, 9, 1, 20, 0), False, "f2"),
|
||||
(_ts(2026, 9, 2, 9, 0), False, "f3")]
|
||||
)
|
||||
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, "f1"), (_ts(2026, 9, 1, 10, 5), False, "f2"),
|
||||
(_ts(2026, 9, 1, 10, 30), True, "t1"), # 回复1:fr=30-10:05=25min
|
||||
(_ts(2026, 9, 1, 11, 0), False, "f3"),
|
||||
(_ts(2026, 9, 1, 12, 0), True, "t2")]) # 回复2:fr=12:00-11:00=60min
|
||||
assert len(facts) == 2
|
||||
assert facts[0].fr_ms == 25 * 60 * 1000
|
||||
assert agg[20260901].threads == 2 and agg[20260901].answered == 2
|
||||
assert agg[20260901].fr_sum_ms == 85 * 60 * 1000 # 25min + 60min
|
||||
|
||||
|
||||
def test_unknown_direction_ignored():
|
||||
facts, _agg = compute_conversation(
|
||||
[(_ts(2026, 9, 1, 10, 0), False, "f1"), (_ts(2026, 9, 1, 10, 10), None, "u1"),
|
||||
(_ts(2026, 9, 1, 10, 20), True, "t1")]
|
||||
)
|
||||
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, "t1"), (_ts(2026, 9, 1, 10, 0), True, "t2")]
|
||||
)
|
||||
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__
|
||||
|
||||
|
||||
def test_tie_break_by_msg_id_deterministic():
|
||||
"""同毫秒 [F(m1), T(m2)]:按 (ts, msg_id) 定序 F 在前 → 1 个被回串;
|
||||
同一 multiset 无论入参顺序,排序后结果一致(SQL 等键序未定义,不能
|
||||
随查询抖动)。"""
|
||||
same = _ts(2026, 9, 1, 10, 0)
|
||||
rows = [(same, False, "m1"), (same, True, "m2")]
|
||||
ordered = sorted(rows, key=lambda r: (r[0], r[2]))
|
||||
assert ordered[0][2] == "m1" # F(m1) 先于 T(m2)
|
||||
facts, agg = compute_conversation(ordered)
|
||||
assert len(facts) == 1 and facts[0].reply_msg_id == "m2"
|
||||
assert agg[20260901].answered == 1
|
||||
shuffled = [rows[1], rows[0]] # 同一 multiset,入参顺序相反
|
||||
facts2, agg2 = compute_conversation(
|
||||
sorted(shuffled, key=lambda r: (r[0], r[2])))
|
||||
assert facts2 == facts and agg2 == agg
|
||||
|
||||
|
||||
# ---------------------------- 落库 ----------------------------
|
||||
|
||||
async def _seed_conversation(session, *, channel="wechat", owner="wx1",
|
||||
peer="p1", rows):
|
||||
for ts, is_send, msg_id in rows:
|
||||
session.add(Message(msg_id=msg_id, 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, "p1-f0"),
|
||||
(_ts(2026, 9, 2, 9, 0), True, "p1-t1")])
|
||||
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
|
||||
assert rows[0].reply_msg_id == "p1-t1"
|
||||
# 幂等:重跑不重复
|
||||
await recompute_pairs(s, [("wechat", "wx1", "p1")])
|
||||
await s.commit()
|
||||
assert len((await s.execute(sa.select(FirstReplyFact))).scalars().all()) == 1
|
||||
|
||||
|
||||
async def test_same_ms_ftft_two_distinct_facts(make_session):
|
||||
"""同毫秒 F,T,F,T 交换:两个样本 (thread_end_ts, reply_ts) 完全并列
|
||||
——reply_msg_id 进 PK 去重后恰好 2 行(评审复现的主键冲突回归)。"""
|
||||
s = make_session
|
||||
same = _ts(2026, 9, 1, 10, 0)
|
||||
await _seed_conversation(s, rows=[
|
||||
(same, False, "m1"), (same, True, "m2"),
|
||||
(same, False, "m3"), (same, True, "m4"),
|
||||
])
|
||||
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) == 2
|
||||
assert {r.reply_msg_id for r in rows} == {"m2", "m4"}
|
||||
|
||||
|
||||
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, "p1-f0"),
|
||||
(_ts(2026, 9, 1, 11, 0), True, "p1-t1")])
|
||||
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_recompute_pairs_empty_conversation(make_session):
|
||||
"""stats 行在但 message 无合格行(空会话/全被过滤):删旧行不插新——
|
||||
delete+insert 语义对空会话同样成立(幂等清空)。"""
|
||||
s = make_session
|
||||
s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1",
|
||||
peer_kind="friend", peer_id="p1", day=20260901))
|
||||
# 预置历史残留 fact,重算后应被清掉
|
||||
s.add(FirstReplyFact(channel="wechat", owner_account_id="wx1", peer_id="p1",
|
||||
thread_end_ts=_ts(2026, 9, 1, 10, 0),
|
||||
reply_ts=_ts(2026, 9, 1, 11, 0), reply_msg_id="stale",
|
||||
day=20260901, fr_ms=3600 * 1000))
|
||||
await s.commit()
|
||||
n = await recompute_pairs(s, [("wechat", "wx1", "p1")])
|
||||
await s.commit()
|
||||
assert n == 1
|
||||
assert (await s.execute(sa.select(FirstReplyFact))).scalars().all() == []
|
||||
|
||||
|
||||
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")]
|
||||
@@ -0,0 +1,72 @@
|
||||
"""客户生命周期判定纯函数测试(批次A Task4)。
|
||||
|
||||
口径(v2 §1.1 #3 + v1 §21 默认,已用户确认):新客 7 天、沉睡 30 天、
|
||||
历史≥5 条才算沉睡(对齐 churn real min_msgs=5)。analytics 现有页
|
||||
(churn/客户资产)暂不切换(口径差异 v2 §5 标注),不动 636 既有测试。
|
||||
"""
|
||||
|
||||
from app.analytics.lifecycle import DORMANT_DAYS, MIN_HISTORY_MSGS, NEW_DAYS, 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"
|
||||
)
|
||||
# 窗口边界:as_of-6=20260830 含当天共 7 天 → new;再早一天 → 非 new
|
||||
assert classify(20260830, 20260830, None, 20260905, 100) == "new"
|
||||
assert classify(20260829, 20260829, None, 20260905, 100) == "cooling"
|
||||
|
||||
|
||||
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"
|
||||
)
|
||||
# 间隔仅 9 天 → 只是 active
|
||||
assert (
|
||||
classify(
|
||||
first_day=20260101, last_day=20260905, prev_day=20260827, as_of=20260905, hist_msgs=50
|
||||
)
|
||||
== "active"
|
||||
)
|
||||
# 间隔恰好 30 天 → 严格 >30 不算回流,落回 active
|
||||
assert (
|
||||
classify(
|
||||
first_day=20260101, last_day=20260905, prev_day=20260806, as_of=20260905, hist_msgs=50
|
||||
)
|
||||
== "active"
|
||||
)
|
||||
# 回流必须当日互动:前次间隔 31 天(>30)但 last_day=as_of-1 → 不算回流,按 gap=1 落 active
|
||||
assert classify(20260101, 20260904, 20260804, 20260905, 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
|
||||
@@ -0,0 +1,139 @@
|
||||
"""管理域 6 张新表的模型冒烟测试(批次A Task1)。
|
||||
|
||||
只验证:建表可用、主键/唯一约束生效、默认值正确。业务逻辑在
|
||||
first_reply_facts / employee_metrics 的测试里。
|
||||
"""
|
||||
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, reply_msg_id="m1",
|
||||
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 = {"channel": "wechat", "owner_account_id": "wx1", "peer_id": "p1",
|
||||
"thread_end_ts": 1, "reply_ts": 2, "reply_msg_id": "m1",
|
||||
"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_metric_daily_pk_duplicate_rejected(make_session):
|
||||
"""同 (day, channel, user_id, metric_version) 复合主键防重。
|
||||
|
||||
Task 6 delete+insert 幂等重算的安全网:版本内重复落表直接 IntegrityError。
|
||||
"""
|
||||
s = make_session
|
||||
s.add(EmployeeMetricDaily(day=20260901, channel="wechat", user_id="u1"))
|
||||
await s.commit()
|
||||
s.add(EmployeeMetricDaily(day=20260901, channel="wechat", user_id="u1"))
|
||||
with pytest.raises(IntegrityError):
|
||||
await s.commit()
|
||||
|
||||
|
||||
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_call_daily_pk_duplicate_rejected(make_session):
|
||||
s = make_session
|
||||
s.add(EmployeeCallDaily(day=20260901, user_id="u1"))
|
||||
await s.commit()
|
||||
s.add(EmployeeCallDaily(day=20260901, user_id="u1"))
|
||||
with pytest.raises(IntegrityError):
|
||||
await s.commit()
|
||||
|
||||
|
||||
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_score_config_config_json_not_null(make_session):
|
||||
"""config_json NOT NULL——注解与 DDL 一致(修复:去掉 | None)。
|
||||
|
||||
注意 JSON 列语义:Python None 经 JSON 序列化变成字符串 'null' 而非
|
||||
SQL NULL,不会违反约束;测 NOT NULL 须用 sa.null() 显式发 SQL NULL。
|
||||
"""
|
||||
s = make_session
|
||||
s.add(EmployeeScoreConfig(version="v1.0", name="空配置",
|
||||
config_json=sa.null()))
|
||||
with pytest.raises(IntegrityError):
|
||||
await s.commit()
|
||||
|
||||
|
||||
async def test_snapshot_batch_user_unique(make_session):
|
||||
s = make_session
|
||||
common = {"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
|
||||
Reference in New Issue
Block a user