diff --git a/yunke-server/app/management/employee_metrics.py b/yunke-server/app/management/employee_metrics.py index d099d08..01dabf0 100644 --- a/yunke-server/app/management/employee_metrics.py +++ b/yunke-server/app/management/employee_metrics.py @@ -26,7 +26,6 @@ import datetime as dt import statistics from collections import namedtuple from collections.abc import Iterable -from zoneinfo import ZoneInfo import sqlalchemy as sa from sqlalchemy.ext.asyncio import AsyncSession @@ -38,11 +37,9 @@ from app.db.models import ( MsgDailyStats, Staff, ) +from app.storage.employee_metrics import METRIC_VERSION from app.storage.staff_summary import account_map -_SH = ZoneInfo("Asia/Shanghai") -METRIC_VERSION = "v1" - PeriodMetrics = namedtuple("PeriodMetrics", [ "user_id", "days_with_data", "msg_sent", "msg_recv", "cust_threads", "threads_answered", "fr_sum_ms", "fr_p50_ms", "fr_p90_ms", @@ -226,7 +223,7 @@ async def period_metrics( a["cooling_customer"] = int(cool or 0) a["dormant_customer"] = int(dorm or 0) - # ---- fact 分位 + 2h/24h 回复计数(同一过滤) ---- + # ---- fact 分位 + 2h/24h 回复计数(同一过滤,Python 侧一次聚合) ---- fr_conds = [FirstReplyFact.day.between(start_day, end_day)] if channel is not None: fr_conds.append(FirstReplyFact.channel == channel) @@ -244,26 +241,11 @@ async def period_metrics( for uid, frs in fr_by_user.items(): acc[uid]["fr_p50_ms"] = _percentile(frs, 1, 2) acc[uid]["fr_p90_ms"] = _percentile(frs, 9, 10) + # 2h/24h 计数与分位同源同窗——回复率分子分母口径一致 + acc[uid]["reply_2h_count"] = sum(1 for fr in frs if fr <= 7_200_000) + acc[uid]["reply_24h_count"] = sum(1 for fr in frs if fr <= 86_400_000) - reply = (await session.execute( - sa.select( - FirstReplyFact.owner_account_id.label("owner"), - sa.func.sum(sa.case( - (FirstReplyFact.fr_ms <= 7_200_000, 1), else_=0 - )).label("c2h"), - sa.func.sum(sa.case( - (FirstReplyFact.fr_ms <= 86_400_000, 1), else_=0 - )).label("c24h"), - ).where(*fr_conds).group_by(FirstReplyFact.owner_account_id) - )).all() - for owner, c2h, c24h in reply: - uid = amap.get(owner or "") - if uid is None or uid not in target: - continue - acc[uid]["reply_2h_count"] += int(c2h or 0) - acc[uid]["reply_24h_count"] += int(c24h or 0) - - # ---- distinct 活跃客户:窗内 (owner, peer) 去重(friend 行) ---- + # ---- distinct 活跃客户:账号归并到员工后按 peer 去重 ---- af_conds = [ MsgDailyStats.day.between(start_day, end_day), MsgDailyStats.peer_kind == "friend", @@ -274,12 +256,14 @@ async def period_metrics( sa.select(MsgDailyStats.owner_account_id, MsgDailyStats.peer_id) .distinct().where(*af_conds) )).all() - active: dict[str, set[tuple[str, str]]] = {} + active: dict[str, set[str]] = {} for owner, peer in pairs: uid = amap.get(owner or "") if uid is None or uid not in target: continue - active.setdefault(uid, set()).add((owner or "", peer or "")) + # (uid, peer) 粒度:同一员工多账号(wx1+wx2)的共享客户只计 1 + # ——与宽表日值口径一致(storage 侧账号归并后 add(peer)) + active.setdefault(uid, set()).add(peer or "") for uid, pset in active.items(): acc[uid]["active_friends"] = len(pset) diff --git a/yunke-server/tests/test_management_period_metrics.py b/yunke-server/tests/test_management_period_metrics.py index 381b179..d66e107 100644 --- a/yunke-server/tests/test_management_period_metrics.py +++ b/yunke-server/tests/test_management_period_metrics.py @@ -6,7 +6,8 @@ - 宽表 SUM 按 (day∈窗, metric_version='v1', channel) 过滤; active_rooms_avg 摊到全窗天数(含无数据日); - 期末生命周期存量=每 (user, channel) 最新 ≤end 行,跨渠道相加; -- distinct 活跃客户=窗内 (owner, peer) 去重(peer_kind='friend'); +- distinct 活跃客户=窗内 (uid, peer) 去重(peer_kind='friend')—— + 同一员工多账号的共享客户只计 1(与宽表日值口径一致); - 窗口工具:闭区间 day_range/等长前窗/28 天个人基线窗; - 幽灵员工(不在 staff)过滤;user_ids 交集;通话恒全渠道。 """ @@ -159,6 +160,22 @@ async def test_distinct_active_cross_day(make_session): assert m["u1"].active_friends == 2 +async def test_active_friends_shared_peer_across_accounts(make_session): + """多账号共享客户去重:u1 的 wx1/wx3 都会话 p1 → active_friends=1 + (账号归并到员工后按 peer 去重,非 (owner, peer) 双计)。""" + s = make_session + await _seed_org(s) + s.add(WechatAccount(wx_id="wx3", user_id="u1")) + await s.commit() + for owner in ("wx1", "wx3"): + s.add(MsgDailyStats(channel="wechat", owner_account_id=owner, + peer_kind="friend", peer_id="p1", day=20260901)) + await s.commit() + m = await period_metrics(s, 20260901, 20260905, "wechat") + assert m["u1"].active_friends == 1 + assert m["u2"].active_friends == 0 + + def test_window_helpers(): # day_range 闭区间 + 月末边界 assert day_range(20260901, 20260903) == [20260901, 20260902, 20260903] @@ -193,6 +210,7 @@ async def test_ghost_and_channel_none(make_session): assert set(m) == {"u1", "u2"} # ghost 不在;零数据 u2 在 u1 = m["u1"] assert u1.msg_sent == 7 # 两渠道行都归 u1 + assert u1.days_with_data == 1 # 双渠道同日只计 1 天(DISTINCT day) assert u1.fr_p50_ms == 100_000 # [100s,300s] 合并后 ceil(0.5*2)=1 assert u1.fr_p90_ms == 300_000 # ceil(0.9*2)=2 assert u1.reply_2h_count == 2