feat(analytics): 看板管理化补强——员工活跃排行表+预警行动入口卡片

管理者反馈「纯汇总数字看不到有用信息」: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 <noreply@anthropic.com>
This commit is contained in:
浅念
2026-09-07 11:48:24 +08:00
parent 8499a74a72
commit 487f45b074
4 changed files with 448 additions and 1 deletions
+186
View File
@@ -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),
}