diff --git a/yunke-server/app/storage/employee_metrics.py b/yunke-server/app/storage/employee_metrics.py index 0e66c23..d0c9a19 100644 --- a/yunke-server/app/storage/employee_metrics.py +++ b/yunke-server/app/storage/employee_metrics.py @@ -12,12 +12,28 @@ from __future__ import annotations import logging +import math +from bisect import bisect_right from collections.abc import Iterable import sqlalchemy as sa from sqlalchemy.ext.asyncio import AsyncSession -from app.db.models import EmployeeCallDaily, Message, PhoneCall, Staff +from app.analytics.lifecycle import classify +from app.db.models import ( + Customer, + EmployeeCallDaily, + EmployeeMetricDaily, + Message, + MsgDailyStats, + PhoneCall, + Staff, +) +from app.storage.first_reply_facts import ( + compute_conversation, + load_conversation, + pairs_active_in_days, +) from app.storage.staff_summary import account_map from app.storage.summary import day_bounds_ms @@ -131,3 +147,202 @@ async def recompute_call_days(session: AsyncSession, days: Iterable[int]) -> int ) written += 1 return written + + +METRIC_VERSION = "v1" + + +def _percentile(vals: list[int], q: float) -> int | None: + """最近邻秩分位 ceil(q·n);空列表→None(与 0 严格区分)。""" + if not vals: + return None + s = sorted(vals) + idx = min(len(s) - 1, max(0, math.ceil(q * len(s)) - 1)) + return s[idx] + + +async def _lifecycle_daily( + session: AsyncSession, + days: list[int], + channel: str, + amap: dict[str, str], + valid_users: set[str], +) -> dict[int, dict[str, dict[str, int]]]: + """逐日 {day: {user: {state: count}}}——customer 当前归属 + 互动日 sweep。 + + 数据源:msg_daily_stats friend 行全历史(互动日/条数)。16k 客户 × 窗口 + 天数 bisect,夜间全量约秒级;4h tick 只算当日+3 天。 + """ + rows = ( + await session.execute( + sa.select( + MsgDailyStats.peer_id, MsgDailyStats.day, + MsgDailyStats.send_cnt, MsgDailyStats.recv_cnt, + ).where( + MsgDailyStats.channel == channel, + MsgDailyStats.peer_kind == "friend", + ) + ) + ).all() + days_map: dict[str, set[int]] = {} + cnt: dict[str, int] = {} + for peer, day, snd, rcv in rows: + days_map.setdefault(peer, set()).add(day) + cnt[peer] = cnt.get(peer, 0) + int(snd or 0) + int(rcv or 0) + interaction = {p: sorted(ds) for p, ds in days_map.items()} + + custs = ( + await session.execute( + sa.select(Customer.owner_account_id, Customer.friend_id).where( + Customer.channel == channel, + sa.or_(Customer.delete_flag.is_(None), Customer.delete_flag == 0), + Customer.owner_account_id.is_not(None), + ) + ) + ).all() + peer_to_user: dict[str, str] = {} + for owner, peer in custs: + uid = amap.get(owner or "") + if uid and uid in valid_users and peer: + peer_to_user[str(peer)] = uid + + out: dict[int, dict[str, dict[str, int]]] = {d: {} for d in days} + for peer, days_list in interaction.items(): + uid = peer_to_user.get(peer) + if uid is None: + continue + first = days_list[0] + hist = cnt.get(peer, 0) + for d in days: + idx = bisect_right(days_list, d) - 1 + if idx < 0: + continue + last = days_list[idx] + prev = days_list[idx - 1] if idx > 0 else None + state = classify(first, last, prev, d, hist) + if state: + out[d].setdefault(uid, {}).setdefault(state, 0) + out[d][uid][state] += 1 + return out + + +async def recompute_metric_days( + session: AsyncSession, + days: Iterable[int], + channels: tuple[str, ...] = ("wechat", "whatsapp"), +) -> int: + """重算给定日的宽表:按 (day, channel) delete+insert,幂等。返回写入行数。 + + 串类列与首响日值:扫窗口内活跃 friend 会话的**全历史**(每会话一次), + 与 first_reply_fact 同一 compute_conversation——分子分母同源(v2 §1.1 #5)。 + """ + days = sorted(set(days)) + valid_users = await known_user_ids(session) + pairs = await pairs_active_in_days(session, days) + pairs_by_channel: dict[str, list] = {} + for p in pairs: + pairs_by_channel.setdefault(p[0], []).append((p[1], p[2])) + + written = 0 + for channel in channels: + amap = await account_map(session, channel) + # ---- 串类+分位:逐会话全历史一次,聚合到 (day, user) ---- + thread_cols: dict[tuple[int, str], dict] = {} + frs: dict[tuple[int, str], list[int]] = {} + day_set = set(days) + for owner, peer in pairs_by_channel.get(channel, []): + uid = amap.get(owner) + if uid is None or uid not in valid_users: + continue + rows = await load_conversation(session, channel, owner, peer) + facts, agg = compute_conversation(rows) + for d, a in agg.items(): + if d not in day_set: + continue + key = (d, uid) + t = thread_cols.setdefault( + key, {"threads": 0, "answered": 0, "fr": 0} + ) + t["threads"] += a.threads + t["answered"] += a.answered + t["fr"] += a.fr_sum_ms + frs.setdefault(key, []).extend( + f.fr_ms for f in facts if f.day == d + ) + + # ---- 生命周期 ---- + life = await _lifecycle_daily(session, days, channel, amap, valid_users) + + # ---- 消息计数+负载:msg_daily_stats(算法无关列) ---- + for day in days: + mds = ( + await session.execute( + sa.select( + MsgDailyStats.owner_account_id, + MsgDailyStats.peer_kind, + MsgDailyStats.peer_id, + MsgDailyStats.send_cnt, + MsgDailyStats.recv_cnt, + ).where( + MsgDailyStats.day == day, MsgDailyStats.channel == channel + ) + ) + ).all() + await session.execute( + sa.delete(EmployeeMetricDaily).where( + EmployeeMetricDaily.day == day, + EmployeeMetricDaily.channel == channel, + EmployeeMetricDaily.metric_version == METRIC_VERSION, + ) + ) + agg: dict[str, dict] = {} + for owner, kind, peer, snd, rcv in mds: + uid = amap.get(owner) + if uid is None or uid not in valid_users: + continue # 幽灵过滤 + a = agg.setdefault( + uid, + {"sent": 0, "recv": 0, "friends": set(), "rooms": set()}, + ) + # 消息计数列 = 1:1 口径(v2 §1.1 #2):room 只进 active_rooms + if kind == "friend": + a["sent"] += int(snd or 0) + a["recv"] += int(rcv or 0) + a["friends"].add(peer) + elif kind == "room": + a["rooms"].add(peer) + day_life = life.get(day, {}) + # 行并集:消息源 ∪ 生命周期源——纯生命周期日(员工当日无消息 + # 但持有降温/沉睡客户)也要出宽表行,否则客户经营维度漏天 + all_uids = sorted(set(agg) | set(day_life)) + for uid in all_uids: + a = agg.get( + uid, {"sent": 0, "recv": 0, "friends": set(), "rooms": set()} + ) + key = (day, uid) + t = thread_cols.get(key, {"threads": 0, "answered": 0, "fr": 0}) + lv = day_life.get(uid, {}) + session.add( + EmployeeMetricDaily( + day=day, + channel=channel, + user_id=uid, + metric_version=METRIC_VERSION, + msg_sent=a["sent"], + msg_recv=a["recv"], + active_friends=len(a["friends"]), + active_rooms=len(a["rooms"]), + cust_threads=t["threads"], + threads_answered=t["answered"], + fr_sum_ms=t["fr"], + first_reply_p50_ms=_percentile(frs.get(key, []), 0.5), + first_reply_p90_ms=_percentile(frs.get(key, []), 0.9), + new_customer_count=lv.get("new", 0), + cooling_customer_count=lv.get("cooling", 0), + dormant_customer_count=lv.get("dormant", 0), + reactivated_customer_count=lv.get("reactivated", 0), + data_complete=True, + ) + ) + written += 1 + return written diff --git a/yunke-server/tests/test_employee_metrics.py b/yunke-server/tests/test_employee_metrics.py index ad7d891..4a35fa0 100644 --- a/yunke-server/tests/test_employee_metrics.py +++ b/yunke-server/tests/test_employee_metrics.py @@ -11,14 +11,18 @@ from zoneinfo import ZoneInfo import sqlalchemy as sa from app.db.models import ( + Customer, EmployeeCallDaily, + EmployeeMetricDaily, + FirstReplyFact, Message, + MsgDailyStats, PhoneCall, Staff, WechatAccount, WhatsappAccount, ) -from app.storage.employee_metrics import recompute_call_days +from app.storage.employee_metrics import recompute_call_days, recompute_metric_days _SH = ZoneInfo("Asia/Shanghai") @@ -93,3 +97,110 @@ async def test_call_daily_ignores_unknown_staff(make_session): await s.commit() assert (await s.execute( sa.select(sa.func.count()).select_from(EmployeeCallDaily))).scalar_one() == 0 + + +async def _seed_msg_rows(s, *, day=20260901, channel="wechat", owner="wx1", + kind="friend", peer="p1", send=3, recv=2): + s.add(MsgDailyStats(channel=channel, owner_account_id=owner, peer_kind=kind, + peer_id=peer, day=day, send_cnt=send, recv_cnt=recv)) + await s.commit() + + +async def test_metric_daily_friend_only_caliber(make_session): + """口径判别(v2 §1.1 #2 防线):room 行不进消息计数/串列,只进 active_rooms。""" + s = make_session + await _seed_org(s) + await _seed_msg_rows(s, send=3, recv=2, kind="friend", peer="p1") + await _seed_msg_rows(s, send=10, recv=0, kind="room", peer="r1") + # friend 会话给一条消息供串扫描(无消息则无串) + s.add(Message(msg_id="p1-0", channel="wechat", owner_account_id="wx1", + talker_id="p1", timestamp=_ts(2026, 9, 1, 10, 0), + is_send=False, type="text", text="q")) + await s.commit() + await recompute_metric_days(s, [20260901]) + await s.commit() + row = (await s.execute(sa.select(EmployeeMetricDaily))).scalar_one() + assert row.msg_sent == 3 and row.msg_recv == 2 # room 的 10 不进 + assert row.active_rooms == 1 + assert row.cust_threads == 1 and row.threads_answered == 0 # 未回复串 + + +async def test_metric_daily_ghost_filtered(make_session): + s = make_session + await _seed_org(s) + await _seed_msg_rows(s, owner="wx-ghost", peer="pg") + await recompute_metric_days(s, [20260901]) + await s.commit() + assert (await s.execute( + sa.select(sa.func.count()).select_from(EmployeeMetricDaily))).scalar_one() == 0 + + +async def test_metric_daily_same_source_as_fact(make_session): + """同源判别(v2 §1.1 #5 防线): + 1) threads_answered ≡ fact 行数(同归属日); + 2) 多日未回串宽表 threads < 日切 cust_threads 之和(分母不再膨胀); + 3) 幂等。""" + from app.storage.first_reply_facts import recompute_pairs + s = make_session + await _seed_org(s) + # 客户 D1 10:00、D1 20:00、D2 09:00 三条消息,始终未回 → 1 个跨日串 + rows = [(_ts(2026, 9, 1, 10, 0), False), (_ts(2026, 9, 1, 20, 0), False), + (_ts(2026, 9, 2, 9, 0), False)] + for i, (ts, is_send) in enumerate(rows): + s.add(Message(msg_id=f"g{i}", channel="wechat", owner_account_id="wx1", + talker_id="pg", timestamp=ts, is_send=is_send, + type="text", text="x")) + # 日切口径的 msg_daily_stats:D1、D2 各计一个未回串(供对照) + await _seed_msg_rows(s, day=20260901, peer="pg", send=0, recv=2) + await _seed_msg_rows(s, day=20260902, peer="pg", send=0, recv=1) + await s.commit() + await recompute_pairs(s, [("wechat", "wx1", "pg")]) + await s.commit() + await recompute_metric_days(s, [20260901, 20260902]) + await s.commit() + got = {r.day: r for r in ( + await s.execute(sa.select(EmployeeMetricDaily))).scalars().all()} + # 1) fact 行数=0(未回),宽表 threads_answered 也=0 且同源 + assert (await s.execute( + sa.select(sa.func.count()).select_from(FirstReplyFact))).scalar_one() == 0 + # 2) 跨日串只归属 D2;日切 D1+D2=2,宽表合计=1 + assert got[20260901].cust_threads == 0 + assert got[20260902].cust_threads == 1 + # 3) 幂等 + await recompute_metric_days(s, [20260901, 20260902]) + await s.commit() + assert len((await s.execute( + sa.select(EmployeeMetricDaily))).scalars().all()) == 2 + + +async def test_metric_daily_lifecycle_counts(make_session): + """生命周期列:cooling/dormant/reactivated 判定进宽表。""" + s = make_session + await _seed_org(s) + # 客户 pc:D1 有互动后静默(历史≥5 条) → as_of=10-10 为 dormant; + # 10-11 回流 → 该日 reactivated + s.add(Customer(friend_id="pc", channel="wechat", owner_account_id="wx1", + delete_flag=0)) + hist = [] + for h in range(9, 17): # D1 8 条消息(≥5) + hist.append((_ts(2026, 9, 1, h, 0), h % 2 == 0)) + for i, (ts, is_send) in enumerate(hist): + s.add(Message(msg_id=f"lc{i}", channel="wechat", owner_account_id="wx1", + talker_id="pc", timestamp=ts, is_send=is_send, + type="text", text="x")) + await _seed_msg_rows(s, day=20260901, peer="pc", send=4, recv=4) + await s.commit() + # D41(10-11)客户回流一条 + s.add(Message(msg_id="lc-b", channel="wechat", owner_account_id="wx1", + talker_id="pc", timestamp=_ts(2026, 10, 11, 9, 0), + is_send=False, type="text", text="back")) + await _seed_msg_rows(s, day=20261011, peer="pc", send=0, recv=1) + await s.commit() + from app.storage.first_reply_facts import recompute_pairs + await recompute_pairs(s, [("wechat", "wx1", "pc")]) + await recompute_metric_days(s, [20261010, 20261011]) + await s.commit() + got = {r.day: r for r in ( + await s.execute(sa.select(EmployeeMetricDaily))).scalars().all()} + assert got[20261010].dormant_customer_count == 1 # 39 天静默+历史 8 条 + assert got[20261011].reactivated_customer_count == 1 # 当日回流且 gap>30