feat(analytics): A3 流失预警三分法——真流失/从未激活/待交接

Co-Authored-By: Claude Code <noreply@anthropic.com>

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}

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,
+    )

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