Compare commits

...

5 Commits

Author SHA1 Message Date
浅念 d42895337d feat(link): customer_account_link 账号↔客户多对多——共享好友员工维度对齐云客
用户报「Lily 好友数 36,点进员工客户页只有 12」。真机探明根因:那 24 个
是共享客户(同一人加多个员工),云客 per-employee 各算一个;而 customer
表 PK=friend_id 全局单行、归属 last-wins,其他员工名下必然「消失」。
抓取器本身每员工都拿全了(日志 declared=36 rows=36),丢在入库模型层。

- 新表 customer_account_link(渠道×账号×客户一行,per-relation 快照:
  昵称/备注/头像/手机号/收发计数/最近沟通/add_time/fromType/isDel)
- 写:repos.upsert_customer 随档案同事务维护(所有好友抓取路径一处挂全);
  新增 prune_customer_links 对账
- 读:/api/customers?owner_account_id= 改读 link(与云客 contacts_num 口径
  一致);该账号 link 为空时回退档案路径(过渡安全);与 dept/user 等
  过滤组合时维持旧行为;CustomerOut 透传 from_type
- 抓取器:friends/wa-friends 逐员工整批 upsert 后按名单对账(取满声明
  total 才删,防半份名单误删)
- 前端:StaffFriends 数据来源列优先 link 的 from_type

