From 487f45b074c2a963d0f1eeb98c6e61a84887d98f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Mon, 7 Sep 2026 11:48:24 +0800 Subject: [PATCH] =?UTF-8?q?feat(analytics):=20=E7=9C=8B=E6=9D=BF=E7=AE=A1?= =?UTF-8?q?=E7=90=86=E5=8C=96=E8=A1=A5=E5=BC=BA=E2=80=94=E2=80=94=E5=91=98?= =?UTF-8?q?=E5=B7=A5=E6=B4=BB=E8=B7=83=E6=8E=92=E8=A1=8C=E8=A1=A8+?= =?UTF-8?q?=E9=A2=84=E8=AD=A6=E8=A1=8C=E5=8A=A8=E5=85=A5=E5=8F=A3=E5=8D=A1?= =?UTF-8?q?=E7=89=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 管理者反馈「纯汇总数字看不到有用信息」:overview 载荷扩 staff_ranking(全员行:msg_total rollup 优先/空表回退 msg_daily_stats 聚合,active_* 两路径统一走 msg_daily_stats 窗内 distinct peer—— stat_daily_staff 的 active_* 是当日快照,跨日 SUM/MAX 语义皆错; call_cnt 按 user_id 落窗直查;环比同期对齐)+ actions(该回未回 2h/ 真流失 30d/敏感词 pending,复用已测试查询路径各取 total)。 前端看板:toolbar 下三张可点行动卡(0 灰显仍可点,直达处置页, 随 channel 切换刷新)+ 趋势卡下员工排行表(max-height 420 滚动, 未映射小 tag,环比色同卡片惯例 None 显 '-')。 测试 574→577 全绿(rollup 聚合/回退/行动计数三新);build 过。 Co-Authored-By: Claude Code --- yunke-server/app/analytics/metrics.py | 186 ++++++++++++++++++ yunke-server/tests/test_analytics_overview.py | 140 +++++++++++++ yunkes/src/api/analytics.ts | 10 + yunkes/src/views/analytics/DashboardView.vue | 113 ++++++++++- 4 files changed, 448 insertions(+), 1 deletion(-) diff --git a/yunke-server/app/analytics/metrics.py b/yunke-server/app/analytics/metrics.py index 5b3caa1..0df5df0 100644 --- a/yunke-server/app/analytics/metrics.py +++ b/yunke-server/app/analytics/metrics.py @@ -19,6 +19,7 @@ from app.db.models import ( Message, MsgDailyStats, PhoneCall, + SensitiveHit, Staff, StatDailyStaff, ) @@ -264,6 +265,189 @@ async def anomaly(session: AsyncSession, channel: str) -> dict | None: return None +async def staff_ranking( + session: AsyncSession, *, channel: str, scope: str, +) -> list[dict]: + """员工活跃排行(A2 核心子集,2026-09-07 管理者反馈补强)。 + + 口径:msg_total=stat_daily_staff 聚合(空表回退 msg_daily_stats+映射, + 同 _active_staff 模式);call_cnt=phone_call 按 user_id 落窗直查(3k 行 + 小表);环比=与 _prev_scope_days 同期对比(None=前窗无数据);名字取 + staff.name,无 staff 行/未映射账号回退 user_id 原值标 unmapped。 + 排序 msg_total desc + user_id asc 决胜。返回全部行(员工 ~41)。 + + active_friends/active_rooms 两路径统一从 msg_daily_stats 算窗内 + COUNT(DISTINCT peer)(映射合并到员工后,同 recompute_days 模式): + stat_daily_staff 的 active_* 是**当日快照**,跨日 SUM 会重复计数、 + MAX 只是峰值,皆非「窗内活跃去重数」——rollup 只供 msg_total, + 避免两路径语义分叉。 + """ + days = scope_day_keys(scope, today_key()) + amap = await account_map(session, channel) + + # 活跃客户/群 + 回退 msg_total:落窗拉行,Python 侧映射后按 user 去重 + # (COUNT(DISTINCT peer) 必须在「账号合并到员工」之后——同人多账号 + # 跨账号去重,写 SQL 反而绕且踩 only_full_group_by) + rows = ( + await session.execute( + sa.select( + MsgDailyStats.owner_account_id, + MsgDailyStats.peer_kind, + MsgDailyStats.peer_id, + MsgDailyStats.send_cnt, + MsgDailyStats.recv_cnt, + ).where( + MsgDailyStats.channel == channel, MsgDailyStats.day.in_(days) + ) + ) + ).all() + act: dict[str, dict[str, set]] = {} + msg_fallback: dict[str, int] = {} + for owner, kind, peer, s, r in rows: + uid = amap.get(owner, owner) + a = act.setdefault(uid, {"friends": set(), "rooms": set()}) + if kind == "friend": # 显式分支:未来新 kind 跳过,不误入 rooms + a["friends"].add(peer) + elif kind == "room": + a["rooms"].add(peer) + msg_fallback[uid] = msg_fallback.get(uid, 0) + int(s or 0) + int(r or 0) + + use_rollup = await _staff_rows(session) + if use_rollup: + rollup = ( + await session.execute( + sa.select( + StatDailyStaff.user_id, + sa.func.coalesce( + sa.func.sum( + StatDailyStaff.msg_sent + StatDailyStaff.msg_recv + ), + 0, + ), + ) + .where( + StatDailyStaff.channel == channel, + StatDailyStaff.day.in_(days), + ) + .group_by(StatDailyStaff.user_id) + ) + ).all() + msg_total = {uid: int(v) for uid, v in rollup} + else: + msg_total = msg_fallback + + # 通话:按 user_id 落窗直查(窗口换算同 _call_count 写法) + calls: dict[str, int] = {} + if days: + lo, _ = day_bounds_ms(days[0]) + hi_start, _ = day_bounds_ms(days[-1]) + call_rows = ( + await session.execute( + sa.select(PhoneCall.user_id, sa.func.count()) + .where( + PhoneCall.user_id.is_not(None), + PhoneCall.start_call_time.is_not(None), + PhoneCall.start_call_time >= lo, + PhoneCall.start_call_time < hi_start + 86_400_000, + ) + .group_by(PhoneCall.user_id) + ) + ).all() + calls = {uid: int(v) for uid, v in call_rows} + + # 环比:同期对齐前窗(仅周/月;今日 vs 昨日全天会误导,与 overview 同口径) + prev_total: dict[str, int] = {} + if scope in ("week", "month"): + prev_days = _prev_scope_days(scope, days) + if use_rollup: + prev_rows = ( + await session.execute( + sa.select( + StatDailyStaff.user_id, + sa.func.coalesce( + sa.func.sum( + StatDailyStaff.msg_sent + StatDailyStaff.msg_recv + ), + 0, + ), + ) + .where( + StatDailyStaff.channel == channel, + StatDailyStaff.day.in_(prev_days), + ) + .group_by(StatDailyStaff.user_id) + ) + ).all() + prev_total = {uid: int(v) for uid, v in prev_rows} + else: + prev_rows = ( + await session.execute( + sa.select( + MsgDailyStats.owner_account_id, + sa.func.coalesce( + sa.func.sum( + MsgDailyStats.send_cnt + MsgDailyStats.recv_cnt + ), + 0, + ), + ) + .where( + MsgDailyStats.channel == channel, + MsgDailyStats.day.in_(prev_days), + ) + .group_by(MsgDailyStats.owner_account_id) + ) + ).all() + for owner, v in prev_rows: + uid = amap.get(owner, owner) # 同人多账号累加,不覆盖 + prev_total[uid] = prev_total.get(uid, 0) + int(v) + + # 名字/未映射:staff 几十行全拉;映射值集=在册员工 user_id 全集 + staff_names = { + uid: name + for uid, name in ( + await session.execute(sa.select(Staff.user_id, Staff.name)) + ).all() + } + mapped_uids = set(amap.values()) + + out = [] + for uid in sorted(set(msg_total) | set(act) | set(calls), + key=lambda u: (-msg_total.get(u, 0), u)): + cur, prev = msg_total.get(uid, 0), prev_total.get(uid, 0) + out.append({ + "user_id": uid, + "name": staff_names.get(uid) or uid, + "unmapped": uid not in staff_names and uid not in mapped_uids, + "msg_total": cur, + "active_friends": len(act[uid]["friends"]) if uid in act else 0, + "active_rooms": len(act[uid]["rooms"]) if uid in act else 0, + "call_cnt": calls.get(uid, 0), + "msg_delta": round((cur - prev) / prev, 3) if cur and prev else None, + }) + return out + + +async def action_counts(session: AsyncSession, *, channel: str) -> dict: + """预警行动入口计数(看板顶部卡片):复用已测试的查询路径,各取 total。""" + un = await unreplied(session, channel=channel, hours=2, page=1, page_size=1) + ch = await churn(session, channel=channel, kind="real", tier=30, min_msgs=5, + page=1, page_size=1) + pending = ( + await session.execute( + sa.select(sa.func.count()).select_from(SensitiveHit).where( + SensitiveHit.channel == channel, + SensitiveHit.disposition == "pending", + ) + ) + ).scalar_one() + return { + "unreplied_2h": int(un["total"]), + "churn_real_30d": int(ch["total"]), + "sensitive_pending": int(pending), + } + + async def overview( session: AsyncSession, *, channel: str, scope: str, trend_days: int = 30 ) -> dict[str, Any]: @@ -341,6 +525,8 @@ async def overview( "cards": cards, "trend": trend, "anomaly": await anomaly(session, channel), + "staff_ranking": await staff_ranking(session, channel=channel, scope=scope), + "actions": await action_counts(session, channel=channel), } diff --git a/yunke-server/tests/test_analytics_overview.py b/yunke-server/tests/test_analytics_overview.py index 5be3780..df724de 100644 --- a/yunke-server/tests/test_analytics_overview.py +++ b/yunke-server/tests/test_analytics_overview.py @@ -1,12 +1,19 @@ """A1 总览指标:卡片/首沟通口径/环比同期对齐(仅周月)/异常警示。""" from __future__ import annotations +import datetime as dt + import pytest from app.analytics import metrics from app.db.models import ( + Customer, + Message, MsgDailyStats, PhoneCall, + SensitiveHit, + SensitiveWord, + Staff, StatDailyStaff, WechatAccount, ) @@ -204,3 +211,136 @@ async def test_anomaly_new_staff_detail_and_sort(make_client, monkeypatch): assert "new" not in staff["u1"] # 老员工不带标记 await make_client(seed) + + +async def test_staff_ranking(make_client, monkeypatch): + """员工活跃排行:rollup 供 msg_total;active_* 统一走 msg_daily_stats 去重。 + + _TODAY=20260904 周五 → 本周 [0831..0904],上周 [0824..0828]。 + rollup 的 active_friends 故意 seed 999——若实现误用 rollup 聚合(跨日 + SUM/MAX 皆错)会立刻暴露;正确值=窗内 distinct peer(u1: f1+f2)。 + """ + monkeypatch.setattr(metrics, "today_key", lambda: _TODAY) + week = [20260831, 20260901, 20260902, 20260903, 20260904] + prev_week = [20260824, 20260825, 20260826, 20260827, 20260828] + async def seed(session): + session.add(WechatAccount(wx_id="wx1", user_id="u1", user_name="张三")) + session.add(WechatAccount(wx_id="wx2", user_id="u2", user_name="李四")) + session.add(Staff(user_id="u1", name="张三")) + session.add(Staff(user_id="u2", name="李四")) + # rollup:u1 本周 100(5×20),u2 50(5×10);active_* 999=诱饵(不得被采用) + for d in week: + session.add(StatDailyStaff(day=d, channel="wechat", user_id="u1", + msg_sent=20, msg_recv=0, + active_friends=999, active_rooms=999)) + session.add(StatDailyStaff(day=d, channel="wechat", user_id="u2", + msg_sent=10, msg_recv=0, + active_friends=1, active_rooms=0)) + # u3 仅存在于 rollup(无账号映射/无 staff 行)→ 未映射,名字回退 user_id + session.add(StatDailyStaff(day=20260904, channel="wechat", user_id="u3", + msg_sent=10, msg_recv=0, + active_friends=0, active_rooms=0)) + # 上周 rollup:u1 80(5×16) → msg_delta=(100-80)/80=0.25;u2 无前窗→None + for d in prev_week: + session.add(StatDailyStaff(day=d, channel="wechat", user_id="u1", + msg_sent=16, msg_recv=0, + active_friends=1, active_rooms=0)) + # 活跃指标真源:wx1→u1 两 friend(f1/f2 不同日)+一 room;wx2→u2 一 friend + for day, kind, peer, owner in [ + (20260831, "friend", "f1", "wx1"), (20260904, "friend", "f2", "wx1"), + (20260902, "room", "r1", "wx1"), (20260903, "friend", "f1", "wx2"), + ]: + session.add(MsgDailyStats(channel="wechat", owner_account_id=owner, + peer_kind=kind, peer_id=peer, day=day, + send_cnt=1, recv_cnt=0, + last_ts=1788500000000, last_is_send=True)) + # 通话:u1 一通落窗(1788500000000=2026-09-04 当日);u2 的在 2025 年不计 + session.add(PhoneCall(call_id="cr1", channel="voice_call", user_id="u1", + start_call_time=1788500000000, call_duration=60)) + session.add(PhoneCall(call_id="cr2", channel="voice_call", user_id="u2", + start_call_time=1756950000000, call_duration=60)) + await session.commit() + + out = await metrics.staff_ranking(session, channel="wechat", scope="week") + assert [r["user_id"] for r in out] == ["u1", "u2", "u3"] # msg_total 降序 + r1, r2, r3 = out + assert r1["name"] == "张三" and r1["unmapped"] is False + assert r1["msg_total"] == 100 # rollup 聚合 + assert r1["active_friends"] == 2 # distinct peer,非 rollup 的 999 + assert r1["active_rooms"] == 1 + assert r1["call_cnt"] == 1 + assert r1["msg_delta"] == 0.25 # (100-80)/80 + assert r2["msg_total"] == 50 and r2["active_friends"] == 1 + assert r2["active_rooms"] == 0 and r2["call_cnt"] == 0 # 窗外通话不计 + assert r2["msg_delta"] is None # 前窗无数据 + assert r3["name"] == "u3" and r3["unmapped"] is True + assert r3["msg_total"] == 10 and r3["active_friends"] == 0 + + await make_client(seed) + + +async def test_staff_ranking_fallback_when_rollup_empty(make_client, monkeypatch): + """stat_daily_staff 空表 → msg_total 回退 msg_daily_stats 聚合(部署即用)。""" + monkeypatch.setattr(metrics, "today_key", lambda: _TODAY) + async def seed(session): + session.add(WechatAccount(wx_id="wx1", user_id="u1", user_name="张三")) + 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=1788500000000, + last_is_send=True)) + # wx9 无账号行 → 映射不上按 owner 原值计入并标未映射 + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx9", + peer_kind="friend", peer_id="t2", day=20260904, + send_cnt=2, recv_cnt=0, last_ts=1788500000000, + last_is_send=True)) + await session.commit() + out = await metrics.staff_ranking(session, channel="wechat", scope="today") + assert [r["user_id"] for r in out] == ["u1", "wx9"] + assert out[0]["msg_total"] == 15 and out[0]["active_friends"] == 1 + assert out[0]["unmapped"] is False + assert out[0]["msg_delta"] is None # 今日 scope 无环比 + assert out[1]["msg_total"] == 2 and out[1]["unmapped"] is True + assert out[1]["name"] == "wx9" + + await make_client(seed) + + +async def test_action_counts(make_client, monkeypatch): + """预警行动入口:unreplied 走真实路径,churn/sensitive 各计 pending/真流失。""" + monkeypatch.setattr(metrics, "_now_ms", lambda: 1756950000000) + async def seed(session): + now = 1756950000000 + h3 = 3 * 3600 * 1000 + # 该回未回 1 条:客户尾向 3h 前 + 有效 text 消息 + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t1", day=20260904, + send_cnt=5, recv_cnt=3, last_ts=now - h3, + last_is_send=False)) + session.add(Message(msg_id="t1m", channel="wechat", owner_account_id="wx1", + talker_id="t1", timestamp=now - h3, is_send=False, + type="text", text="在吗,方案发我一下", + fetched_at=dt.datetime(2026, 9, 4, 12, 0))) + # 真流失 1 条:沉默 45 天 + 历史 ≥5 + day_ms = 86_400_000 + session.add(Customer(friend_id="c_real", channel="wechat", + owner_account_id="wx1", user_id="u1", + last_chat_time=now - 45 * day_ms, delete_flag=0)) + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="c_real", day=20260601, + send_cnt=6, recv_cnt=0, last_ts=now - 50 * day_ms, + last_is_send=True)) + # 敏感词:1 pending + 1 confirmed(只计 pending) + w = SensitiveWord(word="测试词") + session.add(w) + await session.flush() + session.add(SensitiveHit(word_id=w.id, msg_id="m1", channel="wechat", + disposition="pending")) + session.add(SensitiveHit(word_id=w.id, msg_id="m2", channel="wechat", + disposition="confirmed")) + await session.commit() + + out = await metrics.action_counts(session, channel="wechat") + assert out == {"unreplied_2h": 1, "churn_real_30d": 1, + "sensitive_pending": 1} + + await make_client(seed) diff --git a/yunkes/src/api/analytics.ts b/yunkes/src/api/analytics.ts index 1da6544..c87e5de 100644 --- a/yunkes/src/api/analytics.ts +++ b/yunkes/src/api/analytics.ts @@ -19,6 +19,14 @@ export interface OverviewCards { export interface TrendPoint { day: number; msg: number; active_customers: number; call_cnt: number } export interface AnomalyStaff { user_id: string; yesterday: number; prev7_mean: number; new?: boolean } export interface Anomaly { total_delta_pct: number; yesterday_total: number; prev7_mean: number; staff: AnomalyStaff[] } +// 员工活跃排行(2026-09-07 管理者反馈补强):active_*=窗内 distinct peer, +// msg_delta 仅周/月且当前/前窗均非 0 才有值 +export interface StaffRankingRow { + user_id: string; name: string; unmapped: boolean + msg_total: number; active_friends: number; active_rooms: number + call_cnt: number; msg_delta: number | null +} +export interface ActionCounts { unreplied_2h: number; churn_real_30d: number; sensitive_pending: number } export interface OverviewResp { scope: Scope channel: Channel @@ -26,6 +34,8 @@ export interface OverviewResp { cards: OverviewCards trend: TrendPoint[] anomaly: Anomaly | null + staff_ranking: StaffRankingRow[] + actions: ActionCounts } export interface HealthResp { data_as_of: number | null; entities: { table: string; last_seen: string | null }[] } export interface ChurnRow { diff --git a/yunkes/src/views/analytics/DashboardView.vue b/yunkes/src/views/analytics/DashboardView.vue index 36d2010..ed694d9 100644 --- a/yunkes/src/views/analytics/DashboardView.vue +++ b/yunkes/src/views/analytics/DashboardView.vue @@ -13,6 +13,20 @@ 数据截至 {{ asOfText }} + +
+ + {{ a.icon }} + {{ a.label }} + {{ a.count }} + +
+ @@ -49,19 +63,62 @@
+ + + + + + + + + + + + + + + + + + + + + + + + + +