fix(management): drilldown生命周期历史改peer键控对齐批次A口径+comparisons键域统一+补测试

评审修正(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 <noreply@anthropic.com>
2026-09-09 18:59:45 +08:00
parent 31cd2d92aa
commit 3d5644c42a
4 changed files with 201 additions and 87 deletions
@@ -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