From 0b975a8289832f478e7e0e7927367b0f50416498 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Thu, 10 Sep 2026 13:55:11 +0800 Subject: [PATCH] =?UTF-8?q?docs(management):=20=E6=89=B9=E6=AC=A1F?= =?UTF-8?q?=E8=B0=83=E5=BA=A6=E5=AD=90=E6=89=B9=E8=AE=A1=E5=88=92=E2=80=94?= =?UTF-8?q?=E2=80=94tick=E7=AE=A1=E7=90=86=E6=AE=B5+=E5=A4=9C=E9=97=B413/1?= =?UTF-8?q?4=E6=AD=A5(Task=20F1-F3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...agement-console-batch-f-scheduling-plan.md | 142 ++++++++++++++++++ 1 file changed, 142 insertions(+) create mode 100644 docs/superpowers/plans/2026-09-10-management-console-batch-f-scheduling-plan.md diff --git a/docs/superpowers/plans/2026-09-10-management-console-batch-f-scheduling-plan.md b/docs/superpowers/plans/2026-09-10-management-console-batch-f-scheduling-plan.md new file mode 100644 index 0000000..f8f69e2 --- /dev/null +++ b/docs/superpowers/plans/2026-09-10-management-console-batch-f-scheduling-plan.md @@ -0,0 +1,142 @@ +# 云客经营管理后台·批次F-调度子批（管理指标接入定时链）实施计划 + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** 管理域三张派生表（first_reply_fact / employee_metric_daily / employee_call_daily）进入生产定时链——4h 滚动 + 夜间全量自愈，数据每天活起来（同步审计 2026-09-10 结论 #5 的管理域部分）。 + +**Architecture:** 两处接线：`scheduler.py _analytics_tick` 加**独立事务**的管理滚动段（近 3 天，失败不回滚既有 staff/敏感词段）；`yunke-server/scripts/full_sync_all.sh` 在 11/12 步后加 13/14 步（fact 全量 → 宽表+通话全量）+ 终局快照补三表行数。CLI 已有（first-reply-facts / employee-metrics），无新端点。**边界：不碰审计文档 P0 的脚本加固/实体补抓（并行会话领地）**。 + +**Tech Stack:** 同批次A（APScheduler + SQLAlchemy async + bash）。 + +--- + +## 执行者必读 + +1. **上游函数签名（as-built）**：`first_reply_facts.pairs_active_in_days(session, days) / recompute_pairs(session, pairs)`；`storage/employee_metrics.recompute_metric_days(session, days) / recompute_call_days(session, days)`；`staff_summary.rolling_days(today, 3)`（返回当日+前2天列表）。全部幂等 delete+insert。 +2. **隔离语义（v2 Task 12：指标失败不中断主流程）**：tick 内管理段用**独立 session+commit**——失败只回滚管理段自身，staff/敏感词段已提交不受影响（当前实现是单 try 单事务，改造为两段）。 +3. **锁竞争既有结论**（tick docstring 已记录）：与夜间 full_sync 重叠时 DELETE+INSERT 可能锁等待，非致命，幂等下轮自愈——管理段沿用同一语义，docstring 注明。 +4. 脚本在 `yunke-server/scripts/full_sync_all.sh`（注意路径已从根 scripts/ 迁移）；测试先例 `tests/test_scheduler_analytics_tick.py`。 +5. 并行会话活动照旧；中文 conventional 提交。 + +## 任务 + +### Task F1: _analytics_tick 管理段 + 测试 + +**Files:** Modify `yunke-server/app/sync/scheduler.py` + `tests/test_scheduler_analytics_tick.py` + +改造 `_analytics_tick`（:703）为两段： + +```python + async def _analytics_tick(self) -> None: + """4h 分析链(staff 滚动3天 + 敏感词增量 + 管理域滚动)——异常只记日志不断链。 + + 两段独立事务:段1 staff+敏感词;段2 管理域(fact 近3天活跃会话 + 宽表/ + 通话 滚动3天,批次F 2026-09-10)。段2 失败只回滚自身,不牵连段1 已 + 提交结果(v2 Task12:指标失败不中断主分析链)。纯 DB 无浏览器,不抢 + ``_page_lock``;与夜间 full_sync 13/14 步重叠时的锁竞争语义同段1 + (非致命,幂等下轮自愈)。 + """ + # ── 段1:staff 滚动 + 敏感词增量(既有,原样保留) ── + try: + async with self._session_maker() as session: + from app.analytics.sensitive import scanner + from app.storage.staff_summary import recompute_days, rolling_days + from app.storage.summary import day_key + + today = day_key(int(time.time() * 1000)) + n = await recompute_days(session, rolling_days(today, 3)) + scan = await scanner.run_incremental(session) + await session.commit() + logger.info( + "[analytics-tick] staff_days=%s scanned=%s hits=%s", + n, scan.get("scanned"), scan.get("hits"), + ) + except Exception: # noqa: BLE001 — scheduler MUST NOT crash + logger.exception("analytics tick failed") + + # ── 段2:管理域滚动(fact 活跃会话 + 宽表/通话,当日+滚动3天) ── + try: + async with self._session_maker() as session: + from app.storage.employee_metrics import ( + recompute_call_days, + recompute_metric_days, + ) + from app.storage.first_reply_facts import ( + pairs_active_in_days, + recompute_pairs, + ) + from app.storage.staff_summary import rolling_days + from app.storage.summary import day_key + + rd = rolling_days(day_key(int(time.time() * 1000)), 3) + pairs = await pairs_active_in_days(session, rd) + f = await recompute_pairs(session, pairs) + w = await recompute_metric_days(session, rd) + c = await recompute_call_days(session, rd) + await session.commit() + logger.info( + "[analytics-tick] mgmt fact_pairs=%s wide_rows=%s call_rows=%s", + f, w, c, + ) + except Exception: # noqa: BLE001 + logger.exception("management metrics tick failed") +``` + +（导入合并/去重按文件现状微调；today 变量两段各自取或共享只读。注：`recompute_metric_days` 内部会再查一次 `pairs_active_in_days`（与本段显式调用重复但廉价）——非冗余待修，fact 段需要显式拿 pairs 计数记日志。） + +**测试追加**（照既有文件 mock 模式）： +1. tick 调用顺序：staff recompute_days → 敏感 scan → pairs_active_in_days(rolling3) → recompute_pairs → recompute_metric_days → recompute_call_days； +2. **段2 失败隔离**：recompute_metric_days 抛异常 → staff recompute_days 与敏感 scan 仍被调用且各自 commit 成功、tick 不抛、异常被记（caplog）； +3. rolling 窗口传参 = rolling_days(today,3)。 + +- [ ] `pytest tests/test_scheduler_analytics_tick.py -v` + 全量 -q → commit `feat(management): 4h分析链接入管理域滚动段——独立事务隔离/fact活跃会话+宽表+通话滚动3天` + +### Task F2: full_sync_all.sh 13/14 步 + 快照行数 + +**Files:** Modify `yunke-server/scripts/full_sync_all.sh` + +第 12 步之后（COMPLETE 横幅之前）加： + +```bash +# 13:首响事实表全量重建(管理域,跨日精确串;批次A)——会话发现读 +# msg_daily_stats(fact 行直算读 message),故须在 11 步后;与 11/12 +# 同为夜间自愈链(绕过钩子的写入由全量重算纠正) +echo "$(date) --- 13/14 First-reply-facts rebuild ---" +$PY -m app.sync.cli first-reply-facts +# 14:管理宽表+通话小表全量重建(驾驶舱/员工页读路径;串类列与 fact 同 +# 算法各自直算,放 fact 后保持派生链顺序语义) +echo "$(date) --- 14/14 Employee-metrics rebuild ---" +$PY -m app.sync.cli employee-metrics +``` + +终局快照 TABLES 追加三行（放 daily stats rows 后）： + +```python + ("first reply facts", "SELECT COUNT(*) FROM first_reply_fact"), + ("employee metric rows", "SELECT COUNT(*) FROM employee_metric_daily"), + ("employee call rows", "SELECT COUNT(*) FROM employee_call_daily"), +``` + +- [ ] `bash -n yunke-server/scripts/full_sync_all.sh` 语法过 + commit `feat(management): 夜间全量脚本13/14步——fact与宽表全量自愈+终局快照三表行数` + +### Task F3: 本地 scheduler 容器重建 + 真机验证 + 归档 + +1. **真机验证滚动段**（直接跑 tick 逻辑，不等 4h）：临时脚本或 `python -c` 用 scheduler 的 session_maker 调 `_analytics_tick()`（或手动等价：按 rolling 3 天调三个 recompute + commit）——记录耗时（预期 fact 活跃会话数十秒级内）+ 三表当日行数变化（recompute 后 updated 行 freshness）； +2. **幂等复跑**：连跑两次行数一致； +3. **本地 scheduler 容器重建**（新代码进夜间链）：`docker ps` 确认 scheduler 容器名 → `docker compose -f docker-compose.yml up -d --build scheduler`——**范围仅本地**（服务器 scheduler 镜像 2026-09-09 已重建过，服务器侧等部署时随 DEPLOY.md 流程走，本任务不碰）；重建前后 `docker exec python -c "import app.sync.scheduler"` 冒烟；（已知后果不阻塞：phone_call 尚未进 4h/每日页面同步链[审计P0-5,并行会话领地]——通话小表滚动重算在补齐前只是从静态源幂等重生成）； +4. 归档：耗时/行数/容器重建记录 → 核实文档「批次F-调度」小节 + commit。 + +- [ ] commit `docs(management): 批次F调度接入验证——滚动段耗时/幂等/容器重建(清9d3cc8e旧账)` + +## 验收清单 + +- [ ] tick 两段隔离测试过（段2 异常不牵连段1）； +- [ ] 夜间脚本 13/14 步 + 快照三表行数； +- [ ] 滚动段真机耗时记录（预期 <60s）+ 幂等； +- [ ] 本地 scheduler 容器已重建（新代码），夜间 02:00 链将含管理域； +- [ ] 全量测试绿； +- [ ] 未触碰审计 P0 领地（脚本加固/实体补抓——并行会话负责）。 + +## 后续（不在本子批） + +E：辅导+快照；F 余项：dashboard 性能修复（unreplied 索引）、winsorize 决策、服务器部署（届时服务器 scheduler 镜像重建走 DEPLOY.md）。