diff --git a/docs/superpowers/plans/2026-09-04-yunke-analytics-module.md b/docs/superpowers/plans/2026-09-04-yunke-analytics-module.md index 2531f36..98193d9 100644 --- a/docs/superpowers/plans/2026-09-04-yunke-analytics-module.md +++ b/docs/superpowers/plans/2026-09-04-yunke-analytics-module.md @@ -740,6 +740,8 @@ async def recompute_days( ) -> int: """重算给定 day×channel 的员工行:删旧插新,幂等。 + 返回**有数据行的 day×channel 组合数**(空渠道不计——与测试断言的 + 「天数」语义一致,seed 只有 wechat 日行时返回 1)。 单日日行仅数百~两千,拉回 Python 侧按 user_id 聚合——COUNT(DISTINCT peer) 必须在「账号合并到员工」之后算(同人多账号 distinct peer 跨账号 去重),写 SQL 反而绕且踩 only_full_group_by。 @@ -761,6 +763,13 @@ async def recompute_days( ) ) ).all() + await session.execute( + sa.delete(StatDailyStaff).where( + StatDailyStaff.day == day, StatDailyStaff.channel == channel + ) + ) + if not rows: + continue agg: dict[str, dict] = {} for owner, kind, peer, s, r in rows: uid = amap.get(owner, owner) # 映射不上按 owner 原值(读侧标注) @@ -770,11 +779,6 @@ async def recompute_days( a["sent"] += int(s or 0) a["recv"] += int(r or 0) (a["friends"] if kind == "friend" else a["rooms"]).add(peer) - await session.execute( - sa.delete(StatDailyStaff).where( - StatDailyStaff.day == day, StatDailyStats.channel == channel - ) - ) for uid, a in sorted(agg.items()): session.add( StatDailyStaff( @@ -799,7 +803,7 @@ async def recompute_all(session: AsyncSession) -> int: return await recompute_days(session, list(days)) ``` -⚠️ 上面 `recompute_days` 里 delete 语句有一个笔误要修正：`StatDailyStats.channel == channel` 应为 `StatDailyStaff.channel == channel`。写代码时注意。 +⚠️ 上面 `recompute_days` 的 delete 已修正为 `StatDailyStaff.channel == channel`；返回值语义 = **有数据行的 day×channel 组合数**（空渠道不计，测试断言 n==1 成立：seed 只有 wechat 日行）。 - [ ] **Step 5: 跑测试** Run: `cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_staff_summary.py -x -q` → Expected: 4 passed @@ -837,28 +841,34 @@ _TODAY = 20260904 # 测试统一用固定 day 键,monkeypatch today_key async def _seed(session): session.add(WechatAccount(wx_id="wx1", user_id="u1", user_name="张三")) - # 昨日(20260903):u1 100 条;前 7 日均值(20260827-0902):每天 200 → 跌 50% 触发 + # 昨日(20260903):u1 90 条;前 7 日均值(20260827-0902):每天 200 → 偏离 -55%>50% 触发 + # (阈值是严格 >0.5——正好 -50% 不触发,seed 用 90) for day, sent in [(20260827, 200), (20260828, 200), (20260829, 200), (20260830, 200), (20260831, 200), (20260901, 200), - (20260902, 200), (20260903, 100)]: + (20260902, 200), (20260903, 90)]: session.add(StatDailyStaff(day=day, channel="wechat", user_id="u1", msg_sent=sent, msg_recv=0, active_friends=5, active_rooms=1)) + # 当日 rollup 行(active_staff 卡片读 stat_daily_staff,必须有当日行) + session.add(StatDailyStaff(day=20260904, channel="wechat", user_id="u1", + msg_sent=15, msg_recv=7, + active_friends=1, active_rooms=1)) # 当日 + 首沟通:t1 当日首现(首沟通=1),t2 早在 20260827 就有(不算) session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", peer_kind="friend", peer_id="t1", day=20260904, - send_cnt=10, recv_cnt=5, last_ts=1756950000000, + send_cnt=10, recv_cnt=5, last_ts=1788500000000, last_is_send=True)) session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", peer_kind="friend", peer_id="t2", day=20260827, - send_cnt=3, recv_cnt=0, last_ts=1756240000000, + send_cnt=3, recv_cnt=0, last_ts=1787800000000, last_is_send=True)) session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", peer_kind="room", peer_id="r1", day=20260904, - send_cnt=7, recv_cnt=0, last_ts=1756950000000, + send_cnt=7, recv_cnt=0, last_ts=1788500000000, last_is_send=True)) + # 通话时间戳必须是 2026-09-04 当日(1756950000000 是 2025 年——坑) session.add(PhoneCall(call_id="c1", channel="voice_call", - start_call_time=1756950000000, call_duration=60)) + start_call_time=1788500000000, call_duration=60)) await session.commit() @@ -883,9 +893,9 @@ async def test_anomaly_yesterday_vs_prev7(make_client, monkeypatch): await _seed(session) a = metrics.anomaly(session, channel="wechat") assert a is not None - assert a["total_delta_pct"] <= -0.5 # 100 vs 均值 200 = -50% + assert a["total_delta_pct"] <= -0.5 # 90 vs 均值 200 = -55%(严格 >50% 触发) staff = {s["user_id"]: s for s in a["staff"]} - assert staff["u1"]["yesterday"] == 100 + assert staff["u1"]["yesterday"] == 90 assert staff["u1"]["prev7_mean"] == 200.0 await make_client(seed) @@ -1308,7 +1318,7 @@ import datetime as dt from app.analytics import metrics from app.db.models import Customer, MsgDailyStats, Staff -_NOW = 1756950000000 # 2026-09-04 ~ 上海 +_NOW = 1756950000000 # 相对时间基准(绝对日期无关紧要,断言全部相对此值自洽) def _cust(friend_id, *, last_chat=None, user_id="u1", delete=0, block=None): @@ -1885,7 +1895,11 @@ async def delete_word(session: AsyncSession, word_id: int) -> None: async def preview_count( session: AsyncSession, word: str, exclude_words: list[str] | None = None ) -> int: - """加词前 COUNT——MySQL 走 ngram MATCH,其他方言 LIKE(测试)。""" + """加词前 COUNT——MySQL 走 ngram MATCH,其他方言 LIKE(测试)。 + + 口径注:MySQL MATCH 路径不扣排除词(预览=上限,排除词在扫描时精确 + 生效);LIKE 路径(测试库)精确扣。前端文案「预计命中」体现上限语义。 + """ word = (word or "").strip() if not word: return 0 @@ -1914,6 +1928,36 @@ def _hit_exclude(text: str | None, word: str, exclude_words: list[str] | None) - if not exclude_words or not text: return False return any(ex in text for ex in exclude_words) + + +# 内置起步词库(spec §3.6):幂等补种(CLI sensitive-scan 首跑时调),上线后 +# 管理员在词库页自补行业词。 +STARTER_WORDS: list[tuple[str, str]] = [ + ("私单", "飞单"), + ("飞单", "飞单"), + ("加私人微信", "私联"), + ("红包", "资金"), + ("转账", "资金"), + ("离职", "人员风险"), +] + + +async def ensure_starter_words(session: AsyncSession) -> int: + """缺则补(按 word 去重幂等),返回新种入数。""" + existing = set( + ( + await session.execute(sa.select(SensitiveWord.word)) + ).scalars().all() + ) + n = 0 + for word, category in STARTER_WORDS: + if word in existing: + continue + session.add(SensitiveWord(word=word, category=category)) + n += 1 + if n: + await session.flush() + return n ``` - [ ] **Step 4: 端点**（analytics.py 追加；`HTTPException` 已引入） @@ -2211,7 +2255,7 @@ async def run_incremental(session: AsyncSession) -> dict: # 子串判定即可——MATCH 只在全量回填用,增量统一 LIKE 语义 # 精确)。此处直接子串: hit_ids = {r[0] for r in rows if w.word in (texts[r[0]] or "")} - sel_rows = [r for r in rows if r[0] in hit_ids] + sel_rows = [tuple(r[:7]) for r in rows if r[0] in hit_ids] # 切掉第8列 fetched_at existing = set( ( await session.execute( @@ -2526,15 +2570,16 @@ async def _cmd_sensitive_scan(args: argparse.Namespace) -> int: settings = get_settings() engine, session_maker = await _init_db(settings.db_url) try: - from app.analytics.sensitive import scanner + from app.analytics.sensitive import scanner, words as words_svc async with session_maker() as session: + seeded = await words_svc.ensure_starter_words(session) # 幂等补种起步词库 if args.full: out = await scanner.run_full(session) else: out = await scanner.run_incremental(session) await session.commit() - print(f"[sensitive-scan] {out}") + print(f"[sensitive-scan] seeded={seeded} {out}") return 0 finally: await engine.dispose() @@ -2973,7 +3018,7 @@ onMounted(load) **Files:** - Create(填实): `yunkes/src/views/analytics/ChurnView.vue` -- [ ] **Step 1: 实现页面**（三分法 tab + tier 筛选 + 表格 + CSV 导出 + 行点击跳聊天窗；CSV 用项目现成 `csvExport.ts` 全量翻页惯例） +- [ ] **Step 1: 实现页面**（三分法 tab + tier 筛选 + 表格 + CSV 导出 + 行点击跳聊天窗；CSV 用项目现成 `csvExport.ts` 的 **`exportListToCsv({ fetchPage, columns, filename })`**（`yunkes/src/utils/csvExport.ts:45`，传 fetchPage 即可，不用手写翻页循环）；toolbar 右侧加「数据截至 HH:mm」角标——`getAnalyticsHealth()` 一次，三个列表页同款（spec §5 全板块带角标）。CSV 平铺带归属员工列= v1 简化（spec 的「按员工分组」由表格列承担，注释说明）） ```vue