docs(management): 批次F调度子批计划——tick管理段+夜间13/14步(Task F1-F3)
This commit is contained in:
浅念
@@ -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 <scheduler> 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）。
|
||||
Reference in New Issue
Block a user