611 测试绿(link 新 6 例);本地验证空表回退/造数接管/from_type 透传。
上线后需重跑 friends+wa-friends 回填 link(存量关系无法从档案反推)。

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-08 17:17:06 +08:00
浅念 4ad239597e docs(management): v2计划复核收尾——评审确认后修4处文档级问题
评审代理逐案例推演确认P1#5同源修正算法成立(Approval维持),遗留修正:
- §1.1#2修正列更新(串类三列由fact同源输出,不再说全量自算)
- §1.1#5去掉'>100%'夸大表述(单向低估,不可达)
- §9对账点限定消息计数列(串类列禁止相等断言)
- Task0口径只记录文档,配置落库归Task3种子CLI
- P2/P3行号全局重排(#6-#13),交叉引用同步
- §3.2负载列补数据源标注;时效说明措辞精确化

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-08 16:46:40 +08:00
浅念 8257086921 docs(management): v2计划第五轮复查——回复率分子分母同源修正(P1#5)+配置种子等补充
- P1#5:宽表串类三列改由fact计算模块同源输出(串末条归属日),
  不再取msg_daily_stats日切聚合——混口径会使多日未回串膨胀分母,
  回复率低估甚至>100%;msg计数/负载列与串算法无关仍取msg_daily_stats
- P2/P3:Task3补默认配置v1.0种子落库;时效说明;metric_version默认v1;
  call_followup多账号语义进Task0核实;继承索引补v1 §12.1/12.2;
  测试补分子分母同源判据(reply_rate<=1/threads_answered≡fact行数)

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-08 16:40:35 +08:00
浅念 7a7eba766d docs(management): v2计划评审修正——引用定位/删减清单补全/口径表补活跃员工行
评审结论 Approved + 4 条 advisory 全部采纳:
- 降温频次定义/异常裁剪原则的引用从 v1 修正为上游整合计划
- 删减清单补 effective_sessions/activated_customer_count/cust_threads 更名说明
- 口径决议表补活跃员工(幽灵过滤)差异行

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-08 16:35:55 +08:00
浅念 c7b7df5011 docs(management): 经营管理后台v2实施计划——查漏补缺修订版
对 v1 逐项对照现有代码重新设计技术底座:
- P1修正:fact表跨日精确算法独立计算(跨夜串丢失/写钩子零改动);
  宽表1:1口径全量自算(stat_daily_staff混合口径不可拷贝);
  生命周期统一距今天数口径;对账判据预写死
- P2修正:分位数SQL窗口函数;full_sync 12→14步
- P3:核实清单/幽灵员工过滤/功能开关降级/五维图3+2占位

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-08 16:30:36 +08:00
11 changed files with 798 additions and 10 deletions
@@ -0,0 +1,309 @@
# 云客经营管理后台实施计划 v2(贴合代码修订版)
- **日期**:2026-09-08
- **版本**:v2.0
- **状态**:待评审后进入开发
- **与 v1 的关系**:替代 `2026-09-08-yunke-management-console-implementation.md`(plans/ 与 specs/ 两份副本内容相同)的**技术设计部分**;产品范围、页面内容细则、验收标准沿用 v1,本文不重复,冲突处以本文为准
- **上游方案**:[云客数据智能管理平台完整实施计划](./2026-09-08-yunke-data-intelligence-management-plan.md)
- **修订方式**:对 v1 逐项做「现有代码 ↔ 文档要求」查漏补缺,钉在真实代码事实上重新设计底座
---
## 0. 已确认的决策记录
| 决策项 | 结论 |
|---|---|
| 一期范围 | 按 v1 原计划全量:管理驾驶舱 + 员工总览 + 员工详情 + 考核快照(Task 0-13) |
| 业务口径 | 全按 v1 §21 默认值:自然时间 SLA、2h+24h 双 SLA、30 天沉睡、综合分标「试算」、CSV 导出、仅管理层可见、个人基线=过去 4 周 |
| 底座方案 | 方案 B:新增首响事实表 + 收敛版员工宽表(v1 架构骨架不变) |
| 关键修正 | 见 §1 修订总览——fact 表独立计算(跨日精确算法)、宽表 1:1 口径全量自算、生命周期沿用距今天数口径 |
---
## 1. 相对 v1 的修订总览
### 1.1 P1 级(正确性/口径,必须修正)
| # | 问题 | 证据 | 修正 |
|---|---|---|---|
| 1 | **跨夜串丢失**:现有日切算法下「客户 23:00 发→员工次日 9:00 回」的首响样本整条丢失(Day1 串未闭合只计 threads;Day2 重起算时无客户消息不成串,员工回复落空段)。P90/24h 回复率系统性低估,而评分重度依赖这两项 | `app/storage/summary.py` `_thread_stats` 日切近似,docstring 自述「跨夜串归属日偏差」实为样本丢失 | fact 表**不挂写钩子**,由 4h 滚动+夜间全量任务用**跨日精确串算法**(按会话全历史计算)独立生成;写钩子零改动,`_thread_stats` 不改 |
| 2 | **stat_daily_staff 是 friend+room 混合口径**:`sent/recv/threads/answered/fr` 五列聚合无 kind 过滤,群消息污染首响/回复率。v1 说「从 stat_daily_staff 拷贝」不可行 | `app/storage/staff_summary.py` `recompute_days` 聚合循环 | 宽表消息计数列从 msg_daily_stats 按 `peer_kind='friend'` 自算(account_map 归并模式复用);串类三列由 fact 模块同源输出(见 #5);管理页与现有看板排行的数字差异属口径不同,各自标注 |
| 3 | **「降温」定义冲突**:上游整合计划(`2026-09-08-yunke-data-intelligence-management-plan.md` §8.1/§15)降温=频次低于个人基线;现有 churn/客户资产分层=距今天数区间。两套「降温」会并存打架 | `app/analytics/metrics.py` churn 三分法 | 一期生命周期 4 列统一用**现有距今天数口径**;频次基线降温后置二期;差异写入 metric-definitions |
| 4 | **对账口径必须预写死**:fact(跨日精确)与 msg_daily_stats 三列(日切近似)即使 friend 口径也不相等,差集=跨夜闭合串 | 同 #1 | Task 2 验收断言:fact 样本集 ⊇ 日切被回串,差集=跨夜串;禁止相等断言 |
| 5 | **回复率分子分母必须同源**:v2 初稿宽表串类列(cust_threads 等)取自 msg_daily_stats 日切聚合,而回复率分子取自 fact(跨日)——多日未回串使日切分母膨胀,回复率被系统性低估 | 本轮复查发现(#2 修正的连带遗漏) | 宽表串类三列改由 fact 计算模块**同源输出**(按串末条归属日的 threads/answered/fr_sum);msg_sent/recv 与 active_friends/rooms 与串算法无关,仍取 msg_daily_stats |
### 1.2 P2 级(性能/实现修正)
| # | 问题 | 修正 |
|---|---|---|
| 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 之后) |
### 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 个「待数据」占位(行动执行/风险合规),不暗示完整五维 |
---
## 2. 查漏补缺:复用映射总表
| v1 能力项 | 现有可复用 | 本次动作 |
|---|---|---|
| 员工日指标底座 | `stat_daily_staff`(8 列,混合口径)、`msg_daily_stats`(首响三列,日切近似) | 两表**均不动**;新增 `first_reply_fact` + `employee_metric_daily`(§3) |
| 时间窗/环比口径 | `app/analytics/metrics.py` `scope_day_keys`/`_prev_scope_days`(5 种窗+同期对齐) | management 层直接 import,不复制时区逻辑 |
| 生命周期分层 | `metrics.py` 客户资产分层 / churn 三分法(距今天数口径) | 提炼共享函数,日聚合任务与 analytics 共用;analytics 调用点保持不变(包装层) |
| 首响均值/回复率算法 | `staff_ranking` 的 fr_avg/reply_rate | 周期值改为宽表 SUM + fact 表精确分位 |
| 异常检测雏形 | `anomaly`(昨日 vs 前 7 日均值偏离>50%) | `management/anomalies.py` 扩展为规则集:回复恶化(fact P90)、待回复增长(B2)、活动骤降/过载(宽表) |
| 驾驶舱页面 | `DashboardView.vue` 的图表模式、`useChart`、筛选栏、骨架屏/空态分语 | 新建 ManagementDashboardView,复用全部 composables/utils;旧页并存过渡后重定向 |
| 流失/该回未回/敏感词 | 三个现有页面 | 不动,作为下钻目标 + 驾驶舱行动入口聚合 |
| 数据截至/新鲜度 | `app/analytics/health.py` 的 data_as_of 逻辑 | 管理页「数据截至/完整性」角标直接复用 |
| 调度 | `analytics_tick`(4h:staff 汇总+敏感扫描) | 追加第三步:fact 滚动重算 + 宽表滚动重算 |
| 回填/自愈 | `full_sync_all.sh` 第 11/12 步 | 追加第 13/14 步(fact/宽表全量重建)+ 两条回填 CLI |
| 员工事实下钻 | `/staff-chat`、`/chat`、`/groups`、`/calls` | 不动,管理页只做链接跳转 |
| CSV 导出 | `csvExport.ts`(全量翻页+中文 BOM) | 直接复用 |
---
## 3. 数据库设计:5 张新表
### 3.1 `first_reply_fact` 首响事实表(本设计新增)
**用途**:任意时间窗精确 P50/P90、2h/24h 回复率分子、指标解释抽屉的构成明细下钻证据。
**算法(跨日精确串口径,区别于 msg_daily_stats 的日切近似)**:
- 串 = 同一会话内客户连续消息(is_send=False)不被员工消息(is_send=True)打断的段;is_send=None 跳过;
- 每个被回串一个样本:`fr_ms = 员工首条回复 ts − 串末条客户消息 ts`;
- 样本归属 = 串末条消息所在**上海自然日**(跨夜串不再丢失);
- 只写 `peer_kind='friend'`(1:1)行——群串语义不成立且占串数大头,不写;
- 模块同时输出**每日串聚合**(threads/answered/fr_sum,同按串末条归属日)供宽表消费——回复率分子分母同源(§1.1 #5);
- 按会话全历史计算,幂等模式=按会话 delete+insert。
**表结构**:
```text
唯一键: (channel, owner_account_id, peer_id, thread_end_ts, reply_ts)
列: day int 串末条所在上海日 yyyymmdd(冗余物化,索引用)
thread_end_ts BigInteger 串末条客户消息 ts(ms)
reply_ts BigInteger 员工首条回复 ts(ms)
fr_ms BigInteger
索引: (day, channel) 趋势/窗口查询;唯一键自身覆盖会话下钻
```
**生成链路(不挂写钩子)**:
- 4h tick:对近 3 天有消息的会话(从 msg_daily_stats 近 3 天行取 pairs)做全历史重算;
- 夜间全量:full_sync_all.sh 第 13 步,遍历全部 friend 会话(复用 `recompute_all` 的 pairs 遍历模式);
- 回填 CLI:`python -m app.sync.cli first-reply-facts`,分批可重跑可记水位;
- 量级:日均几百行、全年几十万行,远小于 message 表。
### 3.2 `employee_metric_daily` 员工指标宽表(v1 表的收敛版)
**口径总则**:消息串类列 = 1:1(friend-only)口径,从 msg_daily_stats 自算;与现有看板(friend+room 混合)的数字差异属口径不同。
```text
PK: (day, channel, user_id, metric_version)
消息计数列(msg_daily_stats 自算,friend 过滤 + account_map 归并;与串算法无关):
msg_sent, msg_recv
串类列(fact 计算模块同源输出,跨日精确、按串末条归属日——与回复率分子同口径):
cust_threads, threads_answered, fr_sum_ms
负载列(msg_daily_stats,friend/room 分行天然区分):
active_friends(1:1 distinct 客户), active_rooms(distinct 群)
首响日值(趋势图用,从 fact 算):
first_reply_p50_ms, first_reply_p90_ms
生命周期 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
```
**相对 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);
- `effective_sessions` 不单列:v1 §6.3 工作负载视图的「有效会话」≡ cust_threads(1:1 有客户消息串的天级会话数),页面字段名从 cust_threads 映射;
- `activated_customer_count` 一期不做:「有效激活」定义是 v1 §21 #4 未决口径(几轮互动/几天窗口),随二期频次基线生命周期一起定义;
- `customer_threads` 更名为 `cust_threads`(与 msg_daily_stats/stat_daily_staff 现有列名对齐)。
**metric_version 语义**:只影响派生列(生命周期/通话跟进的算法口径);消息计数与串类事实列与版本无关。默认值 `v1`,算法演进时新版本并行落表不覆盖旧数据。
**时效说明**:消息计数列的源数据(msg_daily_stats)随写钩子近实时更新,串类列的源数据(fact)随 4h/夜间重算更新——两类列的上游时效不同,但宽表行统一随 4h/夜间重算落库;computed_at 取宽表生成时间,data_complete 标注 fact 是否已覆盖当日。
**评分用周期值的计算来源**(不在宽表存储):周期 P50/P90、2h/24h 回复率一律从 fact 表精确计算(SQL 窗口函数),宽表日值仅供趋势展示。
### 3.3 `employee_score_config` / `employee_score_snapshot` / `management_followup`
按 v1 §11.1 原样实施,无修改:
- config:version 唯一、status、effective_from/to、config_json(维度/指标/权重/方向/阈值/样本门槛),激活前权重总和与阈值校验;
- snapshot:snapshot_batch_id、period_type/period、score_config_version、metric_version、total_score、grade、score_status(scored/insufficient_sample/incomplete_data)、dimension_scores_json、metric_details_json、data_as_of、frozen_at/frozen_by、supersedes_id;同批次员工唯一;冻结后业务层禁改;
- followup:source_type/code/period/value、reason、action_type、owner_manager_id、status(open/in_progress/resolved/cancelled)、due_at、result、employee_feedback;索引 `(employee_user_id, status, due_at)`、`(owner_manager_id, status)`。
### 3.4 迁移要求
- 一个 Alembic migration 建 5 张表;migration 只建表和索引,**不扫描消息**;
- 数据回填走独立 CLI 分批执行(可暂停/重跑/记水位),先在本地 Docker 快照验证耗时与一致性;
- downgrade 只删本次新增对象;SQLite(测试)/MySQL(生产)双方言;
- 升级前确认当前 head(服务器注意:backend 容器**不自动跑 alembic**,须手动 `exec backend alembic upgrade head`)。
---
## 4. 计算链路与调度
```text
message 入库(写钩子,全路径覆盖——现状,零改动)
→ msg_daily_stats [summary.py]
→ stat_daily_staff [staff_summary.py,现状]
message(按会话全历史,跨日精确算法) [新 storage/first_reply_facts.py]
→ first_reply_fact
│
└─→ employee_metric_daily [新 storage/employee_metrics.py:
msg 计数(msg_daily_stats,friend) + 串类/分位(fact 同源)
+ customer 生命周期 + phone_call]
→ 期聚合/四对比(上期/个人基线4周/团队中位数)
→ 试算评分 + 异常规则 [management/]
→ 周/月快照(不可变) [management/snapshots]
```
- **4h**:`analytics_tick` 追加第三步:fact 滚动重算(近 3 天活跃会话)→ 宽表重算(当日+滚动 3 天);失败只记日志不中断(沿用现有异常隔离模式);
- **夜间**:full_sync_all.sh 第 13 步 fact 全量重建、第 14 步宽表全量重建(顺序:必须在 msg_daily_stats/stat_daily_staff 之后);
- **生命周期历史快照**:补同步触及历史日期时不能只重算那几天(那天的「距上次互动天数」依赖更早历史),统一由夜间全量按会话日序列重放覆盖;4h tick 只算当日;
- **通话后跟进**(call_followup_count):3,190 条通话量级小,逐通话窄查 24h 内员工对该客户的消息(join 键以 Task 0 核实结果为准),聚合任务内完成;
- **性能原则**:管理页面禁止实时扫描 96 万条原始消息;趋势和员工表读宽表+fact;原始消息只用于小范围下钻。
---
## 5. 口径决议表(管理域 vs 现有看板)
| 指标 | 管理域口径 | 现有看板口径 | 差异原因 |
|---|---|---|---|
| 消息量/串/首响均值 | 1:1(friend-only) | friend+room 混合 | 服务质量语义要求 1:1 |
| 首响分位 | 跨日精确串(fact) | 日切近似(跨夜丢失) | P90 正确性 |
| 降温客户 | 距今天数区间(同 churn 页) | 同 | 一期统一,频次基线降温二期 |
| 活跃员工 | 宽表口径(幽灵员工已过滤,§1.3 #10) | `_active_staff` 含映射不上的账号并读侧标注 | 评分场景不接受幽灵员工 |
| 交互天数/日桶 | Asia/Shanghai 自然日 | 同 | 一致 |
三方差异(管理页 vs 数据看板 vs 云客原版)在 metric-definitions 端点与页面口径说明中如实标注,不做静默差异。
---
## 6. API 设计(`/api/management/*`)
沿用 v1 §12.3 的 11 个端点与通用响应约定(比例 0~1、时长带单位后缀、空值≠0、稳定 error_code、data_as_of+metric_version+sample_status 必返),实现落点:
| 端点 | 关键实现落点 |
|---|---|
| `GET /dashboard` | 一次返回 KPI/双趋势/生命周期/维度分布/异常/行动聚合;data_as_of 复用 health.py |
| `GET /employees` | 服务端分页排序;默认按异常严重度排(不做总分榜);每员工带四对比值+sample_status |
| `GET /employees/{user_id}` | 首屏只回头部+维度分+趋势摘要+异常+辅导摘要 |
| `GET /employees/{user_id}/metrics?section=` | service/customer/workload/calls_groups 四页签懒加载;行动/风险页签一期不建 |
| `GET /employees/{user_id}/drilldown?metric_code=` | **白名单**:`first_reply_p90`→fact 逐串明细(含跳聊天定位)、`unreplied`→复用 B2、`cooling_customers`→客户列表带筛选、`call_followup`→通话列表。禁止前端传 SQL 字段/表达式 |
| `GET /metric-definitions` | 指标元数据(口径/单位/方向/样本门槛/公式/版本),含 §5 差异标注 |
| `GET/POST /reviews` | 草稿幂等生成(同 period+config 已有 draft 返回既有批次);`data_complete=false` 拒绝冻结 |
| `POST /reviews/{batch_id}/freeze` | 仅 draft→frozen;MVP 不做 published(RBAC 前置) |
| `GET/POST/PATCH /followups` | 状态机白名单流转 |
**周期分位数**:SQL 窗口函数 `PERCENT_RANK() OVER (PARTITION BY user_id ORDER BY fr_ms)` 取最接近 0.5/0.9 分位的行,双方言一致。
**一期评分**:综合分 = 可用维度(客户经营 30 + 服务质量 25 + 有效负载 15 = 70%)按 `100/70` 归一化 + **「试算」角标**;行动执行/风险维度显示「待数据」,不补 0 不补满分(v1 §10.1 默认)。
---
## 7. 前端设计
- **路由**:`/management/dashboard`、`/management/employees`、`/management/employees/:userId`(activeMenu 高亮父级)、`/management/reviews`;MainLayout 新增「经营管理」分组(二期菜单不渲染、不放空页面);`/analytics/dashboard` 过渡期并存、稳定后重定向;
- **目录**:`views/management/` 四页 + `components/management/`(ManagementFilterBar/ManagementKpiCard/ComparisonValue/DataAsOfBadge/ScoreBar/ScoreBreakdown/EmployeeMetricsTable/MetricDefinitionDrawer/MetricDrilldownDrawer/AnomalyList/FollowupDrawer)+ `composables/useManagementFilters.ts`(URL query 为真源)+ `api/management.ts`;
- **复用清单**:useChart(维度分布用分组条形图,BarChart 已注册,**不引雷达图**)、axios client 的 token/401 处理、`utils/format.ts`、`csvExport.ts`、seq 守卫防旧响应覆盖、DashboardView 的骨架屏/空态分语模式;
- **展示规范**:样本不足显示 `--` 与「样本不足」(非 0 分);「越低越好」指标反转趋势色;1366px 主表可用;导出与当前筛选排序一致。
---
## 8. 任务拆分(v1 Task 0-13 修正版)
| 任务 | 内容 | 相对 v1 的变化 | 估时 |
|---|---|---|---|
| Task 0 | 口径冻结:回测员工名单(3-5 名覆盖微信为主/WA 为主/群为主/高负载/低样本)+ 默认口径记录成文档(配置落库在 Task 3 种子 CLI,本任务不改代码)+ **五项核实**(phone_call 表结构与客户标识 join 键、call_followup 多账号语义——员工任一账号跟进即算、customer 归属字段与换人规则、员工主数据来源、群占比抽样量化混合口径偏差) | 默认值已确认,工作量缩减 | 0.5 天 |
| Task 1 | 5 张表模型 + migration + 模型冒烟测试 | 6→5 张表;核实项前置完成 | 1-1.5 天 |
| Task 2 | fact 独立计算模块(跨日精确算法,同源输出每日串聚合供宽表)+ 宽表生成(消息计数 friend 自算+串类取 fact+生命周期+通话)+ 生命周期共享函数提炼(analytics 调用点不动)+ 两条回填 CLI | **删除「_thread_stats 兼容扩展」**(不再需要);验收含对账判据(§1.1 #4/#5)+ 写钩子零改动确认 | 2.5-3.5 天 |
| Task 3 | scoring.py:归一化/四对比/样本保护 + 配置校验 + **默认配置 v1.0 种子落库**(受控 CLI,Task 11 快照的前置依赖) | 含 100/70 归一试算逻辑 | 2 天 |
| Task 4 | anomalies.py 规则集 | 数据源:fact(P90 恶化)、B2(待回复增长)、宽表(骤降/过载) | 1 天 |
| Task 5 | 管理 API(dashboard/employees/detail/metrics/drilldown/metric-definitions) | 窗口函数分位 + 性能目标 | 2-3 天 |
| Task 6 | 前端公共组件与路由(api/management.ts、useManagementFilters、components、router、MainLayout) | 按文档 | 1.5-2 天 |
| Task 7 | ManagementDashboardView(KPI/双趋势/生命周期/异常榜/下钻) | 复用图表模式,估时较 v1 略降 | 2-2.5 天 |
| Task 8 | EmployeeAnalyticsView(一期 4 视图+分页排序+指标抽屉+员工对比) | 行动/风险视图不建 | 2-3 天 |
| Task 9 | EmployeeAnalyticsDetailView(概览/服务质量/客户经营/工作负载/通话与群五页签) | 按文档 | 3-4 天 |
| Task 10 | followups 闭环(查询/创建/更新/状态流转测试) | 按文档 | 1.5-2 天 |
| Task 11 | 考核快照(草稿/详情/冻结/重算替代/列表/CSV) | 按文档 | 2-3 天 |
| Task 12 | 调度接入:analytics_tick 第三步 + full_sync 第 13/14 步 + 失败隔离 + 手动重跑入口 | 步数 12→14;**服务器 scheduler 镜像重建遗留**(9d3cc8e 部署时未重建)一并处理 | 1-1.5 天 |
| Task 13 | 回测校准 + 上线顺序控制(本地→服务器)+ 试运行记录 | 功能开关→部署顺序控制 | 2-3 天 |
**批次划分**(每批独立验收,不夹带无关重构):
| 批次 | 任务 | 可见结果 |
|---|---|---|
| A | Task 1-2 | 表+fact+宽表可在测试与脚本中查询,回填可重跑 |
| B | Task 3-5 | OpenAPI 可验证全部只读管理接口 |
| C | Task 6-7 | 新导航+驾驶舱可用,异常两次点击到事实 |
| D | Task 8-9 | 员工总览+详情可比较可下钻 |
| E | Task 10-11 | 辅导闭环+冻结周期结果 |
| F | Task 12-13 | 持续运行+内部试运行 |
**总工期**:单链 18-24 个工作日;前后端并行 12-16 天。口径确认(已完成)、数据回测和试运行不可省略。
---
## 9. 测试与验收(v1 §16 基础上新增判别项)
**后端单元/API 测试**(新增文件按 v1 §16.1 清单):
- **跨夜串判别**(P1#1 防线):seed 客户 23:00 消息+员工次日 9:00 回复,断言 fact 产生 fr_ms≈10h 的行——正是日切口径丢失的场景;
- **口径判别**(P1#2 防线):seed 同员工 friend+room 两类会话,断言宽表 1:1 列不含 room 串;
- **分位数精确性**:构造已知 fr_ms 集合,断言周期 P50/P90 精确值;窗口函数 SQLite/MySQL 双方言一致;
- **fact 超集断言**(P1#4):fact ⊇ 日切被回串,差集=跨夜串,不做相等断言;
- **分子分母同源**(P1#5):任意窗口 reply_rate ≤ 1;宽表 threads_answered ≡ fact 行数(同窗口);存在多日未回串时宽表 cust_threads < 日切口径值(分母不再膨胀);
- **fact 回填幂等**:重跑不重不漏;
- **快照不可变**:冻结后 PATCH 拒绝(后端强制);
- 评分边界:正/反向指标、阈值边界、权重合计、样本不足不评分、空值≠0、缺失维度不默认 0/100;
- 下钻白名单:非法 metric_code 拒绝;
- 幽灵员工过滤:映射不上的 owner 不进宽表。
**数据核验**(3-5 名回测员工,覆盖微信为主/WA 为主/群为主/高负载/低样本):抽查 3 日期消息与活跃客户、20 个首响串(对照 fact 明细)、10 个待回复客户、5 条通话后跟进、生命周期变化、评分展开明细;对账点增加宽表消息计数列(msg_sent/msg_recv)↔ msg_daily_stats friend 聚合——**串类列禁止相等断言**(口径差异见 §1.1 #4/#5)。
**性能验收**(v1 §16.5 全部目标基础上追加):员工列表服务质量视图 30 天窗(含 fact 窗口函数分位)p95 ≤ 1.5s。
**前端**:`npm run build` + Vitest(格式化/趋势方向/样本状态/筛选 query/旧响应不覆盖)。
---
## 10. 部署注记(写入 DEPLOY.md)
1. 服务器 backend 容器**不自动跑 alembic**——升级带迁移必须手动 `sudo docker exec <backend> alembic upgrade head`;
2. 两条回填 CLI(`first-reply-facts` / `employee-metrics`)部署后执行一次,DEPLOY.md §7 注记(含耗时参考与避抓取窗口建议);
3. scheduler 镜像重建(含 9d3cc8e 遗留):`cd /opt/yunke && sudo docker compose -f docker-compose.server.yml --env-file .env up -d --build scheduler`——重建前 fr 三列/第 11-14 步不自愈;
4. fact 表索引创建量级小(几十万行),无需特殊避峰;
5. 上线顺序:本地 Docker 全量回填对账 → 本地试用 → 服务器升级+回填 → 管理者只读试用。
---
## 11. 沿用 v1 不变的部分(索引)
以下内容直接按 v1 执行,本文不重复:
- 页面内容细则:v1 §5-§9(驾驶舱布局/KPI 卡清单、员工总览六视图、员工详情页签、快照页、二三期概要);
- 评分标准化公式与等级:v1 §10.2/§10.4;
- 后端模块目录与分层职责:v1 §12.1/§12.2(其中 storage 两模块以本文 §3/§4 为准);
- API 通用响应约定全文:v1 §12.4;
- 前端可视化规范全文:v1 §13.4;
- 验收标准:v1 §19;
- 灰度顺序/影子考核/回滚:v1 §20(其中「功能开关」按本文 §1.3 #11 替换为上线顺序控制)。
+92 -1
View File
@@ -7,6 +7,12 @@ Read-only query over the mirrored ``customer`` table. Supports filtering by
nickname / remark / friend_id / phone, ``%``/``_`` escaped as literals — the nickname / remark / friend_id / phone, ``%``/``_`` escaped as literals — the
通讯录 search box). Results are ordered by ``last_chat_time desc`` (nulls 通讯录 search box). Results are ordered by ``last_chat_time desc`` (nulls
last) then ``friend_id`` so the most-recently-active contacts surface first. last) then ``friend_id`` so the most-recently-active contacts surface first.
**owner_account_id 维度走 ``customer_account_link``**(2026-09-08):customer
档案 PK=friend_id 全局单行、归属 last-wins,共享客户在其他员工名下会漏
(实测 36 vs 12);link 表每个「渠道×账号×客户」一行,员工维度列表/计数与
云客 per-employee 口径一致。带 department/user/add_time/last_chat 等其余
过滤时回退档案路径(组合场景罕见,保持旧行为)。
""" """
from __future__ import annotations from __future__ import annotations
@@ -19,7 +25,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_session from app.api.deps import get_session
from app.api.schemas import CustomerOut, Page from app.api.schemas import CustomerOut, Page
from app.db.models import Customer, Staff, WechatAccount from app.db.models import Customer, CustomerAccountLink, Staff, WechatAccount
router = APIRouter(prefix="/api", tags=["customers"]) router = APIRouter(prefix="/api", tags=["customers"])
@@ -61,6 +67,63 @@ async def customers_declared_total(session: SessionDep) -> dict:
} }
async def _list_owner_links(
session: AsyncSession,
*,
page: int,
page_size: int,
channel: str | None,
owner_account_id: str,
friend_id: str | None,
q: str | None,
) -> Page[CustomerOut]:
"""员工维度客户列表:读 ``customer_account_link``(渠道×账号×客户一行)。
排序/搜索语义与档案路径一致(last_chat_time desc、四列模糊);link 自带
per-relation 快照(备注/收发计数按员工各一份,与云客 per-employee 列表
相同)。CustomerOut 的档案专属列(raw_json/version/send_today 等)为 None。
"""
conditions = [CustomerAccountLink.owner_account_id == owner_account_id]
if channel is not None:
conditions.append(CustomerAccountLink.channel == channel)
if friend_id is not None:
conditions.append(CustomerAccountLink.friend_id == friend_id)
if q:
# LIKE 通配符转义(同档案路径):昵称/备注/微信号/手机号四列任一命中
escaped = q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
like = f"%{escaped}%"
conditions.append(
sa.or_(
CustomerAccountLink.friend_nickname.like(like, escape="\\"),
CustomerAccountLink.friend_remark.like(like, escape="\\"),
CustomerAccountLink.friend_id.like(like, escape="\\"),
CustomerAccountLink.phone.like(like, escape="\\"),
)
)
total = (
await session.execute(
sa.select(sa.func.count())
.select_from(CustomerAccountLink)
.where(*conditions)
)
).scalar_one()
rows = (
await session.execute(
sa.select(CustomerAccountLink)
.where(*conditions)
.order_by(
sa.nullslast(CustomerAccountLink.last_chat_time.desc()),
CustomerAccountLink.friend_id.asc(),
)
.limit(page_size)
.offset((page - 1) * page_size)
)
).scalars().all()
items = [CustomerOut.model_validate(r) for r in rows]
return Page(items=items, total=total, page=page, page_size=page_size)
@router.get("/customers") @router.get("/customers")
async def list_customers( async def list_customers(
session: SessionDep, session: SessionDep,
@@ -76,6 +139,34 @@ async def list_customers(
q: Annotated[str | None, Query()] = None, q: Annotated[str | None, Query()] = None,
) -> Page[CustomerOut]: ) -> Page[CustomerOut]:
"""List customers with optional filters and pagination.""" """List customers with optional filters and pagination."""
# 员工维度优先走 link 表(共享好友每账号一行,见模块 docstring);该账号
# link 还无数据时回退档案路径(过渡安全:回填前页面不至于空;也兼容
# 只写档案的理论路径)。与其余仅档案路径支持的过滤组合时同样回退。
if (
owner_account_id is not None
and department_id is None
and user_id is None
and add_time_since is None
and last_chat_since is None
):
link_exists = (
await session.execute(
sa.select(sa.func.count())
.select_from(CustomerAccountLink)
.where(CustomerAccountLink.owner_account_id == owner_account_id)
)
).scalar_one()
if link_exists:
return await _list_owner_links(
session,
page=page,
page_size=page_size,
channel=channel,
owner_account_id=owner_account_id,
friend_id=friend_id,
q=q,
)
conditions = [] conditions = []
if channel is not None: if channel is not None:
conditions.append(Customer.channel == channel) conditions.append(Customer.channel == channel)
+3
View File
@@ -43,6 +43,9 @@ class CustomerOut(BaseModel):
# 客户页「发送条数/接收条数」列用,2026-09-08) # 客户页「发送条数/接收条数」列用,2026-09-08)
send_num: int | None = None send_num: int | None = None
receive_num: int | None = None receive_num: int | None = None
# 云客数据来源代码(customer_account_link 路径直读;档案路径为 None,
# 前端回退 raw_json.fromType)
from_type: int | None = None
version: str | None = None version: str | None = None
# Verbatim upstream friend record (~60 camelCase fields). Surfaced so the # Verbatim upstream friend record (~60 camelCase fields). Surfaced so the
# all-fields drawer can render every column the 云客 source page shows # all-fields drawer can render every column the 云客 source page shows
+40
View File
@@ -262,6 +262,46 @@ class Customer(Base):
) )
class CustomerAccountLink(Base):
"""账号↔客户 多对多关联(customer_account_link,2026-09-08)。
``customer`` 表 PK=``friend_id`` 全局一客户一行,归属=最后抓取员工
(last-wins)——共享客户(同一人加多个员工)在其他员工名下「消失」:
实测 Lily 云客好友 36 vs 镜像页面 12,24 个全在别的员工名下。云客
的口径是**每个账号各算一个**(per-employee 列表连备注/收发计数都按
关系各存一份);本表每个「渠道×账号×客户」一行,由
:func:`app.storage.repos.upsert_customer` 随档案同事务维护,员工维度
的列表/计数(`/api/customers?owner_account_id=`)读这里。
快照列为员工客户页所需(昵称/备注/头像 per-relation);全字段档案
(raw_json/60 列)仍在 ``customer`` 表按 friend_id 关联。
"""
__tablename__ = "customer_account_link"
channel: Mapped[str] = mapped_column(String(16), primary_key=True)
owner_account_id: Mapped[str] = mapped_column(String(128), primary_key=True)
friend_id: Mapped[str] = mapped_column(String(128), primary_key=True)
# per-relation 展示快照(云客好友记录按员工各一份;备注是员工自己的)
friend_nickname: Mapped[str | None] = mapped_column(String(255), nullable=True)
friend_remark: Mapped[str | None] = mapped_column(String(255), nullable=True)
head_image_url: Mapped[str | None] = mapped_column(String(1024), nullable=True)
real_name: Mapped[str | None] = mapped_column(String(64), nullable=True)
phone: Mapped[str | None] = mapped_column(String(64), nullable=True)
last_chat_time: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
add_time: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
send_num: Mapped[int | None] = mapped_column(Integer, nullable=True)
receive_num: Mapped[int | None] = mapped_column(Integer, nullable=True)
message_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
# 云客 fromType 代码(数据来源列;int,前端 fromTypeLabel 映射)
from_type: Mapped[int | None] = mapped_column(Integer, nullable=True)
# 同 customer.delete_flag 口径(int 0/1)
delete_flag: Mapped[int | None] = mapped_column(Integer, nullable=True)
fetched_at: Mapped[dt.datetime] = _fetched_at()
updated_at: Mapped[dt.datetime] = _updated_at()
class ChatGroup(Base): class ChatGroup(Base):
"""聊天群 — source: ``POST /wechatweb/chatroom/getChatroomList``. """聊天群 — source: ``POST /wechatweb/chatroom/getChatroomList``.
+52 -2
View File
@@ -63,6 +63,7 @@ from sqlalchemy.orm import DeclarativeBase
from app.db.models import ( from app.db.models import (
ChatGroup, ChatGroup,
Customer, Customer,
CustomerAccountLink,
Department, Department,
GroupMember, GroupMember,
Media, Media,
@@ -83,6 +84,7 @@ __all__ = [
"upsert_chat_group", "upsert_chat_group",
"upsert_customer", "upsert_customer",
"upsert_customers", "upsert_customers",
"prune_customer_links",
"upsert_department", "upsert_department",
"upsert_media", "upsert_media",
"upsert_messages", "upsert_messages",
@@ -380,9 +382,34 @@ async def upsert_customer(session: AsyncSession, rec: dict[str, Any]) -> None:
"""Idempotently upsert one :class:`Customer` (PK = ``friend_id``). """Idempotently upsert one :class:`Customer` (PK = ``friend_id``).
Same ``friend_id`` twice -> one row, with non-PK fields refreshed from the Same ``friend_id`` twice -> one row, with non-PK fields refreshed from the
later record. later record. 同时维护 ``customer_account_link``(渠道×账号×客户一行,
2026-09-08):customer 档案全局单行 last-wins,共享客户的其他员工关系
靠 link 表保留;员工维度列表/计数读 link(对齐云客 per-employee 口径)。
""" """
await _upsert_one(session, Customer, rec) await _upsert_one(session, Customer, rec)
owner = rec.get("owner_account_id")
if owner:
raw = rec.get("raw_json") or {}
link_rec = {
"channel": rec.get("channel"),
"owner_account_id": owner,
"friend_id": rec.get("friend_id"),
"friend_nickname": rec.get("friend_nickname"),
"friend_remark": rec.get("friend_remark"),
"head_image_url": rec.get("head_image_url"),
"real_name": rec.get("real_name"),
"phone": rec.get("phone"),
"last_chat_time": rec.get("last_chat_time"),
"add_time": rec.get("add_time"),
"send_num": rec.get("send_num"),
"receive_num": rec.get("receive_num"),
"message_count": rec.get("message_count"),
"delete_flag": rec.get("delete_flag"),
}
# fromType 只在 raw_json(云客数据来源代码,员工客户页列)
if isinstance(raw, dict) and raw.get("fromType") is not None:
link_rec["from_type"] = raw.get("fromType")
await _upsert_one(session, CustomerAccountLink, link_rec)
async def upsert_customers(session: AsyncSession, recs: list[dict[str, Any]]) -> int: async def upsert_customers(session: AsyncSession, recs: list[dict[str, Any]]) -> int:
@@ -392,10 +419,33 @@ async def upsert_customers(session: AsyncSession, recs: list[dict[str, Any]]) ->
single row (the last write wins) at the DB layer. single row (the last write wins) at the DB layer.
""" """
for rec in recs: for rec in recs:
await _upsert_one(session, Customer, rec) await upsert_customer(session, rec)
return len(recs) return len(recs)
async def prune_customer_links(
session: AsyncSession,
*,
channel: str,
owner_account_id: str,
keep_friend_ids: list[str],
) -> int:
"""删除该账号名下**不在最新名单**里的 customer_account_link 行。
云客 isDel=0 只返回当前好友;抓取器逐员工整批 upsert 后调用本函数对账
(离开名单=已不是好友,或好友转移)。**调用方必须保证名单完整**(翻满
声明 total)再调——拿到半份名单就 prune 会误删。返回删除行数。
"""
conditions = [
CustomerAccountLink.channel == channel,
CustomerAccountLink.owner_account_id == owner_account_id,
]
if keep_friend_ids:
conditions.append(CustomerAccountLink.friend_id.not_in(keep_friend_ids))
result = await session.execute(sa.delete(CustomerAccountLink).where(*conditions))
return int(result.rowcount or 0)
async def upsert_messages(session: AsyncSession, recs: list[dict[str, Any]]) -> int: async def upsert_messages(session: AsyncSession, recs: list[dict[str, Any]]) -> int:
"""Batch idempotent upsert for :class:`Message` (PK = ``msg_id``). """Batch idempotent upsert for :class:`Message` (PK = ``msg_id``).
@@ -0,0 +1,78 @@
"""customer_account_link 账号↔客户多对多关联表
Revision ID: f7a8b9c0d1e2
Revises: e6f7a8b9c0d1
Create Date: 2026-09-08
内容
----
customer 表 PK=friend_id 全局一客户一行、归属=最后抓取员工(last-wins)。
共享客户(同一人加多个员工)在镜像里只挂在最后抓的员工名下——实测 Lily
云客好友 36,镜像员工客户页只有 12,其余 24 个全在另外 9 个员工名下
(2026-09-08 用户报「好友数与点进去的数量不一致」真机探明)。
云客口径是**每个账号各算一个**(per-employee 列表连备注/收发计数都按
关系各一份)。本表每个「渠道×账号×客户」一行:
- 写:repos.upsert_customer 随档案同事务维护(所有好友抓取路径一处挂全);
- 读:/api/customers?owner_account_id= 改读本表(员工维度列表/计数),
与云客 contacts_num 完全对齐;
- 对账:scrape_friends 逐员工整批 upsert 后删除该员工名下不在最新名单
里的 link(云客 isDel=0 返回当前好友,删链=已不是好友)。
**上线后需重跑 friends + wa-friends 各一次回填 link**(server:
`docker compose exec backend python scripts/scrape_via_page.py --entity friends`)
——存量关系无法从单行档案反推。
幂等:sa.inspect 先查表存在则跳过。SQLite/MySQL 通用。
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = 'f7a8b9c0d1e2'
down_revision: str | Sequence[str] | None = 'e6f7a8b9c0d1'
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
_TABLE = 'customer_account_link'
def _has_table(table: str) -> bool:
insp = sa.inspect(op.get_bind())
return table in insp.get_table_names()
def upgrade() -> None:
"""Upgrade schema."""
if _has_table(_TABLE):
return # 已建过(或先手工建过),幂等跳过
op.create_table(
_TABLE,
sa.Column('channel', sa.String(16), primary_key=True),
sa.Column('owner_account_id', sa.String(128), primary_key=True),
sa.Column('friend_id', sa.String(128), primary_key=True),
sa.Column('friend_nickname', sa.String(255), nullable=True),
sa.Column('friend_remark', sa.String(255), nullable=True),
sa.Column('head_image_url', sa.String(1024), nullable=True),
sa.Column('real_name', sa.String(64), nullable=True),
sa.Column('phone', sa.String(64), nullable=True),
sa.Column('last_chat_time', sa.BigInteger(), nullable=True),
sa.Column('add_time', sa.BigInteger(), nullable=True),
sa.Column('send_num', sa.Integer(), nullable=True),
sa.Column('receive_num', sa.Integer(), nullable=True),
sa.Column('message_count', sa.Integer(), nullable=True),
sa.Column('from_type', sa.Integer(), nullable=True),
sa.Column('delete_flag', sa.Integer(), nullable=True),
sa.Column('fetched_at', sa.DateTime(), nullable=True),
sa.Column('updated_at', sa.DateTime(), nullable=True),
)
def downgrade() -> None:
"""Downgrade schema."""
if not _has_table(_TABLE):
return
op.drop_table(_TABLE)
+30 -6
View File
@@ -267,10 +267,10 @@ async def scrape_friends(cookie_str: str, max_employees: int = 100):
小漂移可忽略,且与云客后台「按员工看客户列表」口径一致。员工清单来自 小漂移可忽略,且与云客后台「按员工看客户列表」口径一致。员工清单来自
本地 wechat_account(user_id 列)。 本地 wechat_account(user_id 列)。
""" """
from app.db.base import async_engine from app.db.base import async_engine, async_session_maker
from sqlalchemy import text as sa_text from sqlalchemy import text as sa_text
from app.scraper.wechat import _normalize from app.scraper.wechat import _normalize
from app.storage.repos import upsert_customer from app.storage.repos import prune_customer_links, upsert_customer
dept = await _root_dept_id() dept = await _root_dept_id()
async with async_engine.connect() as conn: async with async_engine.connect() as conn:
@@ -334,9 +334,21 @@ async def scrape_friends(cookie_str: str, max_employees: int = 100):
if records: if records:
await _upsert_all(upsert_customer, records) await _upsert_all(upsert_customer, records)
total_upserted += len(records) total_upserted += len(records)
# 对账(2026-09-08):云客 isDel=0 返回当前好友;名单取满(>=声明
# total)才可信,删掉该账号名下已不在名单里的 link(转移/删好友)。
# 半份名单宁可不对账,防误删。
pruned = 0
if declared is not None and len(rows) >= declared:
keep_ids = [r.get("friendWxId") for r in rows if r.get("friendWxId")]
async with async_session_maker() as session:
pruned = await prune_customer_links(
session, channel="wechat", owner_account_id=wx_id,
keep_friend_ids=keep_ids,
)
await session.commit()
logger.info( logger.info(
f" [{i+1}/{len(employees)}] {wx_id[:16]}..: declared={declared} " f" [{i+1}/{len(employees)}] {wx_id[:16]}..: declared={declared} "
f"rows={len(rows)} (total fetched {total_fetched})" f"rows={len(rows)} pruned={pruned} (total fetched {total_fetched})"
) )
await asyncio.sleep(2 + (i % 3)) await asyncio.sleep(2 + (i % 3))
@@ -1463,10 +1475,10 @@ async def scrape_wa_friends(cookie_str: str, max_pages: int = 2000):
全局查询有 ~1 万行硬上限(实测云客声明 11931 vs 全局翻页只到 10000, 全局查询有 ~1 万行硬上限(实测云客声明 11931 vs 全局翻页只到 10000,
2026-08-20);逐员工 ``uid`` 过滤(员工 WA 号)范围小无上限,页码字段是 2026-08-20);逐员工 ``uid`` 过滤(员工 WA 号)范围小无上限,页码字段是
``page``,total-aware 翻页(短页继续,满声明数或连续空页停)。""" ``page``,total-aware 翻页(短页继续,满声明数或连续空页停)。"""
from app.db.base import async_engine from app.db.base import async_engine, async_session_maker
from sqlalchemy import text as sa_text from sqlalchemy import text as sa_text
from app.scraper.whatsapp import _normalize_whatsapp_friend from app.scraper.whatsapp import _normalize_whatsapp_friend
from app.storage.repos import upsert_customer from app.storage.repos import prune_customer_links, upsert_customer
dept = await _root_dept_id() dept = await _root_dept_id()
async with async_engine.connect() as conn: async with async_engine.connect() as conn:
@@ -1520,8 +1532,20 @@ async def scrape_wa_friends(cookie_str: str, max_pages: int = 2000):
if records: if records:
await _upsert_all(upsert_customer, records) await _upsert_all(upsert_customer, records)
total_upserted += len(records) total_upserted += len(records)
# 对账(2026-09-08,同微信 friends):名单取满才删该账号名下
# 不在当前列表里的 link。
pruned = 0
if declared is not None and len(rows) >= declared:
keep_ids = [r.get("friendUid") for r in rows if r.get("friendUid")]
async with async_session_maker() as session:
pruned = await prune_customer_links(
session, channel="whatsapp", owner_account_id=wa_uid,
keep_friend_ids=keep_ids,
)
await session.commit()
logger.info( logger.info(
f" [{i+1}/{len(accounts)}] {wa_uid[:18]}..: declared={declared} rows={len(rows)}" f" [{i+1}/{len(accounts)}] {wa_uid[:18]}..: declared={declared} "
f"rows={len(rows)} pruned={pruned}"
) )
await asyncio.sleep(2 + (i % 3)) await asyncio.sleep(2 + (i % 3))
+78
View File
@@ -148,3 +148,81 @@ async def test_q_escapes_like_wildcards(make_client):
# 只命中字面含 "100%" 的行;不因 % 通配漏扩到 "1001" # 只命中字面含 "100%" 的行;不因 % 通配漏扩到 "1001"
assert body["total"] == 1 assert body["total"] == 1
assert [it["friend_id"] for it in body["items"]] == ["wx_pct"] assert [it["friend_id"] for it in body["items"]] == ["wx_pct"]
# ---------------------------------------------------------------------------
# owner 维度读 customer_account_link(共享好友,2026-09-08)
# ---------------------------------------------------------------------------
async def _seed_links(session):
from app.db.models import CustomerAccountLink as L
# 档案路径的旧行:共享好友 f_shared 档案 last-wins 挂在 wx_b 名下
session.add_all([
L(channel="wechat", owner_account_id="wx_a", friend_id="f_shared",
friend_remark="A的备注", friend_nickname="共享客户", last_chat_time=2000,
send_num=3, receive_num=1, from_type=1),
L(channel="wechat", owner_account_id="wx_b", friend_id="f_shared",
friend_remark="B的备注", friend_nickname="共享客户", last_chat_time=1000),
L(channel="wechat", owner_account_id="wx_a", friend_id="f_only_a",
friend_nickname="A独有", last_chat_time=3000),
])
await session.commit()
async def test_owner_filter_reads_links(make_client):
"""owner 过滤走 link:共享好友在两个员工名下各出现,备注各是各的。"""
client = await make_client(_seed_links)
r_a = await client.get("/api/customers", params={
"channel": "wechat", "owner_account_id": "wx_a"})
assert r_a.status_code == 200
body_a = r_a.json()
assert body_a["total"] == 2
by_id = {it["friend_id"]: it for it in body_a["items"]}
# 排序:last_chat_time desc
assert [it["friend_id"] for it in body_a["items"]] == ["f_only_a", "f_shared"]
assert by_id["f_shared"]["friend_remark"] == "A的备注"
assert by_id["f_shared"]["send_num"] == 3
assert by_id["f_shared"]["from_type"] == 1
r_b = await client.get("/api/customers", params={
"channel": "wechat", "owner_account_id": "wx_b"})
body_b = r_b.json()
assert body_b["total"] == 1
assert body_b["items"][0]["friend_remark"] == "B的备注"
async def test_owner_filter_q_and_friend_id_on_links(make_client):
"""link 路径的 q 四列搜索 + friend_id 精确查找(聊天窗名字解析用)。"""
client = await make_client(_seed_links)
r = await client.get("/api/customers", params={
"channel": "wechat", "owner_account_id": "wx_a", "q": "A的备注"})
assert r.status_code == 200
body = r.json()
assert [it["friend_id"] for it in body["items"]] == ["f_shared"]
r2 = await client.get("/api/customers", params={
"channel": "wechat", "owner_account_id": "wx_a", "friend_id": "f_shared"})
body2 = r2.json()
assert body2["total"] == 1
assert body2["items"][0]["friend_nickname"] == "共享客户"
async def test_owner_filter_with_dept_falls_back_to_archive(make_client):
"""owner+department 组合回退档案路径(旧行为;link 不解析部门)。"""
from app.db.models import Customer as C, Staff as S
async def seed(session):
await _seed_links(session)
session.add(S(user_id="u1", name="员工", dept_id="d1"))
session.add(C(channel="wechat", owner_account_id="wx_a",
friend_id="f_archived", user_id="u1"))
await session.commit()
client = await make_client(seed)
r = await client.get("/api/customers", params={
"channel": "wechat", "owner_account_id": "wx_a", "department_id": "d1"})
assert r.status_code == 200
body = r.json()
# 档案路径:只命中 user_id 属于 d1 的档案行(link 行不带 user_id)
assert [it["friend_id"] for it in body["items"]] == ["f_archived"]
+112
View File
@@ -25,6 +25,7 @@ from app.db.base import Base
from app.db.models import ( from app.db.models import (
ChatGroup, ChatGroup,
Customer, Customer,
CustomerAccountLink,
Department, Department,
GroupMember, GroupMember,
Message, Message,
@@ -39,6 +40,7 @@ from app.storage.repos import (
_build_column_values, _build_column_values,
_coerce_bool, _coerce_bool,
_coerce_int, _coerce_int,
prune_customer_links,
upsert_chat_group, upsert_chat_group,
upsert_customer, upsert_customer,
upsert_customers, upsert_customers,
@@ -107,6 +109,116 @@ def test_coerce_int_returns_none_for_non_numeric_string():
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
async def test_upsert_customer_shared_friend_maintains_links():
"""共享好友:档案全局单行(last-wins),link 每个账号各一行(2026-09-08)。
同一 friend_id 先后在 ownerA/ownerB 名下 upsert——customer 只有一行
且归属=后写;customer_account_link 两行,各自带自己的备注快照与
fromType(来自 raw_json)。
"""
engine = await _make_engine()
try:
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
async with SessionLocal() as session:
base = {
"friend_id": "wxid_shared",
"channel": "wechat",
"friend_nickname": "共享客户",
"phone": "13800000000",
"send_num": 10,
"receive_num": 4,
}
await upsert_customer(session, {
**base, "owner_account_id": "wx_a",
"friend_remark": "A的备注", "raw_json": {"fromType": 1},
})
await session.commit()
await upsert_customer(session, {
**base, "owner_account_id": "wx_b",
"friend_remark": "B的备注", "raw_json": {"fromType": 2},
})
await session.commit()
custs = (await session.execute(
sa.select(Customer).execution_options(populate_existing=True)
)).scalars().all()
assert len(custs) == 1
assert custs[0].owner_account_id == "wx_b" # 档案 last-wins
links = (await session.execute(
sa.select(CustomerAccountLink).execution_options(populate_existing=True)
)).scalars().all()
assert {(l.owner_account_id, l.friend_remark) for l in links} == {
("wx_a", "A的备注"), ("wx_b", "B的备注"),
}
by_owner = {l.owner_account_id: l for l in links}
assert by_owner["wx_a"].from_type == 1
assert by_owner["wx_b"].from_type == 2
assert by_owner["wx_a"].send_num == 10
assert by_owner["wx_a"].phone == "13800000000"
finally:
await engine.dispose()
async def test_upsert_customer_without_owner_skips_link():
"""无 owner_account_id 的档案 upsert(理论路径)只写 customer,不碰 link。"""
engine = await _make_engine()
try:
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
async with SessionLocal() as session:
await upsert_customer(session, {
"friend_id": "wxid_solo", "channel": "wechat",
})
await session.commit()
n_links = (await session.execute(
sa.select(sa.func.count()).select_from(CustomerAccountLink)
)).scalar_one()
assert n_links == 0
finally:
await engine.dispose()
async def test_prune_customer_links_removes_departed():
"""对账:保留名单外的 link 被删;空名单删光;他人名下不动。"""
engine = await _make_engine()
try:
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
async with SessionLocal() as session:
for owner, fid in [("wx_a", "f1"), ("wx_a", "f2"), ("wx_a", "f3"),
("wx_b", "f2")]:
await upsert_customer(session, {
"friend_id": fid, "channel": "wechat",
"owner_account_id": owner,
})
await session.commit()
deleted = await prune_customer_links(
session, channel="wechat", owner_account_id="wx_a",
keep_friend_ids=["f1", "f3"],
)
await session.commit()
assert deleted == 1
left = (await session.execute(
sa.select(CustomerAccountLink.friend_id, CustomerAccountLink.owner_account_id)
)).all()
assert set(left) == {("f1", "wx_a"), ("f3", "wx_a"), ("f2", "wx_b")}
# 空名单=该账号全部删除(wx_b 的保留)
deleted_all = await prune_customer_links(
session, channel="wechat", owner_account_id="wx_a",
keep_friend_ids=[],
)
await session.commit()
assert deleted_all == 2
left2 = (await session.execute(
sa.select(CustomerAccountLink.friend_id, CustomerAccountLink.owner_account_id)
)).all()
assert set(left2) == {("f2", "wx_b")}
finally:
await engine.dispose()
async def test_upsert_customer_is_idempotent_and_updates_version(): async def test_upsert_customer_is_idempotent_and_updates_version():
"""Upsert same friend_id twice with different version -> 1 row, version updated.""" """Upsert same friend_id twice with different version -> 1 row, version updated."""
engine = await _make_engine() engine = await _make_engine()
+3
View File
@@ -36,6 +36,9 @@ export interface Customer {
/** 累计发送/接收条数(同步自云客 friend 记录 sendNum/receiveNum;员工客户页列)。 */ /** 累计发送/接收条数(同步自云客 friend 记录 sendNum/receiveNum;员工客户页列)。 */
send_num?: number | null send_num?: number | null
receive_num?: number | null receive_num?: number | null
/** 云客数据来源代码(customer_account_link 员工维度路径直读;档案路径空,
* 前端回退 raw_json.fromType)。 */
from_type?: number | null
/** ms epoch of friend add time (from raw_json.addTime). */ /** ms epoch of friend add time (from raw_json.addTime). */
add_time?: number | null add_time?: number | null
message_count?: number | null message_count?: number | null
+1 -1
View File
@@ -395,7 +395,7 @@ onMounted(() => {
</template> </template>
</el-table-column> </el-table-column>
<el-table-column label="数据来源" width="110"> <el-table-column label="数据来源" width="110">
<template #default="{ row }">{{ fromTypeLabel(readField(row, 'fromType')) }}</template> <template #default="{ row }">{{ fromTypeLabel(row.from_type ?? readField(row, 'fromType')) }}</template>
</el-table-column> </el-table-column>
<el-table-column label="客户备注名" min-width="120"> <el-table-column label="客户备注名" min-width="120">
<template #default="{ row }">{{ row.friend_remark || '-' }}</template> <template #default="{ row }">{{ row.friend_remark || '-' }}</template>