xinyouhe/yunke
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
57ace987922339df60d00674dcf234925860045e
yunke/docs/superpowers/plans/2026-09-09-management-console-batch-b-api-plan.md
T

31 KiB
Raw Blame History

云客经营管理后台·批次B（评分/异常/管理API）实施计划

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 在批次A数据底座之上交付只读管理 API——期聚合查询、五维试算评分、确定性异常规则、指标元数据、6 个 /api/management/* 端点，OpenAPI 可验证；前端（批次C）紧随其后。

Architecture: 新建 app/management/ 业务层（metric_definitions / employee_metrics 期聚合 / scoring / anomalies），app/api/routers/management.py 只做参数校验与编排（照 analytics 惯例：SessionDep + Annotated 参数 + dict 返回）。本期只做 6 个只读端点（dashboard / employees / employee detail / metrics sections / drilldown 白名单 / metric-definitions）；reviews 与 followups 写端点属批次E。依据：v2 计划 §1（决策记录）/§6（API 设计）/§8（Task 3-5）；批次A已交付全部数据表与回填。

Tech Stack: 同批次A（FastAPI + SQLAlchemy async；测试内存 SQLite；生产 MySQL 8.4）。

执行者必读：批次A交付物与既定口径

  1. 数据表就绪（迁移 b7c8d9e0f1a2，本地已回填）：first_reply_fact（68,476 行，PK 含 reply_msg_id）、employee_metric_daily（12,371 行，metric_version='v1'，串类列与 fact 同源）、employee_call_daily（1,163 行）、employee_score_config（空，本批次 B0 种子）。
  2. 复用模块：app/storage/first_reply_facts.py（fact 表直接查询）、app/analytics/lifecycle.py（阈值常量）、app/analytics/metrics.py 的 action_counts（dashboard 行动入口聚合）、app/analytics/health.py 的 data_as_of。
  3. 周期口径：start/end 为 yyyymmdd 上海自然日（与 analytics custom scope 同校验：缺一 422、日历非法 422、start>end 422、跨度 ≤366 天）；上期=等长紧邻前窗；个人基线=紧邻当前窗之前的 28 天窗。
  4. 评分默认（全部标「试算」，config 里可调）：维度权重 客户经营30/服务25/行动执行20(pending_data)/有效负载15/合规风险10(pending_data)；可用维度归一化 ×100/70；样本门槛 min=30 串/soft=10 串（周期内）；等级 A≥90/B≥80/C≥70/D≥60/E。
  5. 鉴权：app 级 require_auth 全局生效，management 路由无需额外处理；GET only。
  6. 测试命令：cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_management_api.py -v；提交中文 conventional；master 直提（项目惯例）；仓库有并行会话活动——git status 后只 add 自己的文件。
  7. 前端不做（批次C另出计划）；schemas_management.py 留批次E写端点用（analytics 惯例 dict 返回，本期一致）；错误响应沿用 analytics 的 HTTPException(422/404, message) 风格（v1 §12.4 的稳定 error_code 为有意偏差，前端按 HTTP 状态码分支）。

文件地图

文件 职责 动作
yunke-server/app/management/__init__.py 包标记（空） 新建
yunke-server/app/management/metric_definitions.py 指标元数据注册表 + 默认评分配置 DEFAULT_SCORE_CONFIG 新建
yunke-server/app/management/employee_metrics.py 期聚合查询（宽表 SUM/fact 窗口函数分位/期末生命周期/通话/distinct 活跃/四对比） 新建
yunke-server/app/management/scoring.py 标准化/维度分/归一化试算/样本保护/等级 新建
yunke-server/app/management/anomalies.py 4 条确定性规则 → v1 §5.4 异常卡片 新建
yunke-server/app/sync/cli.py score-config 种子子命令 修改
yunke-server/app/api/routers/management.py 6 只读端点 新建
yunke-server/app/api/main.py include_router 修改
yunke-server/tests/test_management_scoring.py 等 4 个测试文件 判别测试 新建
docs/superpowers/plans/2026-09-08-management-batch-a-verification.md 批次B冒烟数据追加 修改（B5）

Task B0: 指标元数据 + 默认评分配置 + 种子 CLI

Files:

  • Create: yunke-server/app/management/__init__.py（空文件）

  • Create: yunke-server/app/management/metric_definitions.py

  • Modify: yunke-server/app/sync/cli.py（score-config 子命令）

  • Test: yunke-server/tests/test_management_scoring.py（本任务先写配置/种子部分）

  • Step 1: 写测试（先失败）

tests/test_management_scoring.py：

"""管理域评分配置/引擎判别测试(批次B)。"""
import json

import sqlalchemy as sa

from app.db.models import EmployeeScoreConfig
from app.management.metric_definitions import (
    DEFAULT_SCORE_CONFIG,
    METRIC_DEFINITIONS,
    validate_config,
)


def test_config_weights_sum_and_structure():
    dims = DEFAULT_SCORE_CONFIG["dimensions"]
    assert sum(d["weight"] for d in dims.values()) == 100
    # 一期 pending_data 的维度不参与试算
    available = [d for d in dims.values() if d.get("status") != "pending_data"]
    assert sum(d["weight"] for d in available) == 70  # 30+25+15
    assert dims["action"]["status"] == "pending_data"
    assert dims["risk"]["status"] == "pending_data"


def test_config_normalization_declared():
    assert DEFAULT_SCORE_CONFIG["normalization"] == "available_only_rescale"


def test_validate_config_rejects_bad_weight_sum():
    bad = json.loads(json.dumps(DEFAULT_SCORE_CONFIG))  # deep copy
    bad["dimensions"]["service"]["weight"] = 26  # 总和 101
    problems = validate_config(bad)
    assert any("权重" in p for p in problems)


def test_metric_definitions_cover_scored_metrics():
    # 每个参与评分的 metric_code 都有元数据(名称/单位/方向)
    for dim in DEFAULT_SCORE_CONFIG["dimensions"].values():
        for code in dim.get("metrics", {}):
            md = METRIC_DEFINITIONS[code]
            assert md["name"] and md["unit"] and md["direction"] in (1, -1)


async def test_seed_and_idempotent(make_session):
    from app.sync.cli import _seed_score_config  # 纯会话函数(见 Step 3)
    s = make_session
    v1 = await _seed_score_config(s)
    await s.commit()
    assert v1 == "v1.0"
    rows = (await s.execute(sa.select(EmployeeScoreConfig))).scalars().all()
    assert len(rows) == 1 and rows[0].status == "active"
    # 幂等:已存在同版本返回 None 不新建
    assert await _seed_score_config(s) is None
    await s.commit()
    assert len((await s.execute(sa.select(EmployeeScoreConfig))).scalars().all()) == 1

  • Step 2: 跑测试确认失败（ModuleNotFoundError: app.management）

  • Step 3: 实现 metric_definitions.py

"""管理域指标元数据与默认评分配置(批次B,v2 §6/§10)。

METRIC_DEFINITIONS 是 /api/management/metric-definitions 的单一数据源:
名称/单位/方向/口径说明(含与现有看板的三方口径差异标注)。
DEFAULT_SCORE_CONFIG 为 v1.0 试算起点——所有阈值标「试算」,调整走
新配置版本(employee_score_config),不静默改数。
"""
from __future__ import annotations

# direction: +1 越大越好 / -1 越小越好;caliber 注明与看板差异
METRIC_DEFINITIONS: dict[str, dict] = {
    "msg_sent": {"name": "发出消息(1:1)", "unit": "条", "direction": 1,
                 "caliber": "仅私聊;现有看板为私聊+群聊混合"},
    "msg_recv": {"name": "收到消息(1:1)", "unit": "条", "direction": 1,
                 "caliber": "仅私聊"},
    "cust_threads": {"name": "客户消息串", "unit": "串", "direction": 1,
                     "caliber": "跨日精确,按串末条归属日;看板为日切口径(多日串重复计)"},
    "threads_answered": {"name": "被回串数", "unit": "串", "direction": 1,
                         "caliber": "≡first_reply_fact行数(同源)"},
    "fr_avg_ms": {"name": "平均首响", "unit": "ms", "direction": -1,
                  "caliber": "fr_sum_ms/threads_answered"},
    "fr_p50_ms": {"name": "首响中位数", "unit": "ms", "direction": -1,
                  "caliber": "周期内 fact 精确分位(窗口函数)"},
    "fr_p90_ms": {"name": "首响P90", "unit": "ms", "direction": -1,
                  "caliber": "周期内 fact 精确分位;日切口径会丢跨夜长尾"},
    "reply_2h_rate": {"name": "2h回复率", "unit": "0-1", "direction": 1,
                      "formula": "count(fr_ms<=2h)/cust_threads",
                      "caliber": "分子分母同源(fact/宽表同一算法)"},
    "reply_24h_rate": {"name": "24h回复率", "unit": "0-1", "direction": 1,
                       "formula": "count(fr_ms<=24h)/cust_threads"},
    "active_friends": {"name": "活跃客户(周期)", "unit": "人", "direction": 1,
                       "caliber": "期内distinct客户(msg_daily_stats直查),非宽表日值求和"},
    "active_rooms": {"name": "活跃群(日均)", "unit": "个", "direction": 1},
    "new_customer_count": {"name": "新客(期末)", "unit": "人", "direction": 1,
                           "caliber": "期末单日持有状态,非期内求和"},
    "cooling_customer_count": {"name": "降温客户(期末)", "unit": "人", "direction": -1,
                               "caliber": "距今天数口径(7-29天);频次基线口径二期"},
    "dormant_customer_count": {"name": "沉睡客户(期末)", "unit": "人", "direction": -1,
                               "caliber": "≥30天且历史≥5条"},
    "reactivated_customer_count": {"name": "回流客户(期内)", "unit": "人", "direction": 1,
                                   "caliber": "当日回流且前次间隔>30天,期内求和"},
    "call_count": {"name": "通话数", "unit": "通", "direction": 1,
                   "caliber": "语音通话(微信+WA),无消息渠道维度"},
    "answered_rate": {"name": "接通率", "unit": "0-1", "direction": 1,
                      "formula": "answered_call/call_count", "caliber": "接通=duration>0"},
    "call_followup_rate": {"name": "通话后跟进率", "unit": "0-1", "direction": 1,
                           "formula": "call_followup_count/answered_call",
                           "caliber": "接通后24h内任一账号有1:1消息(含客户发来)"},
    "active_share": {"name": "活跃客户占比", "unit": "0-1", "direction": 1,
                     "formula": "期末active_friends/(期末active+cooling+dormant)",
                     "caliber": "试算口径:期末单日存量"},
}

# 阈值全部为试算起点(v2 §0 决议:全默认+试算标记);type 三种:
#   target      : floor..target 线性 0..100(方向已含在 target>floor)
#   reverse     : ≤target_ms 100 分,≥floor_ms 0 分,中间线性降
#   band        : 区间内 80..100,低于下限不评分(None),超上限不再加分
DEFAULT_SCORE_CONFIG: dict = {
    "version": "v1.0",
    "name": "一期试算(3维可用)",
    "normalization": "available_only_rescale",  # 可用维度 100/Σweight 再 ×100
    "trial": True,
    "sample_guards": {"min_threads": 30, "soft_threads": 10, "min_days": 5},
    "grade_bounds": {"A": 90, "B": 80, "C": 70, "D": 60},  # 低于 D 线为 E
    "dimensions": {
        "customer": {"weight": 30, "name": "客户经营", "metrics": {
            "active_share": {"type": "target", "floor": 0.20, "target": 0.60},
        }},
        "service": {"weight": 25, "name": "服务质量", "metrics": {
            "reply_2h_rate": {"type": "target", "floor": 0.40, "target": 0.85},
            "reply_24h_rate": {"type": "target", "floor": 0.75, "target": 0.97},
            "fr_p90_ms": {"type": "reverse", "target_ms": 1_800_000,
                          "floor_ms": 14_400_000},  # 30min..4h
        }},
        "action": {"weight": 20, "name": "行动执行",
                   "status": "pending_data"},  # 批次E 接 followups/任务
        "workload": {"weight": 15, "name": "有效负载", "metrics": {
            "active_friends_band": {"type": "band", "metric": "active_friends",
                                    "low": 5, "high": 80,
                                    "basis": "period_distinct"},
            "call_band": {"type": "band", "metric": "call_count",
                          "low": 1, "high": 200, "basis": "period_total",
                          "note": "通话低不扣分仅提示;band 得分下限 80"},
        }},
        "risk": {"weight": 10, "name": "合规风险",
                 "status": "pending_data"},  # 批次E/三期接风险事件
    },
}


def validate_config(config: dict) -> list[str]:
    """配置校验:权重总和/结构/引用的 metric_code 有元数据。返回问题清单。"""
    problems: list[str] = []
    dims = config.get("dimensions", {})
    total = sum(d.get("weight", 0) for d in dims.values())
    if total != 100:
        problems.append(f"维度权重总和须=100,当前 {total}")
    for key, dim in dims.items():
        if dim.get("status") == "pending_data":
            continue
        for code, spec in dim.get("metrics", {}).items():
            src = spec.get("metric", code)
            if src not in METRIC_DEFINITIONS:
                problems.append(f"{key}.{code} 引用未定义指标 {src}")
    return problems
  • Step 4: cli.py 加 score-config 子命令

parser（sensitive-scan 与 first-reply-facts 之间任意位置）：

    p_score = sub.add_parser(
        "score-config",
        help="Seed the default score config (idempotent).",
        description=(
            "Insert DEFAULT_SCORE_CONFIG (v1.0, trial) into "
            "employee_score_config as active. Skips if the version "
            "already exists. Batch B."
        ),
    )

命令函数与分发（照 _cmd_daily_stats 惯例）：

async def _seed_score_config(session) -> str | None:
    """种子默认评分配置;已存在同版本返回 None(幂等)。纯会话函数可测。"""
    from app.db.models import EmployeeScoreConfig
    from app.management.metric_definitions import (
        DEFAULT_SCORE_CONFIG,
        validate_config,
    )

    problems = validate_config(DEFAULT_SCORE_CONFIG)
    if problems:
        raise ValueError(f"score config invalid: {problems}")
    exists = (
        await session.execute(
            sa.select(EmployeeScoreConfig.id).where(
                EmployeeScoreConfig.version == DEFAULT_SCORE_CONFIG["version"]
            )
        )
    ).scalar_one_or_none()
    if exists is not None:
        return None
    session.add(
        EmployeeScoreConfig(
            version=DEFAULT_SCORE_CONFIG["version"],
            name=DEFAULT_SCORE_CONFIG["name"],
            status="active",
            config_json=DEFAULT_SCORE_CONFIG,
            created_by="seed-cli",
        )
    )
    await session.flush()
    return DEFAULT_SCORE_CONFIG["version"]


async def _cmd_score_config(args: argparse.Namespace) -> int:
    settings = get_settings()
    engine, session_maker = await _init_db(settings.db_url)
    try:
        async with session_maker() as session:
            version = await _seed_score_config(session)
            await session.commit()
        print(f"[score-config] {'seeded ' + version if version else 'already present, skipped'}")
        return 0
    finally:
        await engine.dispose()

（cli.py 头部补 import sqlalchemy as sa 若无。）

  • Step 5: 测试通过 + 提交
cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_management_scoring.py -v   # 5 passed
cd yunke-server && .venv/Scripts/python.exe -m app.sync.cli score-config   # 真机种子一次
git add yunke-server/app/management/ yunke-server/app/sync/cli.py yunke-server/tests/test_management_scoring.py
git commit -m "feat(management): 指标元数据+试算评分配置v1.0+种子CLI——权重校验/幂等落库"

Task B1: 期聚合查询服务

Files:

  • Create: yunke-server/app/management/employee_metrics.py
  • Test: yunke-server/tests/test_management_period_metrics.py

核心函数（完整实现）：

"""管理域期聚合查询(批次B)——评分/异常/employees API 的单一数据源。

窗口:days=[start_day..end_day] 闭区间 yyyymmdd;上期=等长紧邻前窗;
个人基线=紧邻当前窗之前的 28 天。所有查询走 (day, ...) 索引,不扫消息表
(除 distinct 活跃客户走 msg_daily_stats friend 行)。
"""
from __future__ import annotations

import datetime as dt
from collections import namedtuple
from zoneinfo import ZoneInfo

import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.models import (
    EmployeeCallDaily,
    EmployeeMetricDaily,
    FirstReplyFact,
    MsgDailyStats,
    Staff,
)
from app.storage.staff_summary import account_map
from app.storage.summary import day_bounds_ms

_SH = ZoneInfo("Asia/Shanghai")

PeriodMetrics = namedtuple("PeriodMetrics", [
    "user_id", "days_with_data", "msg_sent", "msg_recv", "cust_threads",
    "threads_answered", "fr_sum_ms", "fr_p50_ms", "fr_p90_ms",
    "reply_2h_count", "reply_24h_count",
    "active_friends",          # 期内 distinct(直查,非日值求和)
    "active_rooms_avg",        # 日均
    "new_customer", "cooling_customer", "dormant_customer",  # 期末单日存量
    "reactivated_customer",    # 期内求和
    "call_count", "call_seconds", "answered_call", "call_followup",
])


def _day_range(start: int, end: int) -> list[int]:
    """闭区间日键列表(纯 Python,上海自然日语义由调用方保证)。"""
    def _d(k): return dt.date(k // 10000, k // 100 % 100, k % 100)
    def _k(d): return d.year * 10000 + d.month * 100 + d.day
    out, cur = [], _d(start)
    while cur <= _d(end):
        out.append(_k(cur))
        cur += dt.timedelta(days=1)
    return out


def prev_window(start: int, end: int) -> tuple[int, int]:
    """等长紧邻前窗。"""
    n = len(_day_range(start, end))
    def _d(k): return dt.date(k // 10000, k // 100 % 100, k % 100)
    def _k(d): return d.year * 10000 + d.month * 100 + d.day
    pe = _d(start) - dt.timedelta(days=1)
    ps = pe - dt.timedelta(days=n - 1)
    return _k(ps), _k(pe)


def baseline_window(start: int) -> tuple[int, int]:
    """个人基线:紧邻当前窗之前的 28 天。"""
    def _d(k): return dt.date(k // 10000, k // 100 % 100, k % 100)
    def _k(d): return d.year * 10000 + d.month * 100 + d.day
    be = _d(start) - dt.timedelta(days=1)
    bs = be - dt.timedelta(days=27)
    return _k(bs), _k(be)

period_metrics(session, start, end, channel, user_ids=None) -> dict[str, PeriodMetrics] 实现要点（写全代码进计划）：

  1. 宽表 SUM（WHERE day BETWEEN s AND e AND channel=? AND metric_version='v1' AND user_id IN (...) GROUP BY user_id）：msg_sent/msg_recv/cust_threads/threads_answered/fr_sum_ms/active_rooms（SUM→avg=SUM/天数）、reactivated（SUM）、days_with_data=COUNT(*)；期末存量：另查 day=(SELECT MAX(day)≤e 同 user/channel) 的 new/cooling/dormant 行（ correlated 或两段查询——用 sa.select(...).where(day<=end).order_by(day.desc()) per user 取首行，员工 41 人循环可接受；或一次查询取 (user, MAX(day)) 再 join）。
  2. fact 分位与回复计数（单条 SQL，窗口函数，双方言）：
sub = (
    sa.select(
        FirstReplyFact.owner_account_id.label("owner"),
        FirstReplyFact.fr_ms.label("fr"),
        sa.func.row_number().over(
            partition_by=FirstReplyFact.owner_account_id,   # 列表达式,非别名——MySQL 窗口规范不认 SELECT 别名
            order_by=FirstReplyFact.fr_ms,
        ).label("rn"),
        sa.func.count().over(
            partition_by=FirstReplyFact.owner_account_id
        ).label("n"),
    ).where(FirstReplyFact.day.between(start, end),
            FirstReplyFact.channel == channel)
    .subquery()
)
# 每 owner:rn=ceil(0.5*n) 行的 fr 即 p50;rn=ceil(0.9*n) 即 p90;
# 回复计数:SUM(fr<=2h)/SUM(fr<=24h) 另用条件聚合直查(不用窗口)。

owner→user 经 account_map(channel)。p50/p90 取最接近秩（rn == (n+1)//2 与 rn == max(1, (9*n+9)//10)——用「最接近秩」公式并在测试里锁死：ceil(q·n) 最近邻，与批次A _percentile 口径一致）。 3. reply 计数：SELECT owner, SUM(fr_ms<=7200000), SUM(fr_ms<=86400000) FROM first_reply_fact WHERE day BETWEEN ... GROUP BY owner（MySQL 布尔求和；SQLite 同义 SUM(CASE WHEN ...)——用 case 写法双方言安全）。 4. distinct 活跃客户：msg_daily_stats friend 行 in window → peer 集合按 account_map 归并到 user → len。 5. 通话：employee_call_daily SUM（无渠道维；channel 参数只影响消息侧——通话四列在 channel 查询里也返回全渠道值，响应里标注）。 6. team_median：对每指标取全体 user 的中位数（scoring 层用）。 7. channel=all：router 层把 "all" 翻译成 None 传入（照 analytics 惯例，哨兵字符串不进 SQL）；channel=None 时两条渠道的 account_map 合并、宽表/fact 查询去渠道过滤、分位与活跃跨渠道合并到同一 user。

测试（seed 宽表/fact/通话行 + Staff/account，构造已知分位数集合断言 p50/p90 精确值、期末存量取 ≤e 最近行、distinct 活跃跨日去重、prev/baseline 窗口日期算术含月末）——完整测试代码进计划（约 6 个用例）。

  • Steps: 测试先失败 → 实现 → 通过 → 全量回归 → 提交 feat(management): 期聚合查询——fact窗口函数精确分位/期末生命周期存量/distinct活跃/前窗基线窗口

Task B2: 评分引擎 scoring.py

Files:

  • Create: yunke-server/app/management/scoring.py
  • Test: 追加 tests/test_management_scoring.py

核心（完整代码进计划）：

def normalize(spec: dict, value: float | None) -> float | None:
    """单指标 0..100;样本不足/数据缺 → None(不评分,不默认 0/100)。"""
    t = spec["type"]
    if value is None:
        return None
    if t == "target":
        if value <= spec["floor"]:
            return 0.0
        if value >= spec["target"]:
            return 100.0
        return (value - spec["floor"]) / (spec["target"] - spec["floor"]) * 100
    if t == "reverse":
        if value <= spec["target_ms"]:
            return 100.0
        if value >= spec["floor_ms"]:
            return 0.0
        return (spec["floor_ms"] - value) / (spec["floor_ms"] - spec["target_ms"]) * 100
    if t == "band":
        if value < spec["low"]:
            return None  # 负载不足→不评分(样本保护语义)
        return 100.0 if value <= spec["high"] else 80.0  # 超上限 80 不再加
    raise ValueError(t)


def score_employee(config, current: PeriodMetrics, baseline: PeriodMetrics | None,
                   team: dict[str, float]) -> dict:
    """返回 {total_score, grade, score_status, trial, dimensions{...},
    metrics{code: {value, prev, baseline, median, sample_size,
    sample_status, score, formula_note}}}。"""

规则（写死进测试）：

  • 周期值派生：reply_2h_rate = reply_2h_count/cust_threads（threads=0 → None）；fr_p90_ms 直取；active_share 分子=期末该渠道最近一行(≤end)的 active_friends，分母=同一行的 active_friends+cooling_customer+dormant_customer（分母 0 → None；宽表无生命周期 active 存量列，active 以 active_friends 列充当）；band 用 active_friends 周期 distinct（B2 决议:抗单日噪声,与 B1 口径一致）与 call_count 总量。
  • 样本保护：cust_threads < min_threads(30) → 该 service/customer 指标 sample_status="insufficient"，不评分；soft(10)~min 之间 "low_confidence"（评分但标注）。
  • 维度分 = 各指标得分的均值（等权）；任一指标 None 时该指标剔除并按剩余归一；全部 None → 维度 pending。
  • 总分 = Σ(可用维度 weight×dim_score)/Σ(可用维度 weight)×100；可用=pending_data 以外且维度分非 None；全 None → score_status="insufficient_sample"（total None）。
  • grade 按 grade_bounds；insufficient → grade None + status。
  • trial=True 恒返（一期 3 维）。

测试（≥8 用例，完整代码进计划）：target 边界（floor/target/中点 50）、reverse 两端与中间、band 三段、threads=0 → None 非 0、insufficient_sample 整员工、pending 维度不参与（100/70 归一后 A 线仍可达）、试算标记、全部指标有明细含四对比位。

  • Steps: TDD → 提交 feat(management): 评分引擎——三类标准化/维度归一/样本保护/等级/完整明细

Task B3: 异常规则 anomalies.py

Files:

  • Create: yunke-server/app/management/anomalies.py
  • Test: yunke-server/tests/test_management_anomalies.py

4 条确定性规则（输入=period_metrics 当前+基线+团队中位数；输出=v1 §5.4 卡片列表，完整代码进计划）：

rule_code 触发 severity drilldown
reply_p90_worsened p90 > 基线p90×1.5 且样本≥soft；×2.5 升 critical warning/critical metric_code=first_reply_p90
reply_rate_dropped reply_2h_rate < 基线×0.7 且当前≥soft 样本 warning metric_code=unreplied
activity_drop 日均 msg_total < 基线日均×0.5 且基线≥min_days warning 员工详情
workload_overload active_friends 周期 distinct > band 上限 80 warning 员工详情

卡片字段照 v1 §5.4 JSON（rule_code/severity/employee_user_id/title/current_value/baseline_value/unit/sample_size/affected_count/reason[纯数据事实]/drilldown/suggested_action——文案固定模板，不用 LLM）。affected_count 按规则定义：reply_p90_worsened 与 reply_rate_dropped=期内未回串数（cust_threads−threads_answered）；activity_drop=当前期活跃客户数；workload_overload=active_friends 日均。

测试（≥6 用例）：每规则触发/不触发边界、基线 None 不触发、severity 升级阈值、卡片结构字段齐全。

  • Steps: TDD → 提交 feat(management): 确定性异常规则×4——P90恶化/回复率骤降/活动骤降/过载

Task B4: 管理 API 6 端点

Files:

  • Create: yunke-server/app/api/routers/management.py
  • Modify: yunke-server/app/api/main.py（import + include_router(management.router)）
  • Test: yunke-server/tests/test_management_api.py

路由骨架（完整代码进计划，照 analytics 惯例）：

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)]
端点 行为要点
GET /dashboard 参数 channel/start/end/compare(previous|personal_baseline|team_median,默认 previous)。组装：kpis（活跃客户/活跃员工/待回复[复用 analytics unreplied total]/通话量/生命周期期末分布）+ communication_trend（宽表按日 SUM msg+distinct friends 近 trend 天）+ service_trend（fr 日值 p50/p90 宽表列按日 + reply_2h_rate 派生）+ lifecycle（期末存量分布）+ dimension_summary（全员 3 维分布计数）+ anomalies（B3 全员）+ action_summary（复用 analytics action_counts）+ data_as_of（复用 health）。单次响应聚合，内部并发可用 asyncio.gather
GET /employees 参数 channel/start/end/view/keyword/status/sort_by/sort_order/page/page_size(≤100 默认 20)。行=staff 身份(name/is_on_job) + PeriodMetrics + 四对比 + score_employee 结果摘要(total/grade/sample_status/trial) + anomaly_count + anomaly_weight。默认排序=anomaly_severity（权重和 critical=2/warning=1 desc，同分再比 count desc——对应 v2 §6「按异常严重度排序」，非单纯计数）；keyword 匹配 staff.name/user_id
GET /employees/{user_id} 404 校验（staff 无此 user_id）；头部身份+账号清单(wechat/whatsapp account 表)+宽表总览+评分明细+异常+基线摘要
GET /employees/{user_id}/metrics?section= section ∈ overview|service|customer|workload|calls（一期 5 个，v2 §6）；每 section 返回该域 PeriodMetrics 明细+四对比+trend（宽表按日序列）
GET /employees/{user_id}/drilldown?metric_code=&start=&end=&channel=&page=&page_size= 白名单：`first_reply_p90
GET /metric-definitions METRIC_DEFINITIONS + DEFAULT_SCORE_CONFIG 摘要（权重/试算标记/样本门槛）+ 三方口径差异说明

API 测试（完整代码进计划，make_client fixture，≥12 用例）：参数默认值/422（custom 缺参、跨度>366、metric_code 白名单外、page_size>100）/404（员工不存在）/响应结构关键字段（data_as_of、metric_version、sample_status、trial 标记）/employees 分页排序/dashboard 各段齐全/渠道过滤生效。

  • Steps: TDD → main.py 注册 → OpenAPI 冒烟（/docs 可见 6 端点）→ 全量回归 → 提交 feat(management): 只读管理API六端点——dashboard/employees/detail/sections/drilldown白名单/指标定义

Task B5: 真机冒烟 + 性能初测 + 记录

  • 本地起后端（先 docker stop yunke-backend-1 yunke-frontend-1 若在跑；cd yunke-server && .venv/Scripts/python.exe -m uvicorn app.api.main:app --port 8000 后台）；
  • curl 六端点（带 token，从 .env API_TOKEN；无 token 模式则直接）真机各一次，记录响应体积与耗时（目标：dashboard p95≤1.5s、employees≤1.5s——v2 §9 性能线）；
  • 抽查：某回测员工（王云青 369D257E…）30 天窗 employees 行——P90/回复率/维度分人工合理性检查（对照核实文档对账数字）；
  • 结果（耗时表+抽查截图级数字）追加到批次A核实文档「批次B冒烟」小节；
  • 提交 docs(management): 批次B真机冒烟——六端点耗时/回测员工抽查/记录

批次B验收清单

  • score-config 种子幂等，配置过校验，v1.0 active；
  • 期聚合：p50/p90 精确分位（窗口函数）与期末存量口径有判别测试；
  • 评分：三类标准化边界/样本保护/100-70 归一/试算标记全锁测试；
  • 异常：4 规则触发边界 + 卡片结构完整；
  • 6 端点：OpenAPI 可见、参数校验 422/404 齐全、白名单生效、dict 返回含 data_as_of/metric_version/sample_status；
  • 全量测试绿（653 + 新增 ~30）；
  • 真机冒烟耗时达标并记录；
  • 未动批次A任何存储层代码（宽表/fact 生成零改动）。

批次C-F

批次C：前端（api/management.ts + useManagementFilters + components/management/* + ManagementDashboardView + 路由/导航「经营管理」分组）——批次B合入后出计划，消费真实 API 形态。D：员工总览/详情页。E：followups+快照（写端点+schemas）。F：调度接入（analytics_tick 第三步/full_sync 13-14 步/scheduler 镜像重建）+回测试运行。

Reference in New Issue View Git Blame Copy Permalink