From 31cd2d92aa8f185246f7a69f8b0c816a50deb0ee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Wed, 9 Sep 2026 18:34:58 +0800 Subject: [PATCH] =?UTF-8?q?feat(management):=20=E5=8F=AA=E8=AF=BB=E7=AE=A1?= =?UTF-8?q?=E7=90=86API=E5=85=AD=E7=AB=AF=E7=82=B9=E2=80=94=E2=80=94dashbo?= =?UTF-8?q?ard/employees/detail/sections/drilldown=E7=99=BD=E5=90=8D?= =?UTF-8?q?=E5=8D=95/=E6=8C=87=E6=A0=87=E5=AE=9A=E4=B9=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- yunke-server/app/api/main.py | 2 + yunke-server/app/api/routers/management.py | 588 ++++++++++++++++++ yunke-server/app/management/api_assembly.py | 447 ++++++++++++++ yunke-server/tests/test_management_api.py | 633 ++++++++++++++++++++ 4 files changed, 1670 insertions(+) create mode 100644 yunke-server/app/api/routers/management.py create mode 100644 yunke-server/app/management/api_assembly.py create mode 100644 yunke-server/tests/test_management_api.py diff --git a/yunke-server/app/api/main.py b/yunke-server/app/api/main.py index 6971bb3..bf8ccd1 100644 --- a/yunke-server/app/api/main.py +++ b/yunke-server/app/api/main.py @@ -49,6 +49,7 @@ from app.api.routers import ( export, groups, login, + management, media, messages, moments, @@ -101,6 +102,7 @@ app.include_router(sessions.router) app.include_router(export.router) app.include_router(voice_clips.router) app.include_router(analytics.router) +app.include_router(management.router) @app.get("/api/health") diff --git a/yunke-server/app/api/routers/management.py b/yunke-server/app/api/routers/management.py new file mode 100644 index 0000000..bd03586 --- /dev/null +++ b/yunke-server/app/api/routers/management.py @@ -0,0 +1,588 @@ +"""GET /api/management/* —— 经营管理后台只读端组(批次B Task B4)。 + +六个端点:dashboard(总览)/ employees(列表,一期四视图共用行集)/ +employees/{user_id}(详情)/ employees/{user_id}/metrics(分段指标+趋势)/ +employees/{user_id}/drilldown(白名单下钻)/ metric-definitions(口径字典)。 + +惯例同 analytics.py:channel=all 在 router 层翻译为 None 传服务层; +start/end yyyymmdd 缺一/日历非法/start>end/跨度>366 天皆 422(_window); +重查询与行构造在 app.management.api_assembly,本文件只做薄编排。 +所有查询共用单 AsyncSession——**顺序 await,不并发**(见 assembly 模块)。 +""" +from __future__ import annotations + +import datetime as dt +from typing import Annotated, Any, Literal + +import sqlalchemy as sa +from fastapi import APIRouter, Depends, HTTPException, Query +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.management import api_assembly as asm +from app.management.anomalies import detect_anomalies +from app.management.employee_metrics import ( + baseline_window, + day_range, + period_metrics, + prev_window, +) +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"]) + +SessionDep = Annotated[AsyncSession, Depends(get_session)] +AllChannelParam = Annotated[Literal["all", "wechat", "whatsapp"], Query()] +YmdParam = Annotated[int, Query(ge=20000101, le=29991231)] +CompareParam = Annotated[ + Literal["previous", "personal_baseline", "team_median"], Query() +] +ViewParam = Annotated[ + Literal["management", "workload", "service", "customer"], Query() +] +StatusParam = Annotated[Literal["all", "on_job", "offline"], Query()] +SortByParam = Annotated[ + Literal["anomaly_severity", "total_score", "active_friends", "cust_threads"], + Query(), +] +SortOrderParam = Annotated[Literal["asc", "desc"], Query()] +SectionParam = Annotated[ + Literal["overview", "service", "customer", "workload", "calls"], Query() +] + +# drilldown 白名单(外 422 "unknown metric_code") +_DRILLDOWN_CODES = frozenset({ + "first_reply_p50", "first_reply_p90", "unreplied", + "cooling_customers", "dormant_customers", "call_followup", +}) + +# unreplied 适配:上游无私有 owner 过滤,取全量决赛圈结果再按 owner 过滤。 +# 500 = analytics.metrics._FINALIZE_ALL_MAX 同值(候选 ≤500 时单次调用即 +# 拿到全部已过滤 items;超过则上游本就按候选分页,响应标 truncated)。 +_UNREPLIED_FETCH = 500 + +# 三方口径差异说明(硬编码,与 METRIC_DEFINITIONS.caliber 互补的总述) +CALIBER_NOTES = ( + "管理域指标为 1:1 私聊口径且跨日精确(消息串按会话全历史计算、" + "归属串末条所在自然日);现有看板为私聊+群聊混合、日切口径" + "(跨夜串重复计/丢失,回复率分子分母同源近似)。" + "通话指标(通话数/时长/接通/跟进)无消息渠道维度——" + "voice_call 与 wa_voice_call 合并计数,不随 channel 参数过滤。" +) + +# 各 section 的期指标字段(自 PeriodMetrics 挑)与宽表/通话表趋势列 +_SECTION_FIELDS: dict[str, tuple[str, ...]] = { + "overview": ("days_with_data", "msg_sent", "msg_recv", "active_friends", + "active_rooms_avg"), + "service": ("cust_threads", "threads_answered", "fr_sum_ms", + "fr_p50_ms", "fr_p90_ms", "reply_2h_count", "reply_24h_count"), + "customer": ("new_customer", "cooling_customer", "dormant_customer", + "reactivated_customer", "end_active_friends"), + "workload": ("active_friends", "active_rooms_avg", "cust_threads"), + "calls": ("call_count", "call_seconds", "answered_call", "call_followup"), +} +_SECTION_TREND_COLS: dict[str, tuple[str, ...]] = { + "overview": ("msg_sent", "msg_recv", "active_friends"), + "service": ("cust_threads", "threads_answered", "fr_sum_ms", + "first_reply_p50_ms", "first_reply_p90_ms"), + "customer": ("new_customer_count", "cooling_customer_count", + "dormant_customer_count", "reactivated_customer_count"), + "workload": ("active_friends", "active_rooms"), + "calls": ("call_count", "call_seconds", "answered_call_count", + "call_followup_count"), +} + + +def _window(start: int | None, end: int | None) -> tuple[int, int]: + """管理端点公共窗口校验(照 analytics custom scope 惯例)。""" + if start is None or end is None: + raise HTTPException(422, "start and end are required") + try: + sd = dt.date(start // 10000, (start // 100) % 100, start % 100) + ed = dt.date(end // 10000, (end // 100) % 100, end % 100) + except ValueError as exc: # 如 20260230——Query 只拦范围不拦日历 + raise HTTPException(422, "invalid start/end calendar date") from exc + if sd > ed: + raise HTTPException(422, "start must be <= end") + if (ed - sd).days > 365: # 含端点共 366 天 + raise HTTPException(422, "span must be <= 366 days") + return start, end + + +def _translate(channel: str) -> str | None: + return None if channel == "all" else channel + + +async def _get_staff(session: AsyncSession, user_id: str) -> Staff: + staff = (await session.execute( + sa.select(Staff).where(Staff.user_id == user_id) + )).scalar_one_or_none() + if staff is None: + raise HTTPException(404, f"employee {user_id} not found") + return staff + + +@router.get("/dashboard") +async def management_dashboard( + session: SessionDep, + channel: AllChannelParam = "all", + start: YmdParam | None = None, + end: YmdParam | None = None, + compare: CompareParam = "previous", +) -> dict[str, Any]: + """经营管理总览:KPI + 双趋势 + 生命周期 + 维度分布 + 异常汇总 + + 行动入口计数,单往返。compare 仅回显(对比数值在 employees 行内)。""" + start, end = _window(start, end) + ch = _translate(channel) + config, config_source = await asm.load_score_config(session) + assembled = await asm.assemble_employees( + session, config, start, end, ch, with_prev=False + ) + pms = [a["cur"] for a in assembled.values()] + + # ---- 趋势两查:宽表全员逐日 + 通话小表逐日 + fact 逐日 ---- + wide = await asm.wide_daily_rows(session, start, end, ch) + wide_by_day = {d["day"]: d for d in wide} + threads_by_day = {d["day"]: d["cust_threads"] for d in wide} + calls = await asm.call_daily_rows(session, start, end) + call_by_day = {d["day"]: d["call_count"] for d in calls} + comm_trend = [] + for day in day_range(start, end): + d = wide_by_day.get(day) + comm_trend.append({ + "day": day, + "msg_total": (d["msg_sent"] + d["msg_recv"]) if d else 0, + # 团队合计口径:当日各员工宽表 active_friends 求和 + "active_customers": d["active_friends"] if d else 0, + "call_cnt": call_by_day.get(day, 0), + }) + service_trend = await asm.service_daily_trend( + session, start, end, ch, threads_by_day + ) + + lifecycle = { + "new": sum(p.new_customer for p in pms), + "cooling": sum(p.cooling_customer for p in pms), + "dormant": sum(p.dormant_customer for p in pms), + "reactivated": sum(p.reactivated_customer for p in pms), + } + kpis = { + # 全员 active_friends 求和:跨员工可能重复计客户(共享客户) + "active_customers": sum(p.active_friends for p in pms), + "active_employees": sum(1 for p in pms if p.days_with_data > 0), + "msg_total": sum(p.msg_sent + p.msg_recv for p in pms), + "cust_threads": sum(p.cust_threads for p in pms), + "threads_answered": sum(p.threads_answered for p in pms), + "call_count": sum(p.call_count for p in pms), + "call_seconds": sum(p.call_seconds for p in pms), + "answered_call": sum(p.answered_call for p in pms), + "lifecycle": lifecycle, + } + + # ---- 维度分布(全员 score_employee 的维度状态计数) ---- + dim_summary: dict[str, dict] = {} + for a in assembled.values(): + for key, dim in a["score"]["dimensions"].items(): + e = dim_summary.setdefault(key, { + "name": dim.get("name"), "weight": dim.get("weight", 0), + "scored_count": 0, "insufficient_count": 0, + "pending_count": 0, "_scores": [], + }) + if dim["status"] == "scored": + e["scored_count"] += 1 + e["_scores"].append(dim["score"]) + elif dim["status"] == "insufficient": + e["insufficient_count"] += 1 + else: # pending / pending_data + e["pending_count"] += 1 + for e in dim_summary.values(): + scores = e.pop("_scores") + e["avg_score"] = ( + sum(scores) / len(scores) if scores else None + ) + + # ---- 异常汇总:critical 先、warning 后,同 severity 按 affected desc ---- + anomalies = [a for data in assembled.values() for a in data["anomalies"]] + sev_rank = {"critical": 0, "warning": 1} + anomalies.sort(key=lambda a: ( + sev_rank.get(a["severity"], 9), + -a.get("affected_count", 0), + a["employee_user_id"], + )) + + return { + "scope": {"channel": channel, "start": start, "end": end, + "compare": compare}, + "data_as_of": await health_svc.data_as_of(session), + "metric_version": METRIC_VERSION, + "config_source": config_source, + "kpis": kpis, + "communication_trend": comm_trend, + "service_trend": service_trend, + "lifecycle": lifecycle, + "dimension_summary": dim_summary, + "anomalies": anomalies, + "action_summary": await metrics_svc.action_counts(session, channel=ch), + "caliber_notes": { + "active_customers": "kpis.active_customers 为全员周期 distinct " + "求和,跨员工共享客户可能重复计", + "trend_active_customers": "communication_trend.active_customers " + "为宽表日值团队合计(日口径求和)", + "reply_2h_rate": "分子=fact 当日 ≤2h 计数,分母=宽表当日串数", + }, + } + + +@router.get("/employees") +async def management_employees( + session: SessionDep, + channel: AllChannelParam = "all", + start: YmdParam | None = None, + end: YmdParam | None = None, + view: ViewParam = "management", + keyword: str | None = None, + status: StatusParam = "all", + sort_by: SortByParam = "anomaly_severity", + sort_order: SortOrderParam = "desc", + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), +) -> dict[str, Any]: + """员工列表(一期 management/workload/service/customer 四视图共用 + 行集,view 仅回显)。默认 anomaly_severity 排序:含 critical 者 → + 权重和 → 卡片数;total_score 排序 None 恒排最后。""" + start, end = _window(start, end) + ch = _translate(channel) + config, config_source = await asm.load_score_config(session) + assembled = await asm.assemble_employees( + session, config, start, end, ch, with_prev=True + ) + items = list(assembled.items()) + + if ch is not None: + # 渠道过滤=账号归属(该渠道无账号的员工不进列表;有账号无数据仍 + # 展示零值行) + owners_uids = set((await account_map(session, ch)).values()) + items = [(u, d) for u, d in items if u in owners_uids] + if keyword: + kw = keyword.strip().lower() + items = [ + (u, d) for u, d in items + if kw in (d["name"] or "").lower() or kw in u.lower() + ] + if status == "on_job": + items = [(u, d) for u, d in items if d["is_on_job"] is True] + elif status == "offline": + items = [(u, d) for u, d in items if d["is_on_job"] is not True] + + items.sort(key=lambda kv: kv[0]) # 决胜底座:user_id asc(稳定排序) + if sort_by == "anomaly_severity": + items.sort(key=lambda kv: ( + any(a["severity"] == "critical" for a in kv[1]["anomalies"]), + kv[1]["anomaly_weight"], kv[1]["anomaly_count"], + ), reverse=(sort_order == "desc")) + elif sort_by == "total_score": + # None(样本不足等)恒排最后,与升降序无关 + if sort_order == "desc": + items.sort(key=lambda kv: ( + kv[1]["score"]["total_score"] is None, + -(kv[1]["score"]["total_score"] or 0), + )) + else: + items.sort(key=lambda kv: ( + kv[1]["score"]["total_score"] is None, + kv[1]["score"]["total_score"] or 0, + )) + else: # active_friends / cust_threads(PeriodMetrics 数值列) + if sort_order == "desc": + items.sort(key=lambda kv: -getattr(kv[1]["cur"], sort_by)) + else: + items.sort(key=lambda kv: getattr(kv[1]["cur"], sort_by)) + + total = len(items) + page_rows = items[(page - 1) * page_size: page * page_size] + return { + "total": total, + "page": page, + "page_size": page_size, + "rows": [asm.build_employee_row(u, d) 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), + "metric_version": METRIC_VERSION, + "config_source": config_source, + } + + +@router.get("/employees/{user_id}") +async def management_employee_detail( + session: SessionDep, + user_id: str, + channel: AllChannelParam = "all", + start: YmdParam | None = None, + end: YmdParam | None = None, +) -> dict[str, Any]: + """员工详情:身份+账号清单+期指标全字段+评分完整明细(含逐指标 + metrics_detail)+异常卡片+prev/baseline 原始 PeriodMetrics。""" + start, end = _window(start, end) + ch = _translate(channel) + staff = await _get_staff(session, user_id) + config, config_source = await asm.load_score_config(session) + + cur_map = await period_metrics(session, start, end, ch, user_ids={user_id}) + bs, be = baseline_window(start) + base_map = await period_metrics(session, bs, be, ch, user_ids={user_id}) + ps, pe = prev_window(start, end) + prev_map = await period_metrics(session, ps, pe, ch, user_ids={user_id}) + cur, base, prev = (cur_map.get(user_id), base_map.get(user_id), + prev_map.get(user_id)) + if cur is None: # 防御:_get_staff 已保证在 staff,period_metrics 必含 + raise HTTPException(404, f"employee {user_id} not found") + team = asm.team_derived_medians( + await period_metrics(session, start, end, ch) + ) + score = asm.score_with_guards(config, cur, baseline=base, team=team, + prev=prev) + anomalies = detect_anomalies( + user_id, cur, base, n_days=len(day_range(start, end)), config=config + ) + + accounts: list[dict] = [] + for wa in (await session.execute( + sa.select(WechatAccount).where(WechatAccount.user_id == user_id) + )).scalars(): + accounts.append({ + "channel": "wechat", "account_id": wa.wx_id, + "nick_name": wa.nick_name, "user_name": wa.user_name, + }) + for wa in (await session.execute( + sa.select(WhatsappAccount).where(WhatsappAccount.user_id == user_id) + )).scalars(): + accounts.append({ + "channel": "whatsapp", "account_id": wa.uid, + "nick_name": wa.nick_name, "user_name": wa.user_name, + }) + + return { + "user_id": user_id, + "name": staff.name, + "is_on_job": staff.is_on_job, + "accounts": accounts, + "metrics": cur._asdict(), + "score": score, + "anomalies": anomalies, + "prev": prev._asdict() if prev is not None else None, + "baseline": base._asdict() if base is not None else None, + "scope": {"channel": channel, "start": start, "end": end}, + "data_as_of": await health_svc.data_as_of(session), + "metric_version": METRIC_VERSION, + "config_source": config_source, + } + + +@router.get("/employees/{user_id}/metrics") +async def management_employee_metrics( + session: SessionDep, + user_id: str, + section: SectionParam = "overview", + channel: AllChannelParam = "all", + start: YmdParam | None = None, + end: YmdParam | None = None, +) -> dict[str, Any]: + """员工分段指标:该域期指标 + 窗口逐日趋势 + 四对比 + (current/prev/baseline/median,派生率走 scoring.derive_metrics 口径)。""" + start, end = _window(start, end) + ch = _translate(channel) + await _get_staff(session, user_id) + cur_map = await period_metrics(session, start, end, ch, user_ids={user_id}) + bs, be = baseline_window(start) + base_map = await period_metrics(session, bs, be, ch, user_ids={user_id}) + ps, pe = prev_window(start, end) + prev_map = await period_metrics(session, ps, pe, ch, user_ids={user_id}) + cur, base, prev = (cur_map.get(user_id), base_map.get(user_id), + prev_map.get(user_id)) + if cur is None: + raise HTTPException(404, f"employee {user_id} not found") + team = asm.team_derived_medians( + await period_metrics(session, start, end, ch) + ) + + if section == "calls": + src_rows = await asm.call_daily_rows(session, start, end, + user_id=user_id) + else: + src_rows = await asm.wide_daily_rows(session, start, end, ch, + user_id=user_id) + by_day = {r["day"]: r for r in src_rows} + cols = _SECTION_TREND_COLS[section] + trend = [] + for day in day_range(start, end): + row = by_day.get(day) + point: dict[str, Any] = {"day": day} + for col in cols: + if row is not None: + point[col] = row.get(col) + else: # 无数据日:计数列补 0,分位列补 None + point[col] = None if col.startswith("first_reply") else 0 + trend.append(point) + + return { + "section": section, + "metrics": {f: getattr(cur, f) for f in _SECTION_FIELDS[section]}, + "trend": trend, + "comparisons": asm.derived_comparisons(cur, prev, base, team), + "scope": {"channel": channel, "start": start, "end": end, + "user_id": user_id}, + "data_as_of": await health_svc.data_as_of(session), + "metric_version": METRIC_VERSION, + } + + +@router.get("/employees/{user_id}/drilldown") +async def management_employee_drilldown( + session: SessionDep, + user_id: str, + metric_code: str, + channel: AllChannelParam = "all", + start: YmdParam | None = None, + end: YmdParam | None = None, + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), +) -> dict[str, Any]: + """指标下钻(白名单六项,外 422):首响分位逐串证据 / 未回串 / + 降温·沉睡客户名单 / 通话跟进列表。所有分支按员工 owner 账号过滤。""" + if metric_code not in _DRILLDOWN_CODES: + raise HTTPException(422, f"unknown metric_code: {metric_code}") + start, end = _window(start, end) + ch = _translate(channel) + await _get_staff(session, user_id) + scope = {"user_id": user_id, "metric_code": metric_code, + "channel": channel, "start": start, "end": end} + owners = await asm.user_owner_accounts(session, user_id, ch) + + def _paged(items: list[dict]) -> dict[str, Any]: + ix = (page - 1) * page_size + return {"items": items[ix: ix + page_size], "total": len(items), + "page": page, "page_size": page_size, "scope": scope} + + 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, + "page_size": page_size, "scope": scope} + + if metric_code == "unreplied": + # 上游无私有 owner 过滤:一次拉全量决赛圈结果(候选 ≤500 时完整), + # 按 owner 账号过滤后内存分页;候选超限标 truncated(过滤可能不全) + res = await metrics_svc.unreplied( + session, channel=ch, hours=24, page=1, page_size=_UNREPLIED_FETCH + ) + owner_set = set(owners) + owned = [i for i in res["items"] + if i["owner_account_id"] in owner_set] + out = _paged(owned) + out["note"] = ("按该员工 owner 账号自全量未回清单过滤" + "(hours=24,与看板清单同源同判定)") + if res["total"] > len(res["items"]): + out["truncated"] = True + return out + + if metric_code in ("cooling_customers", "dormant_customers"): + want = ("cooling" if metric_code == "cooling_customers" + else "dormant") + if not owners: + return _paged([]) + matches = await asm.lifecycle_customers(session, owners, end, want) + return _paged(matches) + + # 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] + summary_rows = await asm.call_daily_rows(session, start, end, + user_id=user_id) + summary = {col: sum(r[col] for r in summary_rows) + for col in ("call_count", "call_seconds", + "answered_call_count", "call_followup_count")} + return {"items": items, "total": int(total), "page": page, + "page_size": page_size, "summary": summary, "scope": scope, + "note": "逐通话跟进标记留批次C;此处为窗口通话列表+期聚合" + "(followup 为接通后24h内有跟进的通话数)"} + + +@router.get("/metric-definitions") +async def management_metric_definitions( + session: SessionDep, +) -> dict[str, Any]: + """指标口径字典 + 当前评分配置(含 pending 维度)+ 三方口径差异说明。 + config_source: "db"=激活配置 / "default"=未种子回退默认。""" + config, source = await asm.load_score_config(session) + return { + "definitions": METRIC_DEFINITIONS, + "score_config": { + "version": config.get("version"), + "name": config.get("name"), + "trial": config.get("trial"), + "normalization": config.get("normalization"), + "sample_guards": config.get("sample_guards"), + "grade_bounds": config.get("grade_bounds"), + "dimensions": config.get("dimensions"), + }, + "config_source": source, + "caliber_notes": CALIBER_NOTES, + } diff --git a/yunke-server/app/management/api_assembly.py b/yunke-server/app/management/api_assembly.py new file mode 100644 index 0000000..6b8e98d --- /dev/null +++ b/yunke-server/app/management/api_assembly.py @@ -0,0 +1,447 @@ +"""管理 API 组装层(批次B Task B4)——router 六端点共用的重查询与行构造。 + +拆出 router 的原因:employees/dashboard 的组装(多窗聚合 + 评分 + 异常 + +对比派生)上百行,放 router 会淹没端点定义;本模块只做数据组装,参数 +校验/HTTP 语义留在 router。 + +查询纪律(与 analytics.metrics 同):**禁止 asyncio.gather**——单 +AsyncSession 不能并发执行查询,全部顺序 await。重端点不做 N+1: +period_metrics 每窗一次调用全员;员工循环(评分/异常)是纯函数。 +""" +from __future__ import annotations + +import copy +import statistics +from bisect import bisect_right + +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import AsyncSession + +from app.analytics.lifecycle import classify +from app.db.models import ( + Customer, + EmployeeCallDaily, + EmployeeMetricDaily, + EmployeeScoreConfig, + FirstReplyFact, + MsgDailyStats, + Staff, +) +from app.management.anomalies import detect_anomalies +from app.management.employee_metrics import ( + PeriodMetrics, + baseline_window, + day_range, + period_metrics, + prev_window, + team_medians, +) +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 + +# 异常权重(anomaly_severity 排序/汇总):critical=2 / warning=1 +SEVERITY_WEIGHT = {"critical": 2, "warning": 1} + +# 派生指标中需对「员工派生值」另取中位的键(scoring.team 键=派生指标名; +# 其余如 fr_p90_ms/active_friends/call_count 直接用 team_medians 输出) +_DERIVED_MEDIAN_KEYS = ("reply_2h_rate", "reply_24h_rate", "active_share") + + +async def load_score_config(session: AsyncSession) -> tuple[dict, str]: + """激活评分配置(DB)→ (config_json, "db");无 active 行 → + (DEFAULT_SCORE_CONFIG 深拷贝, "default")——未跑种子的环境也能出分。""" + row = (await session.execute( + sa.select(EmployeeScoreConfig) + .where(EmployeeScoreConfig.status == "active") + .order_by(EmployeeScoreConfig.id.desc()) + .limit(1) + )).scalar_one_or_none() + if row is not None: + return row.config_json, "db" + 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)。""" + team = dict(team_medians(metrics)) + for key in _DERIVED_MEDIAN_KEYS: + vals = [ + v for v in (derive_metrics(pm).get(key) for pm in metrics.values()) + if v is not None + ] + team[key] = statistics.median(vals) if vals else None + return team + + +def score_with_guards( + config: dict, + cur: PeriodMetrics, + baseline: PeriodMetrics | None = None, + team: dict[str, float] | None = None, + prev: PeriodMetrics | None = None, +) -> dict: + """score_employee + min_days 接线(B2 评审遗留):当前窗天数 < + config.sample_guards.min_days → score_status 覆盖 insufficient_sample, + total/明细保留展示(按天稀疏的数据不参与考核但可查看)。""" + score = score_employee(config, cur, baseline=baseline, team=team, prev=prev) + min_days = config.get("sample_guards", {}).get("min_days", 5) + if cur.days_with_data < min_days: + score["score_status"] = "insufficient_sample" + return score + + +async def assemble_employees( + session: AsyncSession, + config: dict, + start_day: int, + end_day: int, + channel: str | None, + *, + with_prev: bool = True, +) -> dict[str, dict]: + """全员组装:staff → 当前/基线(/前)窗 period_metrics → 逐员工 + score_with_guards + detect_anomalies(纯函数循环)。 + + dashboard 传 with_prev=False(只需评分+异常,省一次前窗聚合); + employees/detail 需要 prev/团队中位时传 True。返回 + {uid: {name, is_on_job, cur, prev, baseline, score, anomalies, + anomaly_count, anomaly_weight}}——过滤/排序/分页由调用方做。 + """ + staff_rows = (await session.execute( + sa.select(Staff.user_id, Staff.name, Staff.is_on_job) + )).all() + staff = {uid: (name, onjob) for uid, name, onjob in staff_rows if uid} + if not staff: + return {} + + n_days = len(day_range(start_day, end_day)) + cur = await period_metrics(session, start_day, end_day, channel) + bs, be = baseline_window(start_day) + base = await period_metrics(session, bs, be, channel) + prev: dict[str, PeriodMetrics] | None = None + team: dict[str, float | None] | None = None + if with_prev: + ps, pe = prev_window(start_day, end_day) + prev = await period_metrics(session, ps, pe, channel) + team = team_derived_medians(cur) + + out: dict[str, dict] = {} + for uid, (name, onjob) in staff.items(): + c = cur.get(uid) + if c is None: # 防御:period_metrics 恒含全员,此处不该走到 + continue + b = base.get(uid) + p = prev.get(uid) if prev is not None else None + score = score_with_guards(config, c, baseline=b, team=team, prev=p) + anomalies = detect_anomalies(uid, c, b, n_days=n_days, config=config) + out[uid] = { + "name": name, + "is_on_job": onjob, + "cur": c, + "prev": p, + "baseline": b, + "score": score, + "anomalies": anomalies, + "anomaly_count": len(anomalies), + "anomaly_weight": sum( + SEVERITY_WEIGHT.get(a["severity"], 0) for a in anomalies + ), + } + return out + + +def build_employee_row(uid: str, data: dict) -> dict: + """employees 列表行:metrics 全字段 + comparisons 四值(自评分 + metrics_detail 透传)+ score 摘要 + 异常计数/权重。""" + score = data["score"] + return { + "user_id": uid, + "name": data["name"], + "is_on_job": data["is_on_job"], + "metrics": data["cur"]._asdict(), + "comparisons": { + code: { + "current": d["value"], + "prev": d["prev"], + "baseline": d["baseline"], + "median": d["median"], + } + for code, d in score["metrics_detail"].items() + }, + "score": { + "total_score": score["total_score"], + "grade": score["grade"], + "score_status": score["score_status"], + "trial": score["trial"], + "dimensions": { + key: {"score": dim["score"], "status": dim["status"]} + for key, dim in score["dimensions"].items() + }, + }, + "anomaly_count": data["anomaly_count"], + "anomaly_weight": data["anomaly_weight"], + } + + +def derived_comparisons( + cur: PeriodMetrics, + prev: PeriodMetrics | None, + baseline: PeriodMetrics | None, + team: dict[str, float | None] | None, +) -> dict[str, dict]: + """四对比(current/prev/baseline/median),派生口径=scoring.derive_metrics + ——评分、异常、展示三方同一派生式。""" + cv = derive_metrics(cur) + pv = derive_metrics(prev) if prev is not None else {} + bv = derive_metrics(baseline) if baseline is not None else {} + return { + key: { + "current": cv.get(key), + "prev": pv.get(key), + "baseline": bv.get(key), + "median": team.get(key) if team else None, + } + for key in cv + } + + +async def wide_daily_rows( + session: AsyncSession, + start_day: int, + end_day: int, + channel: str | None, + user_id: str | None = None, +) -> list[dict]: + """宽表逐日聚合(加性列跨渠道求和;user_id=None 为全员合计)。 + + first_reply_p50/p90_ms 是 (day,channel,user) 行内单值不可求和: + 单员工视图(channel=None 双渠道同日)取当日 cust_threads 较大渠道的 + 值(当日主力渠道,并列取 wechat);全员视图不取用这两列(分位由 + service_daily_trend 从 fact 精确算)。 + """ + conds = [ + EmployeeMetricDaily.day.between(start_day, end_day), + EmployeeMetricDaily.metric_version == METRIC_VERSION, + ] + if channel is not None: + conds.append(EmployeeMetricDaily.channel == channel) + if user_id is not None: + conds.append(EmployeeMetricDaily.user_id == user_id) + rows = (await session.execute( + sa.select( + EmployeeMetricDaily.day.label("day"), + EmployeeMetricDaily.channel.label("ch"), + sa.func.sum(EmployeeMetricDaily.msg_sent).label("sent"), + sa.func.sum(EmployeeMetricDaily.msg_recv).label("recv"), + sa.func.sum(EmployeeMetricDaily.active_friends).label("af"), + sa.func.sum(EmployeeMetricDaily.active_rooms).label("ar"), + 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.new_customer_count).label("nw"), + sa.func.sum(EmployeeMetricDaily.cooling_customer_count).label("cool"), + sa.func.sum(EmployeeMetricDaily.dormant_customer_count).label("dorm"), + sa.func.sum(EmployeeMetricDaily.reactivated_customer_count).label("react"), + sa.func.max(EmployeeMetricDaily.first_reply_p50_ms).label("p50"), + sa.func.max(EmployeeMetricDaily.first_reply_p90_ms).label("p90"), + ).where(*conds).group_by(EmployeeMetricDaily.day, EmployeeMetricDaily.channel) + )).all() + + per_day: dict[int, dict] = {} + + def _blank(day: int) -> dict: + return { + "day": day, "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, + "first_reply_p50_ms": None, "first_reply_p90_ms": None, + "_pick": (-1, -1), # 主力渠道追踪 (cust_threads, wechat 加分) + } + + for (day, ch, sent, recv, af, ar, thr, ans, frs, nw, cool, dorm, + react, p50, p90) in rows: + d = per_day.setdefault(day, _blank(day)) + d["msg_sent"] += int(sent or 0) + d["msg_recv"] += int(recv or 0) + d["active_friends"] += int(af or 0) + d["active_rooms"] += int(ar or 0) + d["cust_threads"] += int(thr or 0) + d["threads_answered"] += int(ans or 0) + d["fr_sum_ms"] += int(frs or 0) + d["new_customer_count"] += int(nw or 0) + d["cooling_customer_count"] += int(cool or 0) + d["dormant_customer_count"] += int(dorm or 0) + d["reactivated_customer_count"] += int(react or 0) + if user_id is not None: + pick = (int(thr or 0), 1 if ch == "wechat" else 0) + if pick > d["_pick"]: + d["_pick"] = pick + d["first_reply_p50_ms"] = p50 + d["first_reply_p90_ms"] = p90 + + out = [] + for day in sorted(per_day): + d = per_day[day] + d.pop("_pick", None) + out.append(d) + return out + + +async def call_daily_rows( + session: AsyncSession, + start_day: int, + end_day: int, + user_id: str | None = None, +) -> list[dict]: + """通话小表逐日聚合(恒全渠道——表无渠道维);user_id=None 为全员。""" + conds = [EmployeeCallDaily.day.between(start_day, end_day)] + if user_id is not None: + conds.append(EmployeeCallDaily.user_id == user_id) + rows = (await session.execute( + sa.select( + EmployeeCallDaily.day, + sa.func.sum(EmployeeCallDaily.call_count), + sa.func.sum(EmployeeCallDaily.call_seconds), + sa.func.sum(EmployeeCallDaily.answered_call_count), + sa.func.sum(EmployeeCallDaily.call_followup_count), + ).where(*conds).group_by(EmployeeCallDaily.day) + )).all() + return [ + { + "day": day, "call_count": int(cc or 0), + "call_seconds": int(cs or 0), + "answered_call_count": int(ca or 0), + "call_followup_count": int(cf or 0), + } + for day, cc, cs, ca, cf in rows + ] + + +async def service_daily_trend( + session: AsyncSession, + start_day: int, + end_day: int, + channel: str | None, + threads_by_day: dict[int, int], +) -> list[dict]: + """服务趋势(dashboard 全团队):fact 逐日 (day, fr_ms) 行拉回 Python + 最近邻秩分位(跨员工合并样本——团队级分位);reply_2h_rate = fact + 当日 ≤2h 计数 / 宽表当日 cust_threads 合计(分母 0 → None)。 + + 输出全窗口逐日(无数据日 p50/p90 为 None),threads_by_day 来自 + wide_daily_rows 的全员 cust_threads 列。 + """ + conds = [FirstReplyFact.day.between(start_day, end_day)] + if channel is not None: + conds.append(FirstReplyFact.channel == channel) + rows = (await session.execute( + sa.select(FirstReplyFact.day, FirstReplyFact.fr_ms).where(*conds) + )).all() + by_day: dict[int, list[int]] = {} + for day, fr in rows: + by_day.setdefault(day, []).append(int(fr)) + out = [] + for day in day_range(start_day, end_day): + frs = by_day.get(day, []) + cnt_2h = sum(1 for f in frs if f <= 7_200_000) + 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), + "reply_2h_rate": (cnt_2h / denom) if denom else None, + }) + return out + + +async def user_owner_accounts( + session: AsyncSession, user_id: str, channel: str | None +) -> list[str]: + """员工在所选渠道下的 owner 账号(channel=None → 两渠道合并)。""" + chs = ("wechat", "whatsapp") if channel is None else (channel,) + out: list[str] = [] + for ch in chs: + amap = await account_map(session, ch) + out.extend(owner for owner, uid in amap.items() if uid == user_id) + return out + + +async def lifecycle_customers( + session: AsyncSession, + owner_accounts: list[str], + as_of_day: int, + want_state: str, +) -> list[dict]: + """drilldown cooling/dormant:该员工账号归属客户(delete_flag 过滤)+ + msg_daily_stats friend 行全历史 → lifecycle.classify(as_of=窗口末日), + 返回状态匹配行 {friend_id, nickname, last_day, state},按 last_day asc + (最陈旧在前)、friend_id asc 排序。 + + 与批次A _lifecycle_daily 同式:hist 为 ≤as_of 的历史消息数 + (时点语义,as_of 之后的消息不虚增沉睡门槛)。""" + custs = (await session.execute( + sa.select(Customer).where( + sa.or_(Customer.delete_flag.is_(None), Customer.delete_flag == 0), + Customer.owner_account_id.in_(owner_accounts), + ) + )).scalars().all() + if not custs: + return [] + ids = [c.friend_id for c in custs] + rows = (await session.execute( + sa.select( + MsgDailyStats.channel, + MsgDailyStats.owner_account_id, + MsgDailyStats.peer_id, + MsgDailyStats.day, + MsgDailyStats.send_cnt, + MsgDailyStats.recv_cnt, + ).where( + MsgDailyStats.peer_kind == "friend", + 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), {}) + 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)) + if not dm: + continue + days_list = sorted(dm) + idx = bisect_right(days_list, as_of_day) - 1 + if idx < 0: + continue # 窗口末日之前从未互动 + last = days_list[idx] + prev_d = days_list[idx - 1] if idx > 0 else None + hist_msgs = sum(dm[d] for d in days_list[: idx + 1]) + state = classify(days_list[0], last, prev_d, as_of_day, hist_msgs) + if state == want_state: + matches.append({ + "friend_id": c.friend_id, + "nickname": c.friend_nickname or c.friend_remark or c.friend_id, + "last_day": last, + "state": state, + }) + matches.sort(key=lambda m: (m["last_day"], m["friend_id"])) + return matches diff --git a/yunke-server/tests/test_management_api.py b/yunke-server/tests/test_management_api.py new file mode 100644 index 0000000..3f66ea2 --- /dev/null +++ b/yunke-server/tests/test_management_api.py @@ -0,0 +1,633 @@ +"""管理 API 六只读端点判别测试(批次B Task B4)。 + +覆盖: +- 参数校验:422(缺 start / 跨度>366 / 日历非法 / drilldown 白名单外 / + page_size>100)、404(employees/{ghost}、drilldown 员工不存在); +- dashboard 各段齐全与关键数值(kpis/communication_trend/service_trend/ + lifecycle/dimension_summary/anomalies/action_summary/data_as_of/ + metric_version); +- employees:默认分页 20/total/critical 优先默认排序/channel=wechat 时 + whatsapp-only 员工消失/keyword/status/min_days 不足→insufficient_sample; +- detail:账号清单+metrics 全字段+score 完整明细+anomalies+prev/baseline; +- sections:五段各有 trend 数组与四对比; +- drilldown:p90 分页与 fr_ms desc/unreplied 按 owner 过滤/cooling-dormant + 生命周期判定(已删客户剔除)/call_followup 列表+期聚合摘要; +- metric-definitions:trial 标记/口径说明/config_source 回退。 +""" +import datetime as dt +from zoneinfo import ZoneInfo + +from app.db.models import ( + Customer, + EmployeeCallDaily, + EmployeeMetricDaily, + FirstReplyFact, + Message, + MsgDailyStats, + PhoneCall, + Staff, + WechatAccount, + WhatsappAccount, +) + +_SH = ZoneInfo("Asia/Shanghai") + +START, END = 20260901, 20260905 # 当前窗 5 天(min_days=5 恰好够) + + +def _ts(*args) -> int: + return int(dt.datetime(*args, tzinfo=_SH).timestamp() * 1000) + + +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(day // 10000, day // 100 % 100, day % 100, 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 _seed_common(s): + """三员工全量种子:u1(微信,2 warning)/u2(双渠道,1 critical+1 warning)/ + u3(仅 WhatsApp,无异常)——支撑 dashboard/employees/detail/sections/ + drilldown 全部判据。""" + for uid, name, onjob in (("u1", "张三", True), ("u2", "李四", True), + ("u3", "王五", False)): + s.add(Staff(user_id=uid, name=name, is_on_job=onjob)) + s.add(WechatAccount(wx_id="wx1", user_id="u1", nick_name="张三的微信", + user_name="张三")) + s.add(WechatAccount(wx_id="wx2", user_id="u2")) + s.add(WhatsappAccount(uid="wa2", user_id="u2")) + s.add(WhatsappAccount(uid="wa3", user_id="u3")) + + # ---- 当前窗宽表(0901-0905)---- + for d in range(1, 6): + s.add(_mday(20260900 + d, user_id="u1", msg_sent=1, cust_threads=2, + threads_answered=2, fr_sum_ms=240_000, active_friends=3, + active_rooms=1, cooling_customer_count=2, + dormant_customer_count=1, + first_reply_p50_ms=110_000, first_reply_p90_ms=220_000)) + s.add(_mday(20260900 + d, user_id="u2", msg_sent=10, msg_recv=10, + cust_threads=2, threads_answered=2, fr_sum_ms=1_200_000, + active_friends=4, cooling_customer_count=1, + reactivated_customer_count=1)) + s.add(_mday(20260900 + d, channel="whatsapp", user_id="u3", + msg_sent=5, msg_recv=5, cust_threads=1, active_friends=1)) + # ---- 基线窗宽表(0827-0831,u1 高日均触发 activity_drop)---- + for d in range(27, 32): + s.add(_mday(20260800 + d, user_id="u1", msg_sent=14, msg_recv=14)) + s.add(_mday(20260800 + d, user_id="u2", msg_sent=5, msg_recv=5)) + + # ---- fact:当前窗(u1 120s warning / u2 600s critical)+ 基线 60s ---- + s.add(_fact(120_000, peer="pA", day=20260901, i=0)) + s.add(_fact(120_000, peer="pB", day=20260902, i=1)) + s.add(_fact(600_000, owner="wx2", peer="pC", day=20260901, i=2)) + s.add(_fact(600_000, owner="wx2", peer="pD", day=20260902, i=3)) + for owner, peer, day, i in (("wx1", "pA", 20260827, 10), + ("wx1", "pB", 20260828, 11), + ("wx2", "pC", 20260827, 12), + ("wx2", "pD", 20260828, 13)): + s.add(_fact(60_000, owner=owner, peer=peer, day=day, i=i)) + + # ---- msg_daily_stats:活跃客户 distinct + data_as_of + 未回候选 ---- + s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="p1", day=20260901, + send_cnt=0, recv_cnt=1, last_ts=_ts(2026, 9, 1, 10, 0), + last_is_send=False)) + s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="p2", day=20260902, + send_cnt=1, recv_cnt=0, last_ts=_ts(2026, 9, 2, 10, 0), + last_is_send=True)) + s.add(MsgDailyStats(channel="wechat", owner_account_id="wx2", + peer_kind="friend", peer_id="p3", day=20260901, + send_cnt=2, recv_cnt=0, last_ts=_ts(2026, 9, 1, 11, 0), + last_is_send=True)) + s.add(MsgDailyStats(channel="whatsapp", owner_account_id="wa3", + peer_kind="friend", peer_id="p4", day=20260903, + send_cnt=1, recv_cnt=0, last_ts=_ts(2026, 9, 3, 9, 0), + last_is_send=True)) + # drilldown cooling/dormant 客户互动史(全历史,窗外): + # f1 末次 0801(gap 35, hist 12)→ dormant;f2 末次 0825(gap 11)→ cooling + for day, n in ((20260701, 6), (20260801, 6)): + s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="f1", day=day, + send_cnt=n, recv_cnt=0, + last_ts=_ts(day // 10000, day // 100 % 100, + day % 100, 10, 0), + last_is_send=True)) + for day in (20260801, 20260825): + s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="f2", day=day, + send_cnt=2, recv_cnt=2, + last_ts=_ts(2026, 8, day % 100, 11, 0), + last_is_send=True)) + + # ---- 客户档案(f9 已删,须被 drilldown 剔除)---- + s.add(Customer(friend_id="f1", channel="wechat", owner_account_id="wx1", + friend_nickname="客户一", user_id="u1", delete_flag=0)) + s.add(Customer(friend_id="f2", channel="wechat", owner_account_id="wx1", + friend_nickname="客户二", user_id="u1", delete_flag=0)) + s.add(Customer(friend_id="f9", channel="wechat", owner_account_id="wx1", + friend_nickname="已删客户", user_id="u1", delete_flag=1)) + + # ---- 未回候选 p1 的尾部消息(unreplied 决赛圈点查)---- + s.add(Message(msg_id="m-p1", channel="wechat", owner_account_id="wx1", + talker_id="p1", is_send=False, type="text", text="在吗", + timestamp=_ts(2026, 9, 1, 10, 0))) + + # ---- 通话 ---- + s.add(EmployeeCallDaily(day=20260901, user_id="u1", call_count=2, + call_seconds=100, answered_call_count=1, + call_followup_count=1)) + s.add(EmployeeCallDaily(day=20260902, user_id="u2", call_count=1, + call_seconds=60, answered_call_count=1, + call_followup_count=0)) + s.add(PhoneCall(call_id="c1", channel="voice_call", user_id="u1", + start_call_time=_ts(2026, 9, 1, 15, 0), call_duration=120, + call_number="13800000001")) + s.add(PhoneCall(call_id="c2", channel="voice_call", user_id="u1", + start_call_time=_ts(2026, 9, 2, 15, 0), call_duration=0, + call_number="13800000002")) + await s.commit() + + +# ---------------------------------------------------------------- 参数与错误 + + +async def test_dashboard_missing_start_422(make_client): + async def seed(s): + await _seed_common(s) + + client = await make_client(seed) + r = await client.get("/api/management/dashboard", params={"end": END}) + assert r.status_code == 422 + + +async def test_dashboard_span_and_calendar_422(make_client): + async def seed(s): + await _seed_common(s) + + client = await make_client(seed) + r = await client.get("/api/management/dashboard", + params={"start": 20250101, "end": 20261231}) + assert r.status_code == 422 + r = await client.get("/api/management/dashboard", + params={"start": 20260230, "end": 20260905}) + assert r.status_code == 422 # 日历非法(2 月 30 日) + + +async def test_drilldown_unknown_metric_422(make_client): + async def seed(s): + await _seed_common(s) + + client = await make_client(seed) + r = await client.get("/api/management/employees/u1/drilldown", + params={"metric_code": "bogus", + "start": START, "end": END}) + assert r.status_code == 422 + assert "unknown metric_code" in r.json()["detail"] + + +async def test_employees_page_size_over_100_422(make_client): + async def seed(s): + await _seed_common(s) + + client = await make_client(seed) + r = await client.get("/api/management/employees", + params={"start": START, "end": END, + "page_size": 101}) + assert r.status_code == 422 + + +async def test_employee_detail_404(make_client): + async def seed(s): + await s.commit() + + client = await make_client(seed) + r = await client.get("/api/management/employees/ghost", + params={"start": START, "end": END}) + assert r.status_code == 404 + + +async def test_drilldown_employee_404(make_client): + async def seed(s): + await s.commit() + + client = await make_client(seed) + r = await client.get("/api/management/employees/ghost/drilldown", + params={"metric_code": "unreplied", + "start": START, "end": END}) + assert r.status_code == 404 + + +# ---------------------------------------------------------------- dashboard + + +async def test_dashboard_structure_and_kpis(make_client): + async def seed(s): + await _seed_common(s) + + client = await make_client(seed) + r = await client.get("/api/management/dashboard", + params={"channel": "all", "start": START, "end": END}) + assert r.status_code == 200 + body = r.json() + for key in ("scope", "data_as_of", "metric_version", "kpis", + "communication_trend", "service_trend", "lifecycle", + "dimension_summary", "anomalies", "action_summary", + "config_source", "caliber_notes"): + assert key in body, key + assert body["metric_version"] == "v1" + assert body["config_source"] == "default" + assert body["scope"] == {"channel": "all", "start": START, "end": END, + "compare": "previous"} + assert body["data_as_of"] == _ts(2026, 9, 3, 9, 0) + + k = body["kpis"] + # 活跃客户=PM distinct 求和(u1:2 + u2:1 + u3:1) + assert k["active_customers"] == 4 + assert k["active_employees"] == 3 + assert k["msg_total"] == 155 # u1 5 + u2 100 + u3 50 + assert k["cust_threads"] == 25 # 10 + 10 + 5 + assert k["threads_answered"] == 20 # 10 + 10 + 0 + assert k["call_count"] == 3 + assert k["call_seconds"] == 160 + assert k["answered_call"] == 2 + # 期末存量:u1 cooling2/dormant1 + u2 cooling1;reactivated 期内求和 u2 5 + assert body["lifecycle"] == {"new": 0, "cooling": 3, "dormant": 1, + "reactivated": 5} + assert k["lifecycle"] == body["lifecycle"] + + # 趋势逐日 5 点;0901 = u1(1)+u2(20)+u3(10) 消息,宽表 active 合计 8,通话 2 + trend = body["communication_trend"] + assert [t["day"] for t in trend] == [20260901 + i for i in range(5)] + assert trend[0]["msg_total"] == 31 + assert trend[0]["active_customers"] == 8 + assert trend[0]["call_cnt"] == 2 + assert trend[1]["call_cnt"] == 1 + assert trend[4] == {"day": 20260905, "msg_total": 31, + "active_customers": 8, "call_cnt": 0} + + # 服务趋势:0901 fact [120k, 600k] → p50=120k/p90=600k; + # reply_2h_rate = 2/5(两串均 ≤2h——600s=10min / 宽表当日串 5) + st = {t["day"]: t for t in body["service_trend"]} + assert st[20260901]["fr_p50_ms"] == 120_000 + assert st[20260901]["fr_p90_ms"] == 600_000 + assert st[20260901]["reply_2h_rate"] == 0.4 + assert st[20260903]["fr_p50_ms"] is None # 无 fact 日 + assert st[20260903]["reply_2h_rate"] == 0.0 # 分母有串、分子 0 + + # 维度摘要:5 维全员计数,总数=员工数 3 + dims = body["dimension_summary"] + assert set(dims) == {"customer", "service", "action", "workload", "risk"} + for d in dims.values(): + assert d["scored_count"] + d["insufficient_count"] + d["pending_count"] == 3 + assert "avg_score" in d + + # 异常:critical 在前(u2),warning 其后按 affected_count desc + anoms = body["anomalies"] + assert anoms[0]["severity"] == "critical" + assert anoms[0]["employee_user_id"] == "u2" + assert all(a["severity"] == "critical" for a in anoms[:1]) + assert all(a["severity"] == "warning" for a in anoms[1:]) + + assert body["action_summary"] == {"unreplied_2h": 1, "churn_real_30d": 0, + "sensitive_pending": 0} + + +# ---------------------------------------------------------------- employees + + +async def test_employees_default_sort_and_paging(make_client): + async def seed(s): + await _seed_common(s) + + 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["total"] == 3 + assert body["page"] == 1 + assert body["page_size"] == 20 + assert body["metric_version"] == "v1" + assert body["config_source"] == "default" + assert len(body["rows"]) == 3 + + # 默认 anomaly_severity 排序:含 critical 的 u2 在最前 + # (u2=1 critical——基线无串样本率不可算,rate 规则不触发;u1=2 warning) + ids = [row["user_id"] for row in body["rows"]] + assert ids == ["u2", "u1", "u3"] + u2, u1, u3 = body["rows"] + assert u2["anomaly_count"] == 1 and u2["anomaly_weight"] == 2 + assert u1["anomaly_count"] == 2 and u1["anomaly_weight"] == 2 + assert u3["anomaly_count"] == 0 and u3["anomaly_weight"] == 0 + + # 行结构:metrics 全字段 / comparisons 四值 / score 摘要 + assert u1["name"] == "张三" and u1["is_on_job"] is True + assert u1["metrics"]["cust_threads"] == 10 + assert u1["metrics"]["end_active_friends"] == 3 + assert set(u1["comparisons"]) >= { + "active_share", "reply_2h_rate", "reply_24h_rate", "fr_p90_ms", + "active_friends_band", "call_band"} + for c in u1["comparisons"].values(): + assert set(c) == {"current", "prev", "baseline", "median"} + assert u1["comparisons"]["fr_p90_ms"]["current"] == 120_000 + assert u1["comparisons"]["fr_p90_ms"]["baseline"] == 60_000 + assert set(u1["score"]["dimensions"]) == { + "customer", "service", "action", "workload", "risk"} + for d in u1["score"]["dimensions"].values(): + assert set(d) == {"score", "status"} + assert u1["score"]["trial"] is True + assert u1["score"]["score_status"] == "scored" # 5 天数据 ≥ min_days + + +async def test_employees_channel_filter(make_client): + async def seed(s): + await _seed_common(s) + + client = await make_client(seed) + r = await client.get("/api/management/employees", + params={"channel": "wechat", "start": START, + "end": END}) + ids = [row["user_id"] for row in r.json()["rows"]] + assert ids == ["u2", "u1"] # u3 仅 WhatsApp,wechat 视图消失 + r = await client.get("/api/management/employees", + params={"channel": "whatsapp", "start": START, + "end": END}) + ids = [row["user_id"] for row in r.json()["rows"]] + assert ids == ["u2", "u3"] # u1 无 WA 账号 + + +async def test_employees_keyword(make_client): + async def seed(s): + await _seed_common(s) + + client = await make_client(seed) + r = await client.get("/api/management/employees", + params={"start": START, "end": END, "keyword": "张"}) + assert [row["user_id"] for row in r.json()["rows"]] == ["u1"] + r = await client.get("/api/management/employees", + params={"start": START, "end": END, "keyword": "U3"}) + assert [row["user_id"] for row in r.json()["rows"]] == ["u3"] + + +async def test_employees_status_filter(make_client): + async def seed(s): + await _seed_common(s) + + client = await make_client(seed) + r = await client.get("/api/management/employees", + params={"start": START, "end": END, + "status": "on_job"}) + assert [row["user_id"] for row in r.json()["rows"]] == ["u2", "u1"] + r = await client.get("/api/management/employees", + params={"start": START, "end": END, + "status": "offline"}) + assert [row["user_id"] for row in r.json()["rows"]] == ["u3"] + + +async def test_employees_min_days_insufficient(make_client): + """B2 评审遗留接线:days_with_data < min_days → score_status 覆盖 + insufficient_sample,但可评明细(total_score)保留展示。""" + + async def seed(s): + s.add(Staff(user_id="u9", name="赵九", is_on_job=True)) + s.add(WechatAccount(wx_id="wx9", user_id="u9")) + # 仅 2 天宽表数据(< min_days=5),但串样本 30 ≥ min_threads 可评分 + for d in (1, 2): + s.add(_mday(20260900 + d, user_id="u9", cust_threads=15, + threads_answered=15, fr_sum_ms=900_000, + active_friends=10)) + s.add(_fact(60_000, owner="wx9", peer="q1", day=20260901, i=0)) + s.add(_fact(60_000, owner="wx9", peer="q2", day=20260902, i=1)) + for i in range(6): # 6 个 distinct peer → workload band 可评 + s.add(MsgDailyStats(channel="wechat", owner_account_id="wx9", + peer_kind="friend", peer_id=f"q{i}", + day=20260901, send_cnt=1, recv_cnt=0, + last_ts=_ts(2026, 9, 1, 12, 0), + last_is_send=True)) + await s.commit() + + client = await make_client(seed) + r = await client.get("/api/management/employees", + params={"start": START, "end": END}) + assert r.status_code == 200 + row = r.json()["rows"][0] + assert row["user_id"] == "u9" + assert row["metrics"]["days_with_data"] == 2 + assert row["score"]["score_status"] == "insufficient_sample" + assert row["score"]["total_score"] is not None # 明细保留展示 + + +# ---------------------------------------------------------------- detail + + +async def test_employee_detail_full(make_client): + async def seed(s): + await _seed_common(s) + + client = await make_client(seed) + r = await client.get("/api/management/employees/u1", + params={"channel": "all", "start": START, "end": END}) + assert r.status_code == 200 + body = r.json() + assert body["user_id"] == "u1" + assert body["name"] == "张三" + assert body["accounts"] == [ + {"channel": "wechat", "account_id": "wx1", + "nick_name": "张三的微信", "user_name": "张三"}] + # metrics 全字段(22 列 namedtuple) + assert len(body["metrics"]) == 22 + assert body["metrics"]["cust_threads"] == 10 + # score 完整明细:dimensions 含逐指标 metrics;metrics_detail 扁平 + assert body["score"]["dimensions"]["service"]["metrics"]["fr_p90_ms"][ + "score"] == 100.0 + assert body["score"]["metrics_detail"]["fr_p90_ms"]["value"] == 120_000 + assert body["score"]["metrics_detail"]["reply_2h_rate"]["value"] == 0.2 + assert body["score"]["metrics_detail"]["fr_p90_ms"]["baseline"] == 60_000 + # anomalies + prev/baseline 原始值(prev 窗 0827-0831 与基线窗重叠: + # 5 天 × msg_sent 14 = 70) + assert len(body["anomalies"]) == 2 + assert all(a["severity"] == "warning" for a in body["anomalies"]) + assert body["prev"]["msg_sent"] == 70 + assert body["baseline"]["msg_sent"] == 70 + assert body["data_as_of"] == _ts(2026, 9, 3, 9, 0) + + +# ---------------------------------------------------------------- sections + + +async def test_employee_sections(make_client): + async def seed(s): + await _seed_common(s) + + client = await make_client(seed) + base = "/api/management/employees/u1/metrics" + for section in ("overview", "service", "customer", "workload", "calls"): + r = await client.get(base, params={ + "section": section, "channel": "wechat", + "start": START, "end": END}) + assert r.status_code == 200, section + body = r.json() + assert body["section"] == section + assert len(body["trend"]) == 5 # 全窗口逐日(含零填充) + assert all("day" in t for t in body["trend"]) + assert body["comparisons"]["reply_2h_rate"]["current"] == 0.2 + assert set(body["comparisons"]["reply_2h_rate"]) == { + "current", "prev", "baseline", "median"} + + r = await client.get(base, params={"section": "service", + "channel": "wechat", + "start": START, "end": END}) + body = r.json() + assert body["metrics"]["cust_threads"] == 10 + assert body["metrics"]["reply_2h_count"] == 2 + t0 = body["trend"][0] + assert t0 == {"day": 20260901, "cust_threads": 2, "threads_answered": 2, + "fr_sum_ms": 240_000, "first_reply_p50_ms": 110_000, + "first_reply_p90_ms": 220_000} + + r = await client.get(base, params={"section": "calls", "channel": "all", + "start": START, "end": END}) + calls = {t["day"]: t for t in r.json()["trend"]} + assert calls[20260901]["call_count"] == 2 + assert calls[20260901]["call_followup_count"] == 1 + assert calls[20260905]["call_count"] == 0 # 零填充 + assert r.json()["metrics"]["call_count"] == 2 + + +# ---------------------------------------------------------------- drilldown + + +async def test_drilldown_fr_p90(make_client): + async def seed(s): + await _seed_common(s) + + client = await make_client(seed) + url = "/api/management/employees/u2/drilldown" + r = await client.get(url, params={ + "metric_code": "first_reply_p90", "channel": "all", + "start": START, "end": END, "page": 1, "page_size": 1}) + assert r.status_code == 200 + body = r.json() + assert body["total"] == 2 + assert len(body["items"]) == 1 + item = body["items"][0] + # fr_ms desc + 同值按日 desc 决胜:0902 的 pD 在前 + assert item["fr_ms"] == 600_000 + assert item["peer_id"] == "pD" + assert item["owner_account_id"] == "wx2" + assert item["chat_hint"] == {"owner_account_id": "wx2", "talker_id": "pD"} + r = await client.get(url, params={ + "metric_code": "first_reply_p90", "channel": "all", + "start": START, "end": END, "page": 2, "page_size": 1}) + assert r.json()["items"][0]["peer_id"] == "pC" + # u2 的 WA 账号无 fact 行 + r = await client.get(url, params={ + "metric_code": "first_reply_p50", "channel": "whatsapp", + "start": START, "end": END}) + assert r.json()["total"] == 0 + + +async def test_drilldown_unreplied(make_client): + async def seed(s): + await _seed_common(s) + + client = await make_client(seed) + r = await client.get("/api/management/employees/u1/drilldown", + params={"metric_code": "unreplied", + "channel": "wechat", + "start": START, "end": END}) + assert r.status_code == 200 + body = r.json() + assert body["total"] == 1 # p1(wx1)进入,u1 过滤后 + assert body["items"][0]["peer_id"] == "p1" + assert body["items"][0]["owner_account_id"] == "wx1" + # u2 名下无未回候选(p3 尾部为员工消息) + r = await client.get("/api/management/employees/u2/drilldown", + params={"metric_code": "unreplied", + "channel": "wechat", + "start": START, "end": END}) + assert r.json()["total"] == 0 + + +async def test_drilldown_lifecycle_customers(make_client): + async def seed(s): + await _seed_common(s) + + client = await make_client(seed) + r = await client.get("/api/management/employees/u1/drilldown", + params={"metric_code": "cooling_customers", + "channel": "all", + "start": START, "end": END}) + assert r.status_code == 200 + body = r.json() + assert body["total"] == 1 # f2 降温;f1 沉睡、f9 已删 + assert body["items"][0] == {"friend_id": "f2", "nickname": "客户二", + "last_day": 20260825, "state": "cooling"} + + 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"] == 1 + assert body["items"][0]["friend_id"] == "f1" + assert body["items"][0]["state"] == "dormant" + + +async def test_drilldown_call_followup(make_client): + async def seed(s): + await _seed_common(s) + + client = await make_client(seed) + r = await client.get("/api/management/employees/u1/drilldown", + params={"metric_code": "call_followup", + "channel": "all", + "start": START, "end": END}) + assert r.status_code == 200 + body = r.json() + assert body["total"] == 2 + # start_call_time desc:c2(0902)在前 + assert [i["call_id"] for i in body["items"]] == ["c2", "c1"] + assert set(body["items"][0]) == { + "call_id", "channel", "call_number", "start_call_time", + "call_duration"} + assert body["summary"] == {"call_count": 2, "call_seconds": 100, + "answered_call_count": 1, + "call_followup_count": 1} + + +# ------------------------------------------------------- metric-definitions + + +async def test_metric_definitions(make_client): + async def seed(s): + await _seed_common(s) + + client = await make_client(seed) + r = await client.get("/api/management/metric-definitions") + assert r.status_code == 200 + body = r.json() + assert body["definitions"]["cust_threads"]["unit"] == "串" + assert body["score_config"]["trial"] is True + assert body["score_config"]["dimensions"]["risk"]["status"] == "pending_data" + assert body["config_source"] == "default" # 未种子配置→回退默认 + assert "1:1" in body["caliber_notes"]