diff --git a/yunke-server/app/management/employee_metrics.py b/yunke-server/app/management/employee_metrics.py new file mode 100644 index 0000000..d099d08 --- /dev/null +++ b/yunke-server/app/management/employee_metrics.py @@ -0,0 +1,349 @@ +"""管理域期聚合查询(批次B)——评分/异常/employees API 的单一数据源。 + +窗口:days=[start_day..end_day] 闭区间 yyyymmdd;上期=等长紧邻前窗; +个人基线=紧邻当前窗之前的 28 天。所有查询走 (day, ...) 索引,不扫消息表 +(除 distinct 活跃客户走 msg_daily_stats friend 行)。 + +口径要点: +- 分位/回复计数:first_reply_fact 窗内行拉回 Python 按员工合并排序取 + 最近邻秩 ceil(q·n)(整除算术 (n+1)//2 / (9n+9)//10——SQLite 无原生 + ceil,MySQL `/` 返回浮点,Python 侧取秩零方言风险)。合并按员工而非 + owner:一员工多账号(wx1+wx2→u1,channel=None 时人人如此)的样本必须 + 跨账号归并,逐 owner 取分位再映射会互相覆盖。fact 全库全期 ≤6.8万行, + Python 侧聚合可行。 +- 期末存量(new/cooling/dormant):状态快照取「每 (user, channel) 最新 + ≤end 行」——channel=None 时两渠道各取各的最新行再相加(直接取全渠道 + 最大 day 的那一行会丢另一渠道的存量)。 +- active_rooms_avg = SUM(active_rooms)/全窗天数(含无数据日摊薄), + 非有数据天数——「日均负载」语义。 +- days_with_data = COUNT(DISTINCT day):单渠道时与 COUNT(*) 等价 + (PK 含 day+channel,每日至多一行),channel=None 双渠道同日也只计 1 天。 +- 通话四列恒全渠道(employee_call_daily 无渠道维),channel 参数不过滤。 +""" +from __future__ import annotations + +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 + +from app.db.models import ( + EmployeeCallDaily, + EmployeeMetricDaily, + FirstReplyFact, + MsgDailyStats, + Staff, +) +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", + "reply_2h_count", "reply_24h_count", + "active_friends", # 期内 distinct(直查,非日值求和) + "active_rooms_avg", # 日均(全窗天数摊薄) + "new_customer", "cooling_customer", "dormant_customer", # 期末单日存量 + "reactivated_customer", # 期内求和 + "call_count", "call_seconds", "answered_call", "call_followup", +]) + + +def _to_date(k: int) -> dt.date: + return dt.date(k // 10000, k // 100 % 100, k % 100) + + +def _to_key(d: dt.date) -> int: + return d.year * 10000 + d.month * 100 + d.day + + +def day_range(start: int, end: int) -> list[int]: + """闭区间日键列表。""" + out, cur = [], _to_date(start) + while cur <= _to_date(end): + out.append(_to_key(cur)) + cur += dt.timedelta(days=1) + return out + + +def prev_window(start: int, end: int) -> tuple[int, int]: + """等长紧邻前窗。""" + n = len(day_range(start, end)) + pe = _to_date(start) - dt.timedelta(days=1) + ps = pe - dt.timedelta(days=n - 1) + return _to_key(ps), _to_key(pe) + + +def baseline_window(start: int) -> tuple[int, int]: + """个人基线:紧邻当前窗之前的 28 天。""" + be = _to_date(start) - dt.timedelta(days=1) + bs = be - dt.timedelta(days=27) + return _to_key(bs), _to_key(be) + + +def _percentile(vals: list[int], q_num: int, q_den: int) -> int | None: + """最近邻秩分位 ceil(q·n) 的样本值,q=q_num/q_den(1/2、9/10)。 + + 整数算术 rank=(q_num*n + q_den - 1)//q_den 对正整数 n 恒落在 [1,n] + (与 math.ceil(q·n) 等价,与批次A _percentile 同口径);空列表→None + (与 0 严格区分)。 + """ + if not vals: + return None + s = sorted(vals) + n = len(s) + rank = max(1, min(n, (q_num * n + q_den - 1) // q_den)) + return s[rank - 1] + + +# 累加器初值:全员(staff)都出结果行,无数据员工为零值而非缺行—— +# employees API 全员列表语义。 +_ZERO: dict[str, object] = { + "days_with_data": 0, "msg_sent": 0, "msg_recv": 0, "cust_threads": 0, + "threads_answered": 0, "fr_sum_ms": 0, + "fr_p50_ms": None, "fr_p90_ms": None, + "reply_2h_count": 0, "reply_24h_count": 0, + "active_friends": 0, "active_rooms_sum": 0, + "new_customer": 0, "cooling_customer": 0, "dormant_customer": 0, + "reactivated_customer": 0, + "call_count": 0, "call_seconds": 0, "answered_call": 0, "call_followup": 0, +} + + +async def period_metrics( + session: AsyncSession, + start_day: int, + end_day: int, + channel: str | None, + user_ids: Iterable[str] | None = None, +) -> dict[str, PeriodMetrics]: + """窗口 [start_day, end_day] × channel(None=全渠道)的员工期指标。 + + user_ids 传入时与 staff 表求交(幽灵过滤);返回含全部目标员工的 + PeriodMetrics(无数据员工为零值行)。channel=None 时两渠道 + account_map 合并(wx_id 与 WA uid 命名空间不撞,无冲突)。 + """ + staff_ids = { + u for u in (await session.execute(sa.select(Staff.user_id))).scalars() if u + } + target = staff_ids if user_ids is None else staff_ids & set(user_ids) + if not target: + return {} + + if channel is None: + amap: dict[str, str] = {} + for ch in ("wechat", "whatsapp"): + amap.update(await account_map(session, ch)) + else: + amap = await account_map(session, channel) + + n_days = len(day_range(start_day, end_day)) + acc: dict[str, dict] = {uid: dict(_ZERO) for uid in target} + + # ---- 宽表 SUM:消息计数/串类/日均负载/复活 ---- + wide_conds = [ + EmployeeMetricDaily.day.between(start_day, end_day), + EmployeeMetricDaily.metric_version == METRIC_VERSION, + EmployeeMetricDaily.user_id.in_(target), + ] + if channel is not None: + wide_conds.append(EmployeeMetricDaily.channel == channel) + wide = (await session.execute( + sa.select( + EmployeeMetricDaily.user_id.label("uid"), + sa.func.count(sa.distinct(EmployeeMetricDaily.day)).label("nd"), + sa.func.sum(EmployeeMetricDaily.msg_sent).label("sent"), + sa.func.sum(EmployeeMetricDaily.msg_recv).label("recv"), + sa.func.sum(EmployeeMetricDaily.cust_threads).label("thr"), + sa.func.sum(EmployeeMetricDaily.threads_answered).label("ans"), + sa.func.sum(EmployeeMetricDaily.fr_sum_ms).label("frs"), + sa.func.sum(EmployeeMetricDaily.active_rooms).label("rooms"), + sa.func.sum( + EmployeeMetricDaily.reactivated_customer_count + ).label("react"), + ).where(*wide_conds).group_by(EmployeeMetricDaily.user_id) + )).all() + for uid, nd, sent, recv, thr, ans, frs, rooms, react in wide: + a = acc[uid] + a["days_with_data"] = int(nd or 0) + a["msg_sent"] = int(sent or 0) + a["msg_recv"] = int(recv or 0) + a["cust_threads"] = int(thr or 0) + a["threads_answered"] = int(ans or 0) + a["fr_sum_ms"] = int(frs or 0) + a["active_rooms_sum"] = int(rooms or 0) + a["reactivated_customer"] = int(react or 0) + + # ---- 期末存量:每 (user, channel) 最新 ≤end 行,跨渠道相加 ---- + stock_conds = [ + EmployeeMetricDaily.day <= end_day, + EmployeeMetricDaily.metric_version == METRIC_VERSION, + EmployeeMetricDaily.user_id.in_(target), + ] + if channel is not None: + stock_conds.append(EmployeeMetricDaily.channel == channel) + latest = (await session.execute( + sa.select( + EmployeeMetricDaily.user_id.label("uid"), + EmployeeMetricDaily.channel.label("ch"), + sa.func.max(EmployeeMetricDaily.day).label("d"), + ).where(*stock_conds).group_by( + EmployeeMetricDaily.user_id, EmployeeMetricDaily.channel + ) + )).all() + if latest: + # (user, channel, day) 含 metric_version 唯一(主键)——OR 点查 + # 精确取回各渠道最新行,再按 user 跨渠道 SUM + stocks = (await session.execute( + sa.select( + EmployeeMetricDaily.user_id.label("uid"), + sa.func.sum(EmployeeMetricDaily.new_customer_count).label("nw"), + sa.func.sum( + EmployeeMetricDaily.cooling_customer_count + ).label("cool"), + sa.func.sum( + EmployeeMetricDaily.dormant_customer_count + ).label("dorm"), + ).where(sa.or_(*( + sa.and_( + EmployeeMetricDaily.user_id == uid, + EmployeeMetricDaily.channel == ch, + EmployeeMetricDaily.day == d, + EmployeeMetricDaily.metric_version == METRIC_VERSION, + ) + for uid, ch, d in latest + ))).group_by(EmployeeMetricDaily.user_id) + )).all() + for uid, nw, cool, dorm in stocks: + a = acc[uid] + a["new_customer"] = int(nw or 0) + a["cooling_customer"] = int(cool or 0) + a["dormant_customer"] = int(dorm or 0) + + # ---- fact 分位 + 2h/24h 回复计数(同一过滤) ---- + fr_conds = [FirstReplyFact.day.between(start_day, end_day)] + if channel is not None: + fr_conds.append(FirstReplyFact.channel == channel) + fr_rows = (await session.execute( + sa.select( + FirstReplyFact.owner_account_id.label("owner"), + FirstReplyFact.fr_ms.label("fr"), + ).where(*fr_conds) + )).all() + fr_by_user: dict[str, list[int]] = {} + for owner, fr in fr_rows: + uid = amap.get(owner or "") + if uid is not None and uid in target: + fr_by_user.setdefault(uid, []).append(int(fr)) + 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) + + 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 行) ---- + af_conds = [ + MsgDailyStats.day.between(start_day, end_day), + MsgDailyStats.peer_kind == "friend", + ] + if channel is not None: + af_conds.append(MsgDailyStats.channel == channel) + pairs = (await session.execute( + sa.select(MsgDailyStats.owner_account_id, MsgDailyStats.peer_id) + .distinct().where(*af_conds) + )).all() + active: dict[str, set[tuple[str, 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 "")) + for uid, pset in active.items(): + acc[uid]["active_friends"] = len(pset) + + # ---- 通话:恒全渠道(表无渠道维,channel 参数不过滤) ---- + calls = (await session.execute( + sa.select( + EmployeeCallDaily.user_id.label("uid"), + sa.func.sum(EmployeeCallDaily.call_count).label("cc"), + sa.func.sum(EmployeeCallDaily.call_seconds).label("cs"), + sa.func.sum(EmployeeCallDaily.answered_call_count).label("ca"), + sa.func.sum(EmployeeCallDaily.call_followup_count).label("cf"), + ).where( + EmployeeCallDaily.day.between(start_day, end_day), + EmployeeCallDaily.user_id.in_(target), + ).group_by(EmployeeCallDaily.user_id) + )).all() + for uid, cc, cs, ca, cf in calls: + a = acc[uid] + a["call_count"] = int(cc or 0) + a["call_seconds"] = int(cs or 0) + a["answered_call"] = int(ca or 0) + a["call_followup"] = int(cf or 0) + + return { + uid: PeriodMetrics( + user_id=uid, + days_with_data=a["days_with_data"], + msg_sent=a["msg_sent"], + msg_recv=a["msg_recv"], + cust_threads=a["cust_threads"], + threads_answered=a["threads_answered"], + fr_sum_ms=a["fr_sum_ms"], + fr_p50_ms=a["fr_p50_ms"], + fr_p90_ms=a["fr_p90_ms"], + reply_2h_count=a["reply_2h_count"], + reply_24h_count=a["reply_24h_count"], + active_friends=a["active_friends"], + active_rooms_avg=(a["active_rooms_sum"] / n_days) if n_days else 0.0, + new_customer=a["new_customer"], + cooling_customer=a["cooling_customer"], + dormant_customer=a["dormant_customer"], + reactivated_customer=a["reactivated_customer"], + call_count=a["call_count"], + call_seconds=a["call_seconds"], + answered_call=a["answered_call"], + call_followup=a["call_followup"], + ) + for uid, a in acc.items() + } + + +def team_medians(metrics: dict[str, PeriodMetrics]) -> dict[str, float | None]: + """每数值字段的全员中位数(纯函数,评分归一用)。 + + None(无样本的分位等)剔除后取中位——statistics.median:奇数取中位 + 样本、偶数取中间两数均值;全员 None → None(调用方按无基线处理)。 + """ + out: dict[str, float | None] = {} + for field in PeriodMetrics._fields: + if field == "user_id": + continue + vals = [ + v for v in (getattr(m, field) for m in metrics.values()) + if v is not None + ] + out[field] = statistics.median(vals) if vals else None + return out diff --git a/yunke-server/tests/test_management_period_metrics.py b/yunke-server/tests/test_management_period_metrics.py new file mode 100644 index 0000000..381b179 --- /dev/null +++ b/yunke-server/tests/test_management_period_metrics.py @@ -0,0 +1,276 @@ +"""管理域期聚合查询判别测试(批次B Task B1)。 + +判据(v2 §3 / B1 规格): +- 分位=fact 表最近邻秩 ceil(q·n) 精确值(整除算术,无 ceil 方言坑), + 多账号/跨渠道样本按员工合并; +- 宽表 SUM 按 (day∈窗, metric_version='v1', channel) 过滤; + active_rooms_avg 摊到全窗天数(含无数据日); +- 期末生命周期存量=每 (user, channel) 最新 ≤end 行,跨渠道相加; +- distinct 活跃客户=窗内 (owner, peer) 去重(peer_kind='friend'); +- 窗口工具:闭区间 day_range/等长前窗/28 天个人基线窗; +- 幽灵员工(不在 staff)过滤;user_ids 交集;通话恒全渠道。 +""" +import datetime as dt +from zoneinfo import ZoneInfo + +from app.db.models import ( + EmployeeCallDaily, + EmployeeMetricDaily, + FirstReplyFact, + MsgDailyStats, + Staff, + WechatAccount, + WhatsappAccount, +) +from app.management.employee_metrics import ( + PeriodMetrics, + baseline_window, + day_range, + period_metrics, + prev_window, + team_medians, +) + +_SH = ZoneInfo("Asia/Shanghai") + + +def _ts(*args) -> int: + return int(dt.datetime(*args, tzinfo=_SH).timestamp() * 1000) + + +async def _seed_org(s, users=("u1", "u2")): + for u in users: + s.add(Staff(user_id=u, name=u, is_on_job=True)) + s.add(WechatAccount(wx_id="wx1", user_id="u1")) + s.add(WechatAccount(wx_id="wx2", user_id="u2")) + await s.commit() + + +def _mday(day, *, channel="wechat", user_id="u1", metric_version="v1", **kw): + cols = { + "msg_sent": 0, "msg_recv": 0, "active_friends": 0, "active_rooms": 0, + "cust_threads": 0, "threads_answered": 0, "fr_sum_ms": 0, + "new_customer_count": 0, "cooling_customer_count": 0, + "dormant_customer_count": 0, "reactivated_customer_count": 0, + "data_complete": True, + } + cols.update(kw) + return EmployeeMetricDaily( + day=day, channel=channel, user_id=user_id, + metric_version=metric_version, **cols + ) + + +def _fact(fr_ms, *, channel="wechat", owner="wx1", peer="p1", day=20260901, i=0): + end = _ts(2026, 9, 1, 10, 0) + i * 60_000 + return FirstReplyFact( + channel=channel, owner_account_id=owner, peer_id=peer, + thread_end_ts=end, reply_ts=end + fr_ms, + reply_msg_id=f"{owner}-{peer}-m{i}", day=day, fr_ms=fr_ms, + ) + + +async def test_percentile_exact(make_session): + """5 样本 p50=180s/p90=300s(ceil(2.5)=3、ceil(4.5)=5,1-based); + 2 样本 p50=10s/p90=20s;窗外行不参与。""" + s = make_session + await _seed_org(s) + for i, sec in enumerate([60, 120, 180, 240, 300]): + s.add(_fact(sec * 1000, i=i)) + for i, sec in enumerate([10, 20]): + s.add(_fact(sec * 1000, owner="wx2", peer="q1", i=i)) + # 窗外(0906>end)长尾:混入则 p90=999s + s.add(_fact(999_000, i=9, day=20260906)) + await s.commit() + m = await period_metrics(s, 20260901, 20260905, "wechat") + assert m["u1"].fr_p50_ms == 180_000 + assert m["u1"].fr_p90_ms == 300_000 + assert m["u2"].fr_p50_ms == 10_000 + assert m["u2"].fr_p90_ms == 20_000 + + +async def test_period_sums_vs_daily_rows(make_session): + """宽表两天行 SUM 正确;days_with_data=2;active_rooms_avg 摊到全窗 + 5 天(含无数据日);channel/metric_version/窗外行都不计。""" + s = make_session + await _seed_org(s) + s.add(_mday(20260901, msg_sent=3, msg_recv=5, cust_threads=2, + threads_answered=1, fr_sum_ms=7_000, active_rooms=4, + reactivated_customer_count=1)) + s.add(_mday(20260903, msg_sent=4, msg_recv=6, cust_threads=3, + threads_answered=2, fr_sum_ms=11_000, active_rooms=6, + reactivated_customer_count=2)) + # 反例三连:错渠道 / 错版本 / 窗外 + s.add(_mday(20260901, channel="whatsapp", msg_sent=100)) + s.add(_mday(20260901, metric_version="v2", msg_sent=200)) + s.add(_mday(20260906, msg_sent=400)) + await s.commit() + m = await period_metrics(s, 20260901, 20260905, "wechat") + u1 = m["u1"] + assert u1.msg_sent == 7 + assert u1.msg_recv == 11 + assert u1.cust_threads == 5 + assert u1.threads_answered == 3 + assert u1.fr_sum_ms == 18_000 + assert u1.days_with_data == 2 + assert u1.active_rooms_avg == 2.0 # (4+6)/5 全窗天数,非 2 有数据日 + assert u1.reactivated_customer == 3 + + +async def test_end_of_period_stock(make_session): + """期末存量=每渠道最新 ≤end 行:窗 [0901,0903] 取 0901 行、 + [0901,0905] 取 0905 行;channel=None 跨渠道相加(4+2=6)。""" + s = make_session + await _seed_org(s) + s.add(_mday(20260901, new_customer_count=1, cooling_customer_count=2, + dormant_customer_count=3)) + s.add(_mday(20260905, dormant_customer_count=4)) + s.add(_mday(20260905, channel="whatsapp", dormant_customer_count=2)) + await s.commit() + + m1 = await period_metrics(s, 20260901, 20260903, "wechat") + assert m1["u1"].new_customer == 1 + assert m1["u1"].cooling_customer == 2 + assert m1["u1"].dormant_customer == 3 + + m2 = await period_metrics(s, 20260901, 20260905, "wechat") + assert m2["u1"].dormant_customer == 4 # wechat 渠道内最新行 + assert m2["u1"].new_customer == 0 + + m3 = await period_metrics(s, 20260901, 20260905, None) + assert m3["u1"].dormant_customer == 6 # wechat 4 + whatsapp 2 + + +async def test_distinct_active_cross_day(make_session): + """同 peer 两天两行只计 1;room 行与幽灵 owner 不计。""" + s = make_session + await _seed_org(s) + for day, peer in [(20260901, "p1"), (20260902, "p1"), (20260902, "p2")]: + s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id=peer, day=day)) + # room 行不计入活跃客户 + s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="room", peer_id="r1", day=20260902)) + # 幽灵 owner(account_map 无映射)不计 + s.add(MsgDailyStats(channel="wechat", owner_account_id="wx9", + peer_kind="friend", peer_id="p9", day=20260902)) + await s.commit() + m = await period_metrics(s, 20260901, 20260902, "wechat") + assert m["u1"].active_friends == 2 + + +def test_window_helpers(): + # day_range 闭区间 + 月末边界 + assert day_range(20260901, 20260903) == [20260901, 20260902, 20260903] + assert day_range(20260831, 20260901) == [20260831, 20260901] + # prev_window 等长紧邻前窗(7 天窗 → 7 天窗;单日 → 单日) + assert prev_window(20260908, 20260914) == (20260901, 20260907) + assert prev_window(20260901, 20260901) == (20260831, 20260831) + # baseline_window:紧邻窗前 28 天([0915] → [0818, 0914]) + assert baseline_window(20260915) == (20260818, 20260914) + + +async def test_ghost_and_channel_none(make_session): + """幽灵(不在 staff)宽表行/fact 行不计;channel=None 合并两渠道 + amap——u1(wx1+wa1)的宽表 SUM 与 fact 分位都跨渠道归并; + 无数据员工(u2)也以零值行出现。""" + s = make_session + await _seed_org(s) + s.add(WhatsappAccount(uid="wa1", user_id="u1")) + # 账号存在但映射到非 staff 用户 → 幽灵 + s.add(WechatAccount(wx_id="wxg", user_id="ghost")) + await s.commit() + + s.add(_mday(20260901, msg_sent=3)) + s.add(_mday(20260901, channel="whatsapp", msg_sent=4)) + s.add(_mday(20260901, user_id="ghost", msg_sent=50)) + s.add(_fact(100_000, owner="wx1", i=0)) + s.add(_fact(300_000, channel="whatsapp", owner="wa1", peer="w1", i=1)) + s.add(_fact(50_000, owner="wxg", i=2)) + await s.commit() + + m = await period_metrics(s, 20260901, 20260905, None) + assert set(m) == {"u1", "u2"} # ghost 不在;零数据 u2 在 + u1 = m["u1"] + assert u1.msg_sent == 7 # 两渠道行都归 u1 + 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 + assert u1.reply_24h_count == 2 + assert m["u2"].msg_sent == 0 + assert m["u2"].fr_p50_ms is None + + +async def test_reply_counts(make_session): + """fr=90min/3h/25h → 2h_count=1、24h_count=2(2h 是 24h 的子集)。""" + s = make_session + await _seed_org(s) + s.add(_fact(5_400_000, i=0)) # 90min ≤2h + s.add(_fact(10_800_000, i=1)) # 3h ∈(2h,24h] + s.add(_fact(90_000_000, i=2)) # 25h >24h + await s.commit() + m = await period_metrics(s, 20260901, 20260905, "wechat") + assert m["u1"].reply_2h_count == 1 + assert m["u1"].reply_24h_count == 2 + assert m["u2"].reply_2h_count == 0 + + +async def test_user_ids_filter_and_calls(make_session): + """user_ids 与 staff 求交(未知 id 剔除/空集→{});通话四列按窗 SUM + 且不受 channel 参数影响(无渠道维,恒全渠道)。""" + s = make_session + await _seed_org(s) + s.add(_mday(20260901, user_id="u1", msg_sent=1)) + s.add(_mday(20260901, user_id="u2", msg_sent=2)) + s.add(EmployeeCallDaily(day=20260901, user_id="u1", call_count=3, + call_seconds=600, answered_call_count=2, + call_followup_count=1)) + s.add(EmployeeCallDaily(day=20260902, user_id="u1", call_count=1, + call_seconds=60, answered_call_count=1, + call_followup_count=0)) + s.add(EmployeeCallDaily(day=20260901, user_id="u2", call_count=9)) + await s.commit() + + m = await period_metrics(s, 20260901, 20260905, "wechat", user_ids=["u1", "nobody"]) + assert set(m) == {"u1"} + assert m["u1"].msg_sent == 1 + u1 = m["u1"] + assert u1.call_count == 4 + assert u1.call_seconds == 660 + assert u1.answered_call == 3 + assert u1.call_followup == 1 + + empty = await period_metrics(s, 20260901, 20260905, "wechat", user_ids=[]) + assert empty == {} + + # channel="whatsapp" 时通话仍计入(恒全渠道) + mwa = await period_metrics(s, 20260901, 20260905, "whatsapp") + assert mwa["u1"].call_count == 4 + + +def test_team_medians(): + """奇数样本取中位、偶数取均值;None 剔除;全 None → None;user_id 不参与。""" + + def _pm(**kw): + base = dict(zip(PeriodMetrics._fields, [None] * len(PeriodMetrics._fields))) + base.update(user_id="x", days_with_data=0, msg_sent=0, msg_recv=0, + cust_threads=0, threads_answered=0, fr_sum_ms=0, + reply_2h_count=0, reply_24h_count=0, active_friends=0, + active_rooms_avg=0.0, new_customer=0, cooling_customer=0, + dormant_customer=0, reactivated_customer=0, + call_count=0, call_seconds=0, answered_call=0, + call_followup=0) + base.update(kw) + return PeriodMetrics(**base) + + metrics = { + "u1": _pm(user_id="u1", msg_sent=10, fr_p50_ms=None), + "u2": _pm(user_id="u2", msg_sent=20, fr_p50_ms=100), + "u3": _pm(user_id="u3", msg_sent=30, fr_p50_ms=200), + } + med = team_medians(metrics) + assert med["msg_sent"] == 20 + assert med["fr_p50_ms"] == 150.0 # None 剔除后 [100,200] 均值 + assert "user_id" not in med + assert team_medians({})["msg_sent"] is None + assert team_medians({"u1": _pm()})["fr_p50_ms"] is None