xinyouhe/yunke
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(analytics): 管理层指标呈现——首响回复率上墙/客户资产总览/自定义日期/排行五新列

Browse Source 
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-09-07 14:41:08 +08:00
parent feb3b98665
commit 085a9b4fd3
6 changed files with 606 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yunke-server/app/analytics/metrics.py
+216 -30
View File
@@ -43,23 +43,43 @@ def _day_date(day: int) -> dt.date:
return dt.date(day // 10000, (day // 100) % 100, day % 100)
def scope_day_keys(scope: str, today: int) -> list[int]:
def scope_day_keys(
scope: str, today: int, start: int | None = None, end: int | None = None
) -> list[int]:
"""today={今};week=本周一..今天;month=本月1号..今天;last30=近30天
滚动窗(含端点)。last30 让排行/卡片命中任意有数据的月份(镜像停更期
间日历月窗可能全空)。"""
end = _day_date(today)
间日历月窗可能全空)。
custom=显式 [start..end] yyyymmdd(C1):start≤end 且跨度≤366 天
(router 层已 422 校验,这里防御性同检——metrics 直调也拿到 ValueError
而非空窗/怪窗)。
"""
if scope == "custom":
if start is None or end is None:
raise ValueError("custom scope requires start and end")
s, e = _day_date(start), _day_date(end) # 非法日历值此处抛 ValueError
if s > e:
raise ValueError("custom scope requires start <= end")
if (e - s).days > 365: # 含端点共 366 天
raise ValueError("custom span must be <= 366 days")
out, cur = [], s
while cur <= e:
out.append(_date_key(cur))
cur += dt.timedelta(days=1)
return out
end_d = _day_date(today)
if scope == "today":
start = end
start_d = end_d
elif scope == "week":
start = end - dt.timedelta(days=end.weekday()) # 周一=0
start_d = end_d - dt.timedelta(days=end_d.weekday()) # 周一=0
elif scope == "month":
start = end.replace(day=1)
start_d = end_d.replace(day=1)
elif scope == "last30":
start = end - dt.timedelta(days=29)
start_d = end_d - dt.timedelta(days=29)
else:
raise ValueError(f"bad scope: {scope!r}")
out, cur = [], start
while cur <= end:
out, cur = [], start_d
while cur <= end_d:
out.append(_date_key(cur))
cur += dt.timedelta(days=1)
return out
@@ -69,7 +89,8 @@ def _prev_scope_days(scope: str, days: list[int]) -> list[int]:
"""同期对齐环比窗(与 days 等长逐日对齐;今日 scope 不调用)。
week=各日-7天(本周一..今 vs 上周一..上周同日);month=上月同日
(短月钳到上月末,如 3/31→2/28);last30=各日-30天(前一个滚动窗)。
(短月钳到上月末,如 3/31→2/28);last30=各日-30天(前一个滚动窗);
custom=各日-N 天(等长紧邻前窗,例外不做同期对齐)。
同期对齐而非滚动紧邻窗——后者周会跨周末、月取上月尾巴,管理者会
误读成「上周/上月」(spec §8)。
"""
@@ -88,6 +109,11 @@ def _prev_scope_days(scope: str, days: list[int]) -> list[int]:
return out
if scope == "last30":
return [_date_key(_day_date(d) - dt.timedelta(days=30)) for d in days]
if scope == "custom":
# C1 例外:等长紧邻前窗(各日 -N 天滚动)。custom 不做同期对齐——
# 任意起点无「周/月」节律可对,紧邻窗是最不易误读的对照。
n = len(days)
return [_date_key(_day_date(d) - dt.timedelta(days=n)) for d in days]
raise ValueError(f"bad scope: {scope!r}")
@@ -297,8 +323,9 @@ async def anomaly(session: AsyncSession, channel: str | None) -> dict | None:
async def staff_ranking(
session: AsyncSession, *, channel: str | None, scope: str,
start: int | None = None, end: int | None = None,
) -> list[dict]:
"""员工活跃排行(A2 核心子集,2026-09-07 管理者反馈补强)。
"""员工活跃排行(A2 核心子集,2026-09-07 管理者反馈补强;批2 五新列)。
口径:msg_total=stat_daily_staff 聚合(空表回退 msg_daily_stats+映射,
同 _active_staff 模式);call_cnt=phone_call 按 user_id 落窗直查(3k 行
@@ -306,6 +333,19 @@ async def staff_ranking(
staff.name,无 staff 行/未映射账号回退 user_id 原值标 unmapped。
排序 msg_total desc + user_id asc 决胜。返回全部行(员工 ~41)。
批2 新列(A3/A4/A5):
- fr_avg_ms=Σfr_sum_ms/Σthreads_answered(answered=0→None);
reply_rate=Σanswered/Σthreads(threads=0→None)——两路径同源
(rollup 三列 / 回退 msg_daily_stats 三列累加,含 room 行,与批1
管道口径一致);
- first_contact_cnt=该员工当期首沟通客户数:(owner,peer) 全史
MIN(day) 落窗的 distinct peer(映射到 user 后跨账号去重);
- call_seconds/answer_rate=phone_call 落窗 SUM(call_duration)(秒)
与 call_duration>0 占比(count=0→None);
- deleted/blocked_friends=customer 表按 user_id 归属的**全库**存量
delete_flag=1 / is_block 计数(与 churn handover 同归属口径;
不随窗——删粉是存量状态)。
channel=None(全平台):查询去渠道条件+映射合并,同 user 跨渠道量
自然合到一行(员工排行本就是 user 级)。
@@ -315,12 +355,12 @@ async def staff_ranking(
MAX 只是峰值,皆非「窗内活跃去重数」——rollup 只供 msg_total,
避免两路径语义分叉。
"""
days = scope_day_keys(scope, today_key())
days = scope_day_keys(scope, today_key(), start, end)
amap = await _account_map(session, channel)
# 活跃客户/群 + 回退 msg_total:落窗拉行,Python 侧映射后按 user 去重
# (COUNT(DISTINCT peer) 必须在「账号合并到员工」之后——同人多账号
# 跨账号去重,写 SQL 反而绕且踩 only_full_group_by)
# 活跃客户/群 + 回退 msg_total/首响三列:落窗拉行,Python 侧映射后按
# user 去重(COUNT(DISTINCT peer) 必须在「账号合并到员工」之后——同人
# 多账号跨账号去重,写 SQL 反而绕且踩 only_full_group_by)
conds = [MsgDailyStats.day.in_(days)]
if channel is not None:
conds.append(MsgDailyStats.channel == channel)
@@ -332,12 +372,16 @@ async def staff_ranking(
MsgDailyStats.peer_id,
MsgDailyStats.send_cnt,
MsgDailyStats.recv_cnt,
MsgDailyStats.cust_threads,
MsgDailyStats.threads_answered,
MsgDailyStats.fr_sum_ms,
).where(*conds)
)
).all()
act: dict[str, dict[str, set]] = {}
msg_fallback: dict[str, int] = {}
for owner, kind, peer, s, r in rows:
th_fallback: dict[str, list[int]] = {}
for owner, kind, peer, s, r, t, a_ns, fr in rows:
uid = amap.get(owner, owner)
a = act.setdefault(uid, {"friends": set(), "rooms": set()})
if kind == "friend": # 显式分支:未来新 kind 跳过,不误入 rooms
@@ -345,8 +389,13 @@ async def staff_ranking(
elif kind == "room":
a["rooms"].add(peer)
msg_fallback[uid] = msg_fallback.get(uid, 0) + int(s or 0) + int(r or 0)
tf = th_fallback.setdefault(uid, [0, 0, 0])
tf[0] += int(t or 0)
tf[1] += int(a_ns or 0)
tf[2] += int(fr or 0)
use_rollup = await _staff_rows(session)
th: dict[str, tuple[int, int, int]] = {}
if use_rollup:
conds = [StatDailyStaff.day.in_(days)]
if channel is not None:
@@ -361,23 +410,45 @@ async def staff_ranking(
),
0,
),
sa.func.coalesce(sa.func.sum(StatDailyStaff.cust_threads), 0),
sa.func.coalesce(
sa.func.sum(StatDailyStaff.threads_answered), 0
),
sa.func.coalesce(sa.func.sum(StatDailyStaff.fr_sum_ms), 0),
)
.where(*conds)
.group_by(StatDailyStaff.user_id)
)
).all()
msg_total = {uid: int(v) for uid, v in rollup}
msg_total = {}
for uid, mv, t, a_ns, fr in rollup:
msg_total[uid] = int(mv)
th[uid] = (int(t), int(a_ns), int(fr))
else:
msg_total = msg_fallback
th = {uid: (v[0], v[1], v[2]) for uid, v in th_fallback.items()}
# 通话:按 user_id 落窗直查(窗口换算同 _call_count 写法)
# 通话:按 user_id 落窗直查(窗口换算同 _call_count 写法)——量/时长/
# 接通率一次取回(call_duration>0 计为接通)
calls: dict[str, int] = {}
call_secs: dict[str, int] = {}
call_ans: 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())
sa.select(
PhoneCall.user_id,
sa.func.count(),
sa.func.coalesce(sa.func.sum(PhoneCall.call_duration), 0),
sa.func.coalesce(
sa.func.sum(
sa.case((PhoneCall.call_duration > 0, 1), else_=0)
),
0,
),
)
.where(
PhoneCall.user_id.is_not(None),
PhoneCall.start_call_time.is_not(None),
@@ -387,11 +458,61 @@ async def staff_ranking(
.group_by(PhoneCall.user_id)
)
).all()
calls = {uid: int(v) for uid, v in call_rows}
for uid, cnt, secs, ans in call_rows:
calls[uid] = int(cnt)
call_secs[uid] = int(secs)
call_ans[uid] = int(ans)
# 首沟通归属:peer 全史 MIN(day) 落窗 → 记到 (channel,)owner 映射后的
# user;distinct 在 user 级(同人多账号/客户换跟进人不重复计)
fc_conds = [MsgDailyStats.peer_kind == "friend"]
if channel is not None:
fc_conds.append(MsgDailyStats.channel == channel)
fc_rows = (
await session.execute(
sa.select(
MsgDailyStats.owner_account_id,
MsgDailyStats.peer_id,
sa.func.min(MsgDailyStats.day),
)
.where(*fc_conds)
.group_by(MsgDailyStats.owner_account_id, MsgDailyStats.peer_id)
)
).all()
dayset = set(days)
first_contacts: dict[str, set[str]] = {}
for owner, peer, first in fc_rows:
if first in dayset:
first_contacts.setdefault(amap.get(owner, owner), set()).add(peer)
# 删粉/拉黑:customer 全库存量按 user_id 归属(NULL user_id 跳过——
# 排行是员工视图,无归属客户进 overview 的 customer_assets)
del_conds = [Customer.user_id.is_not(None)]
if channel is not None:
del_conds.append(Customer.channel == channel)
del_rows = (
await session.execute(
sa.select(
Customer.user_id,
sa.func.coalesce(
sa.func.sum(sa.case((Customer.delete_flag == 1, 1), else_=0)),
0,
),
sa.func.coalesce(
sa.func.sum(sa.case((Customer.is_block.is_(True), 1), else_=0)),
0,
),
)
.where(*del_conds)
.group_by(Customer.user_id)
)
).all()
deleted = {uid: int(d) for uid, d, _b in del_rows}
blocked = {uid: int(b) for uid, _d, b in del_rows}
# 环比:同期对齐前窗(仅周/月;今日 vs 昨日全天会误导,与 overview 同口径)
prev_total: dict[str, int] = {}
if scope in ("week", "month", "last30"):
if scope in ("week", "month", "last30", "custom"):
prev_days = _prev_scope_days(scope, days)
if use_rollup:
conds = [StatDailyStaff.day.in_(prev_days)]
@@ -449,6 +570,8 @@ async def staff_ranking(
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)
t, a_ns, fr = th.get(uid, (0, 0, 0))
cnt = calls.get(uid, 0)
out.append({
"user_id": uid,
"name": staff_names.get(uid) or uid,
@@ -456,8 +579,15 @@ async def staff_ranking(
"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),
"call_cnt": cnt,
"msg_delta": round((cur - prev) / prev, 3) if cur and prev else None,
"fr_avg_ms": round(fr / a_ns) if a_ns else None,
"reply_rate": round(a_ns / t, 4) if t else None,
"first_contact_cnt": len(first_contacts.get(uid, ())),
"call_seconds": call_secs.get(uid, 0),
"answer_rate": round(call_ans[uid] / cnt, 4) if cnt else None,
"deleted_friends": deleted.get(uid, 0),
"blocked_friends": blocked.get(uid, 0),
})
return out
@@ -485,11 +615,62 @@ async def action_counts(session: AsyncSession, *, channel: str | None) -> dict:
}
async def _customer_assets(session: AsyncSession) -> dict:
"""B1 客户资产总览:全库存量快照,不随 scope/channel(分层/删粉是存量
状态,与筛选无关)。
剔除 delete_flag=1 后按 last_chat_time 分层(边界与 churn real 同式:
>now-Nd 为近 N 天):active30=≤30 天;silent30_90=(30,90] 天前;
sleeping90p=≥90 天前;从未激活(last_chat_time NULL,云客没聊过/新加)
单列 never。渠道计数与已删好友单列。customer ~12k 行单条聚合直查。
"""
day_ms = 86_400_000
now = _now_ms()
not_deleted = sa.or_(Customer.delete_flag.is_(None), Customer.delete_flag == 0)
lt = Customer.last_chat_time
def _n(cond: Any) -> int:
return sa.func.coalesce(sa.func.sum(sa.case((cond, 1), else_=0)), 0)
row = (
await session.execute(
sa.select(
_n(sa.and_(not_deleted, lt.is_(None))),
_n(sa.and_(not_deleted, lt.is_not(None), lt > now - 30 * day_ms)),
_n(sa.and_(not_deleted, lt.is_not(None),
lt <= now - 30 * day_ms, lt > now - 90 * day_ms)),
_n(sa.and_(not_deleted, lt.is_not(None), lt <= now - 90 * day_ms)),
_n(sa.and_(not_deleted, Customer.channel == "wechat")),
_n(sa.and_(not_deleted, Customer.channel == "whatsapp")),
_n(Customer.delete_flag == 1),
)
)
).one()
never, active30, silent, sleeping, wechat, whatsapp, deleted = (
int(v) for v in row
)
return {
"total": never + active30 + silent + sleeping,
"active30": active30,
"silent30_90": silent,
"sleeping90p": sleeping,
"never": never,
"wechat": wechat,
"whatsapp": whatsapp,
"deleted": deleted,
}
async def overview(
session: AsyncSession, *, channel: str | None, scope: str, trend_days: int = 30
session: AsyncSession, *, channel: str | None, scope: str, trend_days: int = 30,
start: int | None = None, end: int | None = None,
) -> dict[str, Any]:
"""A1:卡片 + 环比(仅周/月,同期对齐) + 近 N 天趋势 + 异常警示。
批2:C1 自定义日期(scope=custom + start/end yyyymmdd,环比=等长紧邻
前窗滚动,不做同期对齐——见 _prev_scope_days);B1 customer_assets
客户资产总览(全库存量,不随 scope/channel)。
channel=None(全平台,看板 v2 默认):所有指标跨渠道合并——msg/活跃客户
直接合计(peer_id 跨渠道不碰撞);活跃员工按 user 去重(跨渠道重复计数
不拆);通话量本就是 call-type 无渠道语义。此时追加:
@@ -502,7 +683,7 @@ async def overview(
if trend_days < 1 or trend_days > 90:
raise ValueError("trend_days must be 1..90")
today = today_key()
days = scope_day_keys(scope, today)
days = scope_day_keys(scope, today, start, end)
rows = await _day_rows(session, channel, days)
msg_total = sum(int(s or 0) + int(r or 0) for _d, _k, _p, s, r in rows)
active_customers = len({p for _d, kind, p, _s, _r in rows if kind == "friend"})
@@ -517,8 +698,9 @@ async def overview(
"first_contact": await _first_contact(session, channel, days),
}
# 环比仅周/月,同期对齐(今日 vs 昨日全天会误导,spec §8)
if scope in ("week", "month", "last30"):
# 环比仅周/月,同期对齐(今日 vs 昨日全天会误导,spec §8);
# custom=等长紧邻前窗滚动(例外,见 _prev_scope_days)
if scope in ("week", "month", "last30", "custom"):
prev_days = _prev_scope_days(scope, days)
prev_rows = await _day_rows(session, channel, prev_days)
prev_total = sum(int(s or 0) + int(r or 0) for _d, _k, _p, s, r in prev_rows)
@@ -528,10 +710,11 @@ async def overview(
)
# 趋势:近 trend_days 天 每日 消息量/活跃客户/通话量
end = _day_date(today)
# (t_end 不复用 end 参数名——那是 custom scope 的 yyyymmdd 入参)
t_end = _day_date(today)
t_days = []
cur = end - dt.timedelta(days=trend_days - 1)
while cur <= end:
cur = t_end - dt.timedelta(days=trend_days - 1)
while cur <= t_end:
t_days.append(_date_key(cur))
cur += dt.timedelta(days=1)
# 通话趋势:phone_call 拉窗内 (start_call_time) Python 侧按日分桶(3k 行)
@@ -571,9 +754,12 @@ async def overview(
"channel": "all" if channel is None else channel,
"days": days,
"cards": cards,
"customer_assets": await _customer_assets(session),
"trend": trend,
"anomaly": await anomaly(session, channel),
"staff_ranking": await staff_ranking(session, channel=channel, scope=scope),
"staff_ranking": await staff_ranking(
session, channel=channel, scope=scope, start=start, end=end
),
"actions": await action_counts(session, channel=channel),
}
if channel is None:
yunke-server/app/api/routers/analytics.py
+27 -2
View File
@@ -1,6 +1,7 @@
"""GET /api/analytics/* —— 数据分析板块端组(前缀分组=未来角色权限落点)。"""
from __future__ import annotations
import datetime as dt
from typing import Annotated, Any, Literal
import sqlalchemy as sa
@@ -21,7 +22,11 @@ ChannelParam = Annotated[Literal["wechat", "whatsapp"], Query()]
# 看板 v2(2026-09-07):overview 默认全平台——"all" 在 router 层翻译成
# metrics 的 channel=None,不把哨兵字符串传进 SQL
AllChannelParam = Annotated[Literal["all", "wechat", "whatsapp"], Query()]
ScopeParam = Annotated[Literal["today", "week", "month", "last30"], Query()]
# C1(批2):custom=自定义日期范围,配合 start/end yyyymmdd
ScopeParam = Annotated[
Literal["today", "week", "month", "last30", "custom"], Query()
]
YmdParam = Annotated[int, Query(ge=20000101, le=29991231)]
@router.get("/health")
@@ -36,6 +41,8 @@ async def analytics_overview(
session: SessionDep,
channel: AllChannelParam = "all",
scope: ScopeParam = "today",
start: YmdParam | None = None,
end: YmdParam | None = None,
) -> dict[str, Any]:
"""A1 经营总览:卡片+趋势+异常警示一次返回(单往返)。
@@ -43,9 +50,27 @@ async def analytics_overview(
双渠道趋势序列;单渠道行为不变。
cards 不含 data_as_of——数据截至走 /health(data_as_of 为 ms-epoch int
与 entities[].last_seen 为 ISO 字符串的双格式)。
scope=custom(C1):start/end=yyyymmdd(缺一/日历非法/Start>End/跨度
>366 天皆 422);环比=等长紧邻前窗滚动(不做同期对齐)。
"""
s = e = None
if scope == "custom":
if start is None or end is None:
raise HTTPException(422, "scope=custom requires start and end")
try:
sd = dt.date(start // 10000, (start // 100) % 100, start % 100)
ed = dt.date(end // 10000, (end // 100) % 100, end % 100)
except ValueError as exc: # 如 20260230——Query 只拦范围不拦日历
raise HTTPException(422, "invalid start/end calendar date") from exc
if sd > ed:
raise HTTPException(422, "start must be <= end")
if (ed - sd).days > 365: # 含端点共 366 天
raise HTTPException(422, "custom span must be <= 366 days")
s, e = start, end
return await metrics_svc.overview(
session, channel=None if channel == "all" else channel, scope=scope
session, channel=None if channel == "all" else channel, scope=scope,
start=s, end=e,
)