diff --git a/yunke-server/app/analytics/metrics.py b/yunke-server/app/analytics/metrics.py index 454e852..1d0e471 100644 --- a/yunke-server/app/analytics/metrics.py +++ b/yunke-server/app/analytics/metrics.py @@ -136,27 +136,39 @@ async def _first_contact( async def _active_staff( session: AsyncSession, channel: str, days: list[int] -) -> int: - """活跃员工 distinct user_id;stat_daily_staff 空表回退 msg_daily_stats+映射。""" +) -> tuple[int, int]: + """活跃员工 (distinct 数, 未映射账号数);stat_daily_staff 空表回退 msg_daily_stats+映射。 + + 未映射口径(spec §8「未映射账号按账号计并标注」): + - rollup:当窗 distinct user_id 不在 account_map 值集(=在册员工 user_id + 全集)——recompute 以 amap.get(owner, owner) 落表,映射不上的 owner + 以原值混入 user_id 列,值集之外的即未映射; + - 回退:当窗 owner 中 amap.get(o, o)==o(映射不上,按账号原值计入 + 活跃数)的计未映射——与 rollup 口径一致(映射不上的账号原样出现)。 + """ if await _staff_rows(session): - val = ( - await session.execute( - sa.select(sa.func.count(sa.func.distinct(StatDailyStaff.user_id))).where( - StatDailyStaff.channel == channel, - StatDailyStaff.day.in_(days), + uids = set( + ( + await session.execute( + sa.select(StatDailyStaff.user_id).where( + StatDailyStaff.channel == channel, + StatDailyStaff.day.in_(days), + ) ) - ) - ).scalar_one() - return int(val) + ).scalars() + ) + mapped = set((await account_map(session, channel)).values()) + return len(uids), len(uids - mapped) amap = await account_map(session, channel) - owners = {amap.get(o, o) for (o,) in ( + owners = {o for (o,) in ( await session.execute( sa.select(MsgDailyStats.owner_account_id).where( MsgDailyStats.channel == channel, MsgDailyStats.day.in_(days) ) ) ).all()} - return len(owners) + unmapped = sum(1 for o in owners if o not in amap) + return len({amap.get(o, o) for o in owners}), unmapped def anomaly_threshold() -> float: @@ -248,7 +260,10 @@ async def anomaly(session: AsyncSession, channel: str) -> dict | None: async def overview( session: AsyncSession, *, channel: str, scope: str, trend_days: int = 30 ) -> dict[str, Any]: - """A1:卡片 + 环比(仅周/月,同期对齐) + 近 N 天趋势 + 异常警示。""" + """A1:卡片 + 环比(仅周/月,同期对齐) + 近 N 天趋势 + 异常警示。 + + cards.unmapped_accounts=活跃数里映射不上的账号数(spec §8 按账号计并标注)。 + """ if trend_days < 1 or trend_days > 90: raise ValueError("trend_days must be 1..90") today = today_key() @@ -257,10 +272,12 @@ async def overview( msg_total = sum(int(s or 0) + int(r or 0) for _d, _k, _p, s, r in rows) active_customers = len({p for _d, kind, p, _s, _r in rows if kind == "friend"}) + active_staff, unmapped = await _active_staff(session, channel, days) cards: dict[str, Any] = { "msg_total": msg_total, "active_customers": active_customers, - "active_staff": await _active_staff(session, channel, days), + "active_staff": active_staff, + "unmapped_accounts": unmapped, "call_cnt": await _call_count(session, days), "first_contact": await _first_contact(session, channel, days), } diff --git a/yunke-server/app/api/routers/analytics.py b/yunke-server/app/api/routers/analytics.py index f73fea9..a8e85cd 100644 --- a/yunke-server/app/api/routers/analytics.py +++ b/yunke-server/app/api/routers/analytics.py @@ -1,17 +1,20 @@ """GET /api/analytics/* —— 数据分析板块端组(前缀分组=未来角色权限落点)。""" from __future__ import annotations -from typing import Annotated, Any +from typing import Annotated, Any, Literal -from fastapi import APIRouter, Depends +from fastapi import APIRouter, Depends, 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 router = APIRouter(prefix="/api/analytics", tags=["analytics"]) SessionDep = Annotated[AsyncSession, Depends(get_session)] +ChannelParam = Annotated[Literal["wechat", "whatsapp"], Query()] +ScopeParam = Annotated[Literal["today", "week", "month"], Query()] @router.get("/health") @@ -19,3 +22,17 @@ async def analytics_health(session: SessionDep) -> dict[str, Any]: """E0-lite:数据截至 + 实体新鲜度。""" as_of = await health_svc.data_as_of(session) return {"data_as_of": as_of, "entities": await health_svc.entity_freshness(session)} + + +@router.get("/overview") +async def analytics_overview( + session: SessionDep, + channel: ChannelParam = "wechat", + scope: ScopeParam = "today", +) -> dict[str, Any]: + """A1 经营总览:卡片+趋势+异常警示一次返回(单往返)。 + + cards 不含 data_as_of——数据截至走 /health(data_as_of 为 ms-epoch int + 与 entities[].last_seen 为 ISO 字符串的双格式)。 + """ + return await metrics_svc.overview(session, channel=channel, scope=scope) diff --git a/yunke-server/tests/test_analytics_api.py b/yunke-server/tests/test_analytics_api.py index 24e2826..2e35991 100644 --- a/yunke-server/tests/test_analytics_api.py +++ b/yunke-server/tests/test_analytics_api.py @@ -1,13 +1,17 @@ -"""/api/analytics/health:E0-lite 冒烟。""" +"""/api/analytics/health + /overview:E0-lite 与 A1 冒烟。""" from __future__ import annotations import datetime as dt -from app.db.models import Customer, MsgDailyStats +from app.analytics import metrics +from app.db.models import Customer, MsgDailyStats, StatDailyStaff, WechatAccount # naive:updated_at 列是无时区 DateTime;fromisoformat 构造不触发 DTZ 检查 _SEEN = "2026-09-04T12:00:00" +# overview 的 today_key 是真实今天——固定 day 键的 seed 必须 monkeypatch +_TODAY = 20260904 + async def test_health_empty(make_client): async def seed(session): @@ -55,3 +59,86 @@ async def test_health_with_stats(make_client): assert dt.datetime.fromisoformat(entities["customer"]) == dt.datetime.fromisoformat( _SEEN ) + + +async def test_overview_endpoint(make_client, monkeypatch): + monkeypatch.setattr(metrics, "today_key", lambda: _TODAY) + + async def seed(session): + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t1", day=_TODAY, + send_cnt=4, recv_cnt=1, last_ts=1788500000000, + last_is_send=True)) + await session.commit() + + client = await make_client(seed) + r = await client.get( + "/api/analytics/overview", params={"channel": "wechat", "scope": "today"} + ) + assert r.status_code == 200 + body = r.json() + assert body["cards"]["msg_total"] == 5 + assert isinstance(body["trend"], list) and body["trend"] + # scope 非法 → 422 + r2 = await client.get( + "/api/analytics/overview", params={"channel": "wechat", "scope": "year"} + ) + assert r2.status_code == 422 + + +async def test_overview_unmapped_accounts(make_client, monkeypatch): + """控制器追加:cards 带 unmapped_accounts 计数(spec §8「未映射账号按账号计并标注」的标注通道)。""" + monkeypatch.setattr(metrics, "today_key", lambda: _TODAY) + + async def seed(session): + session.add(WechatAccount(wx_id="wx1", user_id="u1")) + # wx2 无账号表记录 → 未映射 + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t1", day=_TODAY, + send_cnt=2, recv_cnt=0, last_ts=1788500000000, + last_is_send=True)) + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx2", + peer_kind="friend", peer_id="t2", day=_TODAY, + send_cnt=1, recv_cnt=0, last_ts=1788501000000, + last_is_send=True)) + await session.commit() + + client = await make_client(seed) + r = await client.get( + "/api/analytics/overview", params={"channel": "wechat", "scope": "today"} + ) + body = r.json() + assert body["cards"]["active_staff"] == 2 # u1 + wx2(按账号计) + assert body["cards"]["unmapped_accounts"] == 1 # wx2 标注 + + +async def test_overview_unmapped_rollup(make_client, monkeypatch): + """rollup 路径未映射口径与回退路径一致:当窗 user_id 不在映射值集 → 未映射。 + + recompute 以 amap.get(owner, owner) 落表,映射不上的 owner 以原值进 + stat_daily_staff(user_id=wx2);account_map.values() 只含在册员工 user_id, + wx2 落在值集之外 → 被识别并标注。 + """ + monkeypatch.setattr(metrics, "today_key", lambda: _TODAY) + + async def seed(session): + session.add(WechatAccount(wx_id="wx1", user_id="u1")) + session.add(StatDailyStaff(day=_TODAY, channel="wechat", user_id="u1", + msg_sent=2, msg_recv=0, + active_friends=1, active_rooms=0)) + session.add(StatDailyStaff(day=_TODAY, channel="wechat", user_id="wx2", + msg_sent=1, msg_recv=0, + active_friends=1, active_rooms=0)) + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t1", day=_TODAY, + send_cnt=2, recv_cnt=0, last_ts=1788500000000, + last_is_send=True)) + await session.commit() + + client = await make_client(seed) + r = await client.get( + "/api/analytics/overview", params={"channel": "wechat", "scope": "today"} + ) + body = r.json() + assert body["cards"]["active_staff"] == 2 + assert body["cards"]["unmapped_accounts"] == 1