From 3d5644c42a367e695c5e86ef79579a26dc9551d8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Wed, 9 Sep 2026 18:59:45 +0800 Subject: [PATCH] =?UTF-8?q?fix(management):=20drilldown=E7=94=9F=E5=91=BD?= =?UTF-8?q?=E5=91=A8=E6=9C=9F=E5=8E=86=E5=8F=B2=E6=94=B9peer=E9=94=AE?= =?UTF-8?q?=E6=8E=A7=E5=AF=B9=E9=BD=90=E6=89=B9=E6=AC=A1A=E5=8F=A3?= =?UTF-8?q?=E5=BE=84+comparisons=E9=94=AE=E5=9F=9F=E7=BB=9F=E4=B8=80+?= =?UTF-8?q?=E8=A1=A5=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 评审修正(I1/M1-M4/M6): - I1 lifecycle_customers 历史改 (channel,peer) 键控跨owner合并,归属仍由 Customer.owner_account_id 决定——与批次A _lifecycle_daily 同口径;新增 共享客户测试(历史在他人账号下仍全量呈现) - M3 employees 行 comparisons 键域统一为指标源名(config code→spec.metric 映射,active_friends_band/call_band 不再出现),与 sections 一致 - M4 未回复分支空 owners 先短路,免去无效 finalize 调用 - M1 employee_metrics._percentile 改公有 percentile 供同包复用,删除 api_assembly.percentile_nearest(存储层 _percentile 不同层保留) - M2 drilldown 内联 SQL(FirstReplyFact/PhoneCall)下沉 api_assembly (fr_fact_rows/user_calls_page),路由层无裸查询 - M6 补 start>end→422 断言;新增 config_source=db 测试(激活配置 min_days=9→五天数据员工翻 insufficient_sample,守护生效可见) tests/test_management_api.py: 21 passed;全量 721 passed Co-Authored-By: Claude Code --- yunke-server/app/api/routers/management.py | 67 ++------- yunke-server/app/management/api_assembly.py | 132 ++++++++++++++---- .../app/management/employee_metrics.py | 13 +- yunke-server/tests/test_management_api.py | 76 +++++++++- 4 files changed, 201 insertions(+), 87 deletions(-) diff --git a/yunke-server/app/api/routers/management.py b/yunke-server/app/api/routers/management.py index bd03586..f642fdc 100644 --- a/yunke-server/app/api/routers/management.py +++ b/yunke-server/app/api/routers/management.py @@ -21,13 +21,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.analytics import health as health_svc from app.analytics import metrics as metrics_svc from app.api.deps import get_session -from app.db.models import ( - FirstReplyFact, - PhoneCall, - Staff, - WechatAccount, - WhatsappAccount, -) +from app.db.models import Staff, WechatAccount, WhatsappAccount from app.management import api_assembly as asm from app.management.anomalies import detect_anomalies from app.management.employee_metrics import ( @@ -39,7 +33,6 @@ from app.management.employee_metrics import ( from app.management.metric_definitions import METRIC_DEFINITIONS from app.storage.employee_metrics import METRIC_VERSION from app.storage.staff_summary import account_map -from app.storage.summary import day_bounds_ms router = APIRouter(prefix="/api/management", tags=["management"]) @@ -316,7 +309,7 @@ async def management_employees( "total": total, "page": page, "page_size": page_size, - "rows": [asm.build_employee_row(u, d) for u, d in page_rows], + "rows": [asm.build_employee_row(u, d, config) for u, d in page_rows], "scope": {"channel": channel, "start": start, "end": end, "view": view}, "data_as_of": await health_svc.data_as_of(session), @@ -479,35 +472,15 @@ async def management_employee_drilldown( if metric_code in ("first_reply_p50", "first_reply_p90"): if not owners: return _paged([]) - conds = [FirstReplyFact.day.between(start, end), - FirstReplyFact.owner_account_id.in_(owners)] - if ch is not None: - conds.append(FirstReplyFact.channel == ch) - total = (await session.execute( - sa.select(sa.func.count()).select_from(FirstReplyFact) - .where(*conds) - )).scalar_one() - rows = (await session.execute( - sa.select(FirstReplyFact).where(*conds) - .order_by(FirstReplyFact.fr_ms.desc(), FirstReplyFact.day.desc(), - FirstReplyFact.thread_end_ts.desc()) - .offset((page - 1) * page_size).limit(page_size) - )).scalars().all() - items = [{ - "day": f.day, - "owner_account_id": f.owner_account_id, - "peer_id": f.peer_id, - "thread_end_ts": f.thread_end_ts, - "reply_ts": f.reply_ts, - "fr_ms": f.fr_ms, - # 会话定位参数提示(owner+talker 直达聊天记录) - "chat_hint": {"owner_account_id": f.owner_account_id, - "talker_id": f.peer_id}, - } for f in rows] - return {"items": items, "total": int(total), "page": page, + total, items = await asm.fr_fact_rows( + session, owners, ch, start, end, page=page, page_size=page_size + ) + return {"items": items, "total": total, "page": page, "page_size": page_size, "scope": scope} if metric_code == "unreplied": + if not owners: # 无账号→必空,免进上游决赛圈点查 + return _paged([]) # 上游无私有 owner 过滤:一次拉全量决赛圈结果(候选 ≤500 时完整), # 按 owner 账号过滤后内存分页;候选超限标 truncated(过滤可能不全) res = await metrics_svc.unreplied( @@ -533,27 +506,9 @@ async def management_employee_drilldown( # call_followup:窗口通话列表(start desc)+ 期聚合摘要 # (逐通话跟进标记复用批次A _followup_exists 太重,留批次C按需再加) - lo, _ = day_bounds_ms(start) - _, hi = day_bounds_ms(end) - conds = [PhoneCall.user_id == user_id, - PhoneCall.start_call_time.is_not(None), - PhoneCall.start_call_time >= lo, - PhoneCall.start_call_time < hi] - total = (await session.execute( - sa.select(sa.func.count()).select_from(PhoneCall).where(*conds) - )).scalar_one() - rows = (await session.execute( - sa.select(PhoneCall).where(*conds) - .order_by(PhoneCall.start_call_time.desc(), PhoneCall.call_id.desc()) - .offset((page - 1) * page_size).limit(page_size) - )).scalars().all() - items = [{ - "call_id": c.call_id, - "channel": c.channel, - "call_number": c.call_number, - "start_call_time": c.start_call_time, - "call_duration": c.call_duration, - } for c in rows] + total, items = await asm.user_calls_page( + session, user_id, start, end, page=page, page_size=page_size + ) summary_rows = await asm.call_daily_rows(session, start, end, user_id=user_id) summary = {col: sum(r[col] for r in summary_rows) diff --git a/yunke-server/app/management/api_assembly.py b/yunke-server/app/management/api_assembly.py index 6b8e98d..71c20c9 100644 --- a/yunke-server/app/management/api_assembly.py +++ b/yunke-server/app/management/api_assembly.py @@ -25,6 +25,7 @@ from app.db.models import ( EmployeeScoreConfig, FirstReplyFact, MsgDailyStats, + PhoneCall, Staff, ) from app.management.anomalies import detect_anomalies @@ -32,6 +33,7 @@ from app.management.employee_metrics import ( PeriodMetrics, baseline_window, day_range, + percentile, period_metrics, prev_window, team_medians, @@ -40,6 +42,7 @@ from app.management.metric_definitions import DEFAULT_SCORE_CONFIG from app.management.scoring import derive_metrics, score_employee from app.storage.employee_metrics import METRIC_VERSION from app.storage.staff_summary import account_map +from app.storage.summary import day_bounds_ms # 异常权重(anomaly_severity 排序/汇总):critical=2 / warning=1 SEVERITY_WEIGHT = {"critical": 2, "warning": 1} @@ -63,18 +66,6 @@ async def load_score_config(session: AsyncSession) -> tuple[dict, str]: return copy.deepcopy(DEFAULT_SCORE_CONFIG), "default" -def percentile_nearest( - vals: list[int], q_num: int, q_den: int -) -> int | None: - """最近邻秩分位 ceil(q·n)(与批次A/employee_metrics 同口径,自写 5 行 - 避免跨层引私有 _percentile);空列表→None(与 0 严格区分)。""" - if not vals: - return None - s = sorted(vals) - rank = max(1, min(len(s), (q_num * len(s) + q_den - 1) // q_den)) - return s[rank - 1] - - def team_derived_medians(metrics: dict[str, PeriodMetrics]) -> dict[str, float | None]: """评分用团队中位:PeriodMetrics 字段中位 + 派生率/占比的员工级中位 (scoring 要求 team 键=派生指标名,见其模块 docstring)。""" @@ -165,17 +156,31 @@ async def assemble_employees( return out -def build_employee_row(uid: str, data: dict) -> dict: +def _code_to_source(config: dict) -> dict[str, str]: + """配置指标 code → 源指标名(spec.metric,无则 code 本身)。 + + comparisons 键域统一为源指标名(与 derive_metrics / sections 端点一致, + 冻结批次C契约):active_friends_band→active_friends、call_band→call_count, + 其余 code 本身即源名。""" + out: dict[str, str] = {} + for dim in config.get("dimensions", {}).values(): + for code, spec in dim.get("metrics", {}).items(): + out[code] = spec.get("metric", code) + return out + + +def build_employee_row(uid: str, data: dict, config: dict) -> dict: """employees 列表行:metrics 全字段 + comparisons 四值(自评分 - metrics_detail 透传)+ score 摘要 + 异常计数/权重。""" + metrics_detail 透传,键映射为源指标名)+ score 摘要 + 异常计数/权重。""" score = data["score"] + code_src = _code_to_source(config) return { "user_id": uid, "name": data["name"], "is_on_job": data["is_on_job"], "metrics": data["cur"]._asdict(), "comparisons": { - code: { + code_src.get(code, code): { "current": d["value"], "prev": d["prev"], "baseline": d["baseline"], @@ -364,8 +369,8 @@ async def service_daily_trend( denom = threads_by_day.get(day, 0) out.append({ "day": day, - "fr_p50_ms": percentile_nearest(frs, 1, 2), - "fr_p90_ms": percentile_nearest(frs, 9, 10), + "fr_p50_ms": percentile(frs, 1, 2), + "fr_p90_ms": percentile(frs, 9, 10), "reply_2h_rate": (cnt_2h / denom) if denom else None, }) return out @@ -394,8 +399,10 @@ async def lifecycle_customers( 返回状态匹配行 {friend_id, nickname, last_day, state},按 last_day asc (最陈旧在前)、friend_id asc 排序。 - 与批次A _lifecycle_daily 同式:hist 为 ≤as_of 的历史消息数 - (时点语义,as_of 之后的消息不虚增沉睡门槛)。""" + 历史按 (channel, peer) 键控、跨 owner 合并——与批次A _lifecycle_daily + 的 peer 键控口径一致(归属由 Customer.owner_account_id 决定,互动史 + 不随归属走:共享/换归属客户的全部历史都算,否则 KPI 计了而名单丢人)。 + hist 为 ≤as_of 的历史消息数(时点语义,as_of 之后的消息不虚增沉睡门槛)。""" custs = (await session.execute( sa.select(Customer).where( sa.or_(Customer.delete_flag.is_(None), Customer.delete_flag == 0), @@ -408,7 +415,6 @@ async def lifecycle_customers( rows = (await session.execute( sa.select( MsgDailyStats.channel, - MsgDailyStats.owner_account_id, MsgDailyStats.peer_id, MsgDailyStats.day, MsgDailyStats.send_cnt, @@ -418,14 +424,14 @@ async def lifecycle_customers( MsgDailyStats.peer_id.in_(ids), ) )).all() - hist: dict[tuple[str, str, str], dict[int, int]] = {} - for ch, owner, peer, day, snd, rcv in rows: - dm = hist.setdefault((ch, owner, peer), {}) + hist: dict[tuple[str, str], dict[int, int]] = {} + for ch, peer, day, snd, rcv in rows: + dm = hist.setdefault((ch, peer), {}) dm[day] = dm.get(day, 0) + int(snd or 0) + int(rcv or 0) matches: list[dict] = [] for c in custs: - dm = hist.get((c.channel, c.owner_account_id or "", c.friend_id)) + dm = hist.get((c.channel, c.friend_id)) if not dm: continue days_list = sorted(dm) @@ -445,3 +451,81 @@ async def lifecycle_customers( }) matches.sort(key=lambda m: (m["last_day"], m["friend_id"])) return matches + + +async def fr_fact_rows( + session: AsyncSession, + owner_accounts: list[str], + channel: str | None, + start_day: int, + end_day: int, + *, + page: int, + page_size: int, +) -> tuple[int, list[dict]]: + """drilldown 首响分位:该员工账号窗内 fact 逐串证据行,fr_ms desc + (同值按日/串末时刻 desc 决胜)分页。chat_hint 给出会话定位参数 + (owner+talker 直达聊天记录)。返回 (total, items)。""" + conds = [ + FirstReplyFact.day.between(start_day, end_day), + FirstReplyFact.owner_account_id.in_(owner_accounts), + ] + if channel is not None: + conds.append(FirstReplyFact.channel == channel) + total = (await session.execute( + sa.select(sa.func.count()).select_from(FirstReplyFact).where(*conds) + )).scalar_one() + rows = (await session.execute( + sa.select(FirstReplyFact).where(*conds) + .order_by(FirstReplyFact.fr_ms.desc(), FirstReplyFact.day.desc(), + FirstReplyFact.thread_end_ts.desc()) + .offset((page - 1) * page_size).limit(page_size) + )).scalars().all() + items = [{ + "day": f.day, + "owner_account_id": f.owner_account_id, + "peer_id": f.peer_id, + "thread_end_ts": f.thread_end_ts, + "reply_ts": f.reply_ts, + "fr_ms": f.fr_ms, + "chat_hint": {"owner_account_id": f.owner_account_id, + "talker_id": f.peer_id}, + } for f in rows] + return int(total), items + + +async def user_calls_page( + session: AsyncSession, + user_id: str, + start_day: int, + end_day: int, + *, + page: int, + page_size: int, +) -> tuple[int, list[dict]]: + """drilldown 通话跟进:该员工窗内通话行(start_call_time desc, + call_id desc 决胜)分页。返回 (total, items)。""" + lo, _ = day_bounds_ms(start_day) + _, hi = day_bounds_ms(end_day) # 末日次日 00:00(左闭右开上界) + conds = [ + PhoneCall.user_id == user_id, + PhoneCall.start_call_time.is_not(None), + PhoneCall.start_call_time >= lo, + PhoneCall.start_call_time < hi, + ] + total = (await session.execute( + sa.select(sa.func.count()).select_from(PhoneCall).where(*conds) + )).scalar_one() + rows = (await session.execute( + sa.select(PhoneCall).where(*conds) + .order_by(PhoneCall.start_call_time.desc(), PhoneCall.call_id.desc()) + .offset((page - 1) * page_size).limit(page_size) + )).scalars().all() + items = [{ + "call_id": c.call_id, + "channel": c.channel, + "call_number": c.call_number, + "start_call_time": c.start_call_time, + "call_duration": c.call_duration, + } for c in rows] + return int(total), items diff --git a/yunke-server/app/management/employee_metrics.py b/yunke-server/app/management/employee_metrics.py index 977d255..e0b3959 100644 --- a/yunke-server/app/management/employee_metrics.py +++ b/yunke-server/app/management/employee_metrics.py @@ -88,12 +88,13 @@ def baseline_window(start: int) -> tuple[int, int]: return _to_key(bs), _to_key(be) -def _percentile(vals: list[int], q_num: int, q_den: int) -> int | None: +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 严格区分)。 + 公有供同包 api_assembly(B4 服务趋势)复用;整数算术 + rank=(q_num*n + q_den - 1)//q_den 对正整数 n 恒落在 [1,n] + (与 math.ceil(q·n) 等价,与批次A storage 层 _percentile 同口径—— + 后者为不同层单独保留);空列表→None(与 0 严格区分)。 """ if not vals: return None @@ -246,8 +247,8 @@ async def period_metrics( 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) + 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) diff --git a/yunke-server/tests/test_management_api.py b/yunke-server/tests/test_management_api.py index 3f66ea2..1fe4264 100644 --- a/yunke-server/tests/test_management_api.py +++ b/yunke-server/tests/test_management_api.py @@ -14,6 +14,7 @@ 生命周期判定(已删客户剔除)/call_followup 列表+期聚合摘要; - metric-definitions:trial 标记/口径说明/config_source 回退。 """ +import copy import datetime as dt from zoneinfo import ZoneInfo @@ -21,6 +22,7 @@ from app.db.models import ( Customer, EmployeeCallDaily, EmployeeMetricDaily, + EmployeeScoreConfig, FirstReplyFact, Message, MsgDailyStats, @@ -29,6 +31,7 @@ from app.db.models import ( WechatAccount, WhatsappAccount, ) +from app.management.metric_definitions import DEFAULT_SCORE_CONFIG _SH = ZoneInfo("Asia/Shanghai") @@ -190,6 +193,9 @@ async def test_dashboard_span_and_calendar_422(make_client): r = await client.get("/api/management/dashboard", params={"start": 20260230, "end": 20260905}) assert r.status_code == 422 # 日历非法(2 月 30 日) + r = await client.get("/api/management/dashboard", + params={"start": 20260906, "end": 20260901}) + assert r.status_code == 422 # start > end async def test_drilldown_unknown_metric_422(make_client): @@ -343,9 +349,13 @@ async def test_employees_default_sort_and_paging(make_client): assert u1["name"] == "张三" and u1["is_on_job"] is True assert u1["metrics"]["cust_threads"] == 10 assert u1["metrics"]["end_active_friends"] == 3 + # M3:comparisons 键域=源指标名(与 sections 端点一致;band 配置 code + # 映射为其 spec.metric——active_friends/call_count,不出现 *_band) assert set(u1["comparisons"]) >= { "active_share", "reply_2h_rate", "reply_24h_rate", "fr_p90_ms", - "active_friends_band", "call_band"} + "active_friends", "call_count"} + assert "active_friends_band" not in u1["comparisons"] + assert "call_band" not in u1["comparisons"] for c in u1["comparisons"].values(): assert set(c) == {"current", "prev", "baseline", "median"} assert u1["comparisons"]["fr_p90_ms"]["current"] == 120_000 @@ -436,6 +446,34 @@ async def test_employees_min_days_insufficient(make_client): assert row["score"]["total_score"] is not None # 明细保留展示 +async def test_employees_db_score_config_guard(make_client): + """config_source=db 路径:激活配置的 min_days=9 生效——u1 五天数据 + 在默认配置(min_days=5)下 scored,在 DB 配置下翻为 insufficient_sample。""" + + async def seed(s): + await _seed_common(s) + cfg = copy.deepcopy(DEFAULT_SCORE_CONFIG) + cfg["version"] = "v1-test" + cfg["name"] = "评审测试配置" + cfg["sample_guards"]["min_days"] = 9 + s.add(EmployeeScoreConfig(version="v1-test", name="评审测试配置", + status="active", config_json=cfg, + created_by="test")) + await s.commit() + + client = await make_client(seed) + r = await client.get("/api/management/employees", + params={"channel": "all", "start": START, + "end": END}) + assert r.status_code == 200 + body = r.json() + assert body["config_source"] == "db" + u1 = next(row for row in body["rows"] if row["user_id"] == "u1") + assert u1["metrics"]["days_with_data"] == 5 + assert u1["score"]["score_status"] == "insufficient_sample" # 5 < 9 + assert u1["score"]["total_score"] is not None + + # ---------------------------------------------------------------- detail @@ -593,6 +631,42 @@ async def test_drilldown_lifecycle_customers(make_client): assert body["items"][0]["state"] == "dormant" +async def test_drilldown_shared_customer_history(make_client): + """I1:共享/换归属客户——互动史挂在他人账号(wx2)名下而 Customer.owner + =wx1:历史按 (channel, peer) 键控跨 owner 合并(对齐批次A _lifecycle_daily + 的 peer 键控口径),客户不丢、历史完整(hist 12 ≥ 5 判沉睡)。""" + + async def seed(s): + await _seed_common(s) + for day, month in ((20260701, 7), (20260801, 8)): + s.add(MsgDailyStats(channel="wechat", owner_account_id="wx2", + peer_kind="friend", peer_id="f3", day=day, + send_cnt=3, recv_cnt=3, + last_ts=_ts(2026, month, day % 100, 12, 0), + last_is_send=True)) + s.add(Customer(friend_id="f3", channel="wechat", + owner_account_id="wx1", friend_nickname="共享客户", + user_id="u1", delete_flag=0)) + await s.commit() + + client = await make_client(seed) + r = await client.get("/api/management/employees/u1/drilldown", + params={"metric_code": "dormant_customers", + "channel": "all", + "start": START, "end": END}) + body = r.json() + assert body["total"] == 2 # f1 + f3(历史全在 wx2 名下也不丢) + f3 = next(i for i in body["items"] if i["friend_id"] == "f3") + assert f3 == {"friend_id": "f3", "nickname": "共享客户", + "last_day": 20260801, "state": "dormant"} + # 归属口径仍是 Customer.owner(wx1):u2(wx2)名下看不到 + r = await client.get("/api/management/employees/u2/drilldown", + params={"metric_code": "dormant_customers", + "channel": "all", + "start": START, "end": END}) + assert r.json()["total"] == 0 + + async def test_drilldown_call_followup(make_client): async def seed(s): await _seed_common(s)