From ad3beb76253a5a0a4661d8a431ed38afacae4bba Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Wed, 9 Sep 2026 14:44:35 +0800 Subject: [PATCH] =?UTF-8?q?fix(management):=20=E5=AE=BD=E8=A1=A8=E7=94=9F?= =?UTF-8?q?=E5=91=BD=E5=91=A8=E6=9C=9Fhist=E6=94=B9=E6=97=B6=E7=82=B9?= =?UTF-8?q?=E8=AF=AD=E4=B9=89(=E2=89=A4as=5Fof=E7=B4=AF=E7=A7=AF)=E2=80=94?= =?UTF-8?q?=E2=80=94=E5=8E=86=E5=8F=B2=E6=97=A5dormant=E4=B8=8D=E5=86=8D?= =?UTF-8?q?=E8=A2=AB=E6=9C=AA=E6=9D=A5=E6=B6=88=E6=81=AF=E8=99=9A=E5=A2=9E?= =?UTF-8?q?;=E8=A1=A5=E6=AD=A3=E5=90=91=E5=90=8C=E6=BA=90/percentile=20Non?= =?UTF-8?q?e=E5=88=A4=E5=88=AB+=E9=98=B2=E5=BE=A1=E6=80=A7flush?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- yunke-server/app/storage/employee_metrics.py | 26 +++++--- yunke-server/tests/test_employee_metrics.py | 63 ++++++++++++++++++++ 2 files changed, 80 insertions(+), 9 deletions(-) diff --git a/yunke-server/app/storage/employee_metrics.py b/yunke-server/app/storage/employee_metrics.py index d0c9a19..b9d337b 100644 --- a/yunke-server/app/storage/employee_metrics.py +++ b/yunke-server/app/storage/employee_metrics.py @@ -15,6 +15,7 @@ import logging import math from bisect import bisect_right from collections.abc import Iterable +from itertools import accumulate import sqlalchemy as sa from sqlalchemy.ext.asyncio import AsyncSession @@ -170,8 +171,10 @@ async def _lifecycle_daily( ) -> dict[int, dict[str, dict[str, int]]]: """逐日 {day: {user: {state: count}}}——customer 当前归属 + 互动日 sweep。 - 数据源:msg_daily_stats friend 行全历史(互动日/条数)。16k 客户 × 窗口 - 天数 bisect,夜间全量约秒级;4h tick 只算当日+3 天。 + 数据源:msg_daily_stats friend 行全历史(互动日/条数)。hist 为 ≤as_of + 的历史消息数(时点语义:与 last 同一 bisect 位取按日累积数组 cum[idx], + as_of 之后的消息不虚增历史日的沉睡门槛)。16k 客户 × 窗口天数 + bisect+classify,夜间全量实测见批次A核实文档;4h tick 只算当日+3 天。 """ rows = ( await session.execute( @@ -184,12 +187,13 @@ async def _lifecycle_daily( ) ) ).all() - days_map: dict[str, set[int]] = {} - cnt: dict[str, int] = {} + # peer → {day: 消息条数};days_list 与累积数组 cum 对齐: + # cum[i] = 该客户 ≤days_list[i] 的历史消息数(hist 时点语义) + day_msgs: dict[str, dict[int, 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()} + dm = day_msgs.setdefault(peer, {}) + dm[day] = dm.get(day, 0) + int(snd or 0) + int(rcv or 0) + interaction = {p: sorted(dm) for p, dm in day_msgs.items()} custs = ( await session.execute( @@ -212,14 +216,15 @@ async def _lifecycle_daily( if uid is None: continue first = days_list[0] - hist = cnt.get(peer, 0) + cum = list(accumulate(day_msgs[peer][d] for d in days_list)) 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) + hist_asof = cum[idx] + state = classify(first, last, prev, d, hist_asof) if state: out[d].setdefault(uid, {}).setdefault(state, 0) out[d][uid][state] += 1 @@ -345,4 +350,7 @@ async def recompute_metric_days( ) ) written += 1 + # 防御性 flush:防 autoflush=False 类配置下逐日 add 的 ORM 对象 + # 攒到事务尾(夜间全量约 52k 行;与 Task3 周期性 flush 同教训) + await session.flush() return written diff --git a/yunke-server/tests/test_employee_metrics.py b/yunke-server/tests/test_employee_metrics.py index 4a35fa0..9f76565 100644 --- a/yunke-server/tests/test_employee_metrics.py +++ b/yunke-server/tests/test_employee_metrics.py @@ -123,6 +123,7 @@ async def test_metric_daily_friend_only_caliber(make_session): 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 # 未回复串 + assert row.first_reply_p50_ms is None and row.first_reply_p90_ms is None # 空样本→None(≠0) async def test_metric_daily_ghost_filtered(make_session): @@ -173,6 +174,35 @@ async def test_metric_daily_same_source_as_fact(make_session): sa.select(EmployeeMetricDaily))).scalars().all()) == 2 +async def test_metric_daily_same_source_as_fact_answered(make_session): + """同源判别正向路径(v2 §1.1 #5):被回串——threads_answered ≡ fact 行数, + 单样本 p50=p90=fr_ms(锁分位非 None 映射)。""" + from app.storage.first_reply_facts import recompute_pairs + s = make_session + await _seed_org(s) + # 客户 10:00 提问,员工 10:25 回复 → fr=25min,串末条归属 09-01 + fr_ms = 25 * 60 * 1000 + s.add(Message(msg_id="an0", channel="wechat", owner_account_id="wx1", + talker_id="pa", timestamp=_ts(2026, 9, 1, 10, 0), + is_send=False, type="text", text="q")) + s.add(Message(msg_id="an1", channel="wechat", owner_account_id="wx1", + talker_id="pa", timestamp=_ts(2026, 9, 1, 10, 25), + is_send=True, type="text", text="a")) + await _seed_msg_rows(s, day=20260901, peer="pa", send=1, recv=1) + await s.commit() + await recompute_pairs(s, [("wechat", "wx1", "pa")]) + await s.commit() + await recompute_metric_days(s, [20260901]) + await s.commit() + n_facts = (await s.execute( + sa.select(sa.func.count()).select_from(FirstReplyFact))).scalar_one() + row = (await s.execute(sa.select(EmployeeMetricDaily))).scalar_one() + assert row.cust_threads == 1 + assert row.threads_answered == 1 == n_facts + assert row.fr_sum_ms == fr_ms + assert row.first_reply_p50_ms == fr_ms == row.first_reply_p90_ms + + async def test_metric_daily_lifecycle_counts(make_session): """生命周期列:cooling/dormant/reactivated 判定进宽表。""" s = make_session @@ -204,3 +234,36 @@ async def test_metric_daily_lifecycle_counts(make_session): 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 + + +async def test_metric_daily_lifecycle_hist_point_in_time(make_session): + """hist 时点语义判别:回填历史日 M,沉睡门槛只计 ≤M 的消息。 + + 客户 pd:X=09-01 共 2 条 + Y=10-02 共 3 条(全历史 5 条): + - M=10-01(X+30,