From 3b4254639d2ed5f6284731200b5e311779ed645d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Fri, 4 Sep 2026 18:25:08 +0800 Subject: [PATCH] =?UTF-8?q?feat(analytics):=20A3=20=E6=B5=81=E5=A4=B1?= =?UTF-8?q?=E9=A2=84=E8=AD=A6=E4=B8=89=E5=88=86=E6=B3=95=E2=80=94=E2=80=94?= =?UTF-8?q?=E7=9C=9F=E6=B5=81=E5=A4=B1/=E4=BB=8E=E6=9C=AA=E6=BF=80?= =?UTF-8?q?=E6=B4=BB/=E5=BE=85=E4=BA=A4=E6=8E=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- yunke-server/app/analytics/metrics.py | 106 +++++++++++++++++- yunke-server/app/api/routers/analytics.py | 21 +++- .../tests/test_analytics_churn_unreplied.py | 79 +++++++++++++ 3 files changed, 204 insertions(+), 2 deletions(-) create mode 100644 yunke-server/tests/test_analytics_churn_unreplied.py diff --git a/yunke-server/app/analytics/metrics.py b/yunke-server/app/analytics/metrics.py index 1d0e471..a7c4c80 100644 --- a/yunke-server/app/analytics/metrics.py +++ b/yunke-server/app/analytics/metrics.py @@ -14,7 +14,7 @@ from zoneinfo import ZoneInfo import sqlalchemy as sa from sqlalchemy.ext.asyncio import AsyncSession -from app.db.models import MsgDailyStats, PhoneCall, StatDailyStaff +from app.db.models import Customer, MsgDailyStats, PhoneCall, Staff, StatDailyStaff from app.storage.staff_summary import account_map from app.storage.staff_summary import has_rows as _staff_rows from app.storage.summary import day_bounds_ms @@ -335,3 +335,107 @@ async def overview( "trend": trend, "anomaly": await anomaly(session, channel), } + + +def _now_ms() -> int: + return int(dt.datetime.now(_SH).timestamp() * 1000) + + +def _next_tier(tier: int) -> int: + return {30: 60, 60: 90, 90: 10_000}[tier] + + +async def churn( + session: AsyncSession, *, channel: str, kind: str, tier: int, + min_msgs: int, page: int, page_size: int, +) -> dict: + """A3 三分法(spec §3.3)。tier ∈ {30,60,90} 分层区间 [tier, next) ,90=≥90。""" + day_ms = 86_400_000 + now = _now_ms() + base = [ + Customer.channel == channel, + sa.or_(Customer.delete_flag.is_(None), Customer.delete_flag == 0), + ] + if kind == "real": + lo = now - _next_tier(tier) * day_ms + hi = now - tier * day_ms + # last_chat_time ∈ (lo, hi]:沉默 tier..next_tier 天 + base += [Customer.last_chat_time.is_not(None), + Customer.last_chat_time > lo, + Customer.last_chat_time <= hi] + # 历史消息 ≥ min_msgs 放 SQL 端(分页 total 语义正确):半连接聚合 + base.append( + Customer.friend_id.in_( + sa.select(MsgDailyStats.peer_id) + .where( + MsgDailyStats.channel == channel, + MsgDailyStats.peer_kind == "friend", + ) + .group_by(MsgDailyStats.peer_id) + .having( + sa.func.coalesce( + sa.func.sum(MsgDailyStats.send_cnt + MsgDailyStats.recv_cnt), 0 + ) >= min_msgs + ) + ) + ) + order = Customer.last_chat_time.asc() + elif kind == "inactive": + base += [Customer.last_chat_time.is_(None)] + order = Customer.friend_id.asc() + elif kind == "handover": + base += [Customer.user_id.in_( + sa.select(Staff.user_id).where(Staff.is_on_job.is_(False)) + )] + # MySQL 由 compat.py 编译补丁剥 NULLS LAST(DESC 下语义天然等价) + order = Customer.last_chat_time.desc().nulls_last() + else: + raise ValueError(f"bad kind: {kind!r}") + + total = ( + await session.execute( + sa.select(sa.func.count()).select_from(Customer).where(*base) + ) + ).scalar_one() + rows = ( + await session.execute( + sa.select(Customer).where(*base).order_by(order) + .offset((page - 1) * page_size).limit(page_size) + ) + ).scalars().all() + + # enrich:历史消息量/活跃天数(msg_daily_stats 按 peer 聚合) + items = [] + if rows: + ids = [c.friend_id for c in rows] + agg = ( + await session.execute( + sa.select( + MsgDailyStats.peer_id, + sa.func.coalesce( + sa.func.sum(MsgDailyStats.send_cnt + MsgDailyStats.recv_cnt), 0 + ), + sa.func.count(), + ) + .where( + MsgDailyStats.channel == channel, + MsgDailyStats.peer_kind == "friend", + MsgDailyStats.peer_id.in_(ids), + ) + .group_by(MsgDailyStats.peer_id) + ) + ).all() + hist = {p: (int(m), int(d)) for p, m, d in agg} + for c in rows: + m, d = hist.get(c.friend_id, (0, 0)) + items.append({ + "friend_id": c.friend_id, + "channel": c.channel, + "nickname": c.friend_nickname or c.friend_remark or c.friend_id, + "owner_account_id": c.owner_account_id, + "user_id": c.user_id, + "last_chat_time": c.last_chat_time, + "hist_msgs": m, + "active_days": d, + }) + return {"items": items, "total": int(total), "page": page, "page_size": page_size} diff --git a/yunke-server/app/api/routers/analytics.py b/yunke-server/app/api/routers/analytics.py index a8e85cd..0efc6aa 100644 --- a/yunke-server/app/api/routers/analytics.py +++ b/yunke-server/app/api/routers/analytics.py @@ -3,7 +3,7 @@ from __future__ import annotations from typing import Annotated, Any, Literal -from fastapi import APIRouter, Depends, Query +from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.ext.asyncio import AsyncSession from app.analytics import health as health_svc @@ -36,3 +36,22 @@ async def analytics_overview( 与 entities[].last_seen 为 ISO 字符串的双格式)。 """ return await metrics_svc.overview(session, channel=channel, scope=scope) + + +@router.get("/churn") +async def churn_list( + session: SessionDep, + channel: ChannelParam = "wechat", + type: Literal["real", "inactive", "handover"] = "real", + tier: int = Query(30, ge=30, le=90), + min_msgs: int = Query(5, ge=1, le=100), + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), +) -> dict[str, Any]: + """A3 流失预警名单(三分法)。tier 只认 30/60/90。""" + if tier not in (30, 60, 90): + raise HTTPException(422, "tier must be 30/60/90") + return await metrics_svc.churn( + session, channel=channel, kind=type, tier=tier, + min_msgs=min_msgs, page=page, page_size=page_size, + ) diff --git a/yunke-server/tests/test_analytics_churn_unreplied.py b/yunke-server/tests/test_analytics_churn_unreplied.py new file mode 100644 index 0000000..75897dc --- /dev/null +++ b/yunke-server/tests/test_analytics_churn_unreplied.py @@ -0,0 +1,79 @@ +"""A3 三分法流失 + B2 该回未回。""" +from __future__ import annotations + +from app.analytics import metrics +from app.db.models import Customer, MsgDailyStats, Staff + +_NOW = 1756950000000 # 相对时间基准(绝对日期无关紧要,断言全部相对此值自洽) + + +def _cust(friend_id, *, last_chat=None, user_id="u1", delete=0, block=None): + return Customer(friend_id=friend_id, channel="wechat", + owner_account_id="wx1", user_id=user_id, + last_chat_time=last_chat, delete_flag=delete, is_block=block) + + +async def _seed_churn(session): + now = _NOW + day_ms = 86_400_000 + session.add(Staff(user_id="u9", name="离职的", is_on_job=False)) + session.add(Staff(user_id="u1", name="在职的", is_on_job=True)) + # 真流失:45 天没聊 + 历史消息 ≥5 + session.add(_cust("c_real", last_chat=now - 45 * day_ms)) + # 历史消息不足 5 → 不算真流失 + session.add(_cust("c_few", last_chat=now - 45 * day_ms)) + # 只沉默 10 天 → 不入 30 档 + session.add(_cust("c_recent", last_chat=now - 10 * day_ms)) + # 从未激活 + session.add(_cust("c_null", last_chat=None)) + # 待交接:归属离职员工 + session.add(_cust("c_hand", user_id="u9", last_chat=now - 3 * day_ms)) + # 已删好友 → 全部剔除 + session.add(_cust("c_del", last_chat=now - 45 * day_ms, delete=1)) + # 历史消息量:real=6(≥5),few=2 + for peer, n in (("c_real", 6), ("c_few", 2)): + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id=peer, day=20260601, + send_cnt=n, recv_cnt=0, last_ts=now - 50 * day_ms, + last_is_send=True)) + await session.commit() + + +async def test_churn_real_tier30(make_client, monkeypatch): + monkeypatch.setattr(metrics, "_now_ms", lambda: _NOW) + async def seed(session): + await _seed_churn(session) + page = await metrics.churn(session, channel="wechat", kind="real", + tier=30, min_msgs=5, page=1, page_size=20) + ids = [r["friend_id"] for r in page["items"]] + assert "c_real" in ids and "c_few" not in ids + assert "c_recent" not in ids and "c_del" not in ids + real = next(r for r in page["items"] if r["friend_id"] == "c_real") + assert real["hist_msgs"] == 6 and real["active_days"] == 1 + + await make_client(seed) + + +async def test_churn_inactive_and_handover(make_client, monkeypatch): + monkeypatch.setattr(metrics, "_now_ms", lambda: _NOW) + async def seed(session): + await _seed_churn(session) + p1 = await metrics.churn(session, channel="wechat", kind="inactive", + tier=30, min_msgs=5, page=1, page_size=20) + assert {r["friend_id"] for r in p1["items"]} == {"c_null"} + p2 = await metrics.churn(session, channel="wechat", kind="handover", + tier=30, min_msgs=5, page=1, page_size=20) + assert {r["friend_id"] for r in p2["items"]} == {"c_hand"} + + await make_client(seed) + + +async def test_churn_endpoint_422(make_client): + async def seed(session): + ... + + client = await make_client(seed) + r = await client.get("/api/analytics/churn", params={"tier": 45}) + assert r.status_code == 422 + r2 = await client.get("/api/analytics/churn", params={"channel": "sms"}) + assert r2.status_code == 422