diff --git a/yunke-server/app/analytics/metrics.py b/yunke-server/app/analytics/metrics.py index a7c4c80..85ebbcf 100644 --- a/yunke-server/app/analytics/metrics.py +++ b/yunke-server/app/analytics/metrics.py @@ -342,14 +342,20 @@ def _now_ms() -> int: def _next_tier(tier: int) -> int: - return {30: 60, 60: 90, 90: 10_000}[tier] + try: + return {30: 60, 60: 90, 90: 10_000}[tier] + except KeyError: + raise ValueError(f"bad tier: {tier!r}") from None 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。""" + """A3 三分法(spec §3.3)。tier ∈ {30,60,90} 分层区间 [tier, next) ,90=≥90。 + + active_days=历史活跃天数(distinct 自然日;peer 跨账号转移同日多行只计一日)。 + """ day_ms = 86_400_000 now = _now_ms() base = [ @@ -379,16 +385,19 @@ async def churn( ) ) ) - order = Customer.last_chat_time.asc() + # 决胜列 friend_id:count/分页两查询间同 last_chat_time 组内顺序稳定 + order = [Customer.last_chat_time.asc(), Customer.friend_id.asc()] elif kind == "inactive": base += [Customer.last_chat_time.is_(None)] - order = Customer.friend_id.asc() + 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() + # MySQL 由 compat.py 编译补丁剥 NULLS LAST(DESC 下语义天然等价); + # NULL 尾部密集(离职员工的未激活客户),必须带 friend_id 决胜列防翻页重复 + order = [Customer.last_chat_time.desc().nulls_last(), + Customer.friend_id.asc()] else: raise ValueError(f"bad kind: {kind!r}") @@ -399,12 +408,13 @@ async def churn( ).scalar_one() rows = ( await session.execute( - sa.select(Customer).where(*base).order_by(order) + sa.select(Customer).where(*base).order_by(*order) .offset((page - 1) * page_size).limit(page_size) ) ).scalars().all() - # enrich:历史消息量/活跃天数(msg_daily_stats 按 peer 聚合) + # enrich:历史消息量/活跃天数(msg_daily_stats 按 peer 聚合; + # 活跃天数按 distinct 自然日计——peer 跨账号同日多行不重复计) items = [] if rows: ids = [c.friend_id for c in rows] @@ -415,7 +425,7 @@ async def churn( sa.func.coalesce( sa.func.sum(MsgDailyStats.send_cnt + MsgDailyStats.recv_cnt), 0 ), - sa.func.count(), + sa.func.count(sa.distinct(MsgDailyStats.day)), ) .where( MsgDailyStats.channel == channel, diff --git a/yunke-server/tests/test_analytics_churn_unreplied.py b/yunke-server/tests/test_analytics_churn_unreplied.py index 75897dc..8e1814c 100644 --- a/yunke-server/tests/test_analytics_churn_unreplied.py +++ b/yunke-server/tests/test_analytics_churn_unreplied.py @@ -1,6 +1,8 @@ """A3 三分法流失 + B2 该回未回。""" from __future__ import annotations +import pytest + from app.analytics import metrics from app.db.models import Customer, MsgDailyStats, Staff @@ -77,3 +79,58 @@ async def test_churn_endpoint_422(make_client): assert r.status_code == 422 r2 = await client.get("/api/analytics/churn", params={"channel": "sms"}) assert r2.status_code == 422 + + +async def test_churn_tier_membership_and_min_msgs_boundary(make_client, monkeypatch): + """tier60/90 会员资格([tier, next) 区间)+ min_msgs 恰等阈值的包含边界。""" + monkeypatch.setattr(metrics, "_now_ms", lambda: _NOW) + day_ms = 86_400_000 + + async def seed(session): + # 沉默 75 天:∈[60,90) → tier60 含、tier30 不含 + session.add(_cust("c75", last_chat=_NOW - 75 * day_ms)) + # 沉默 100 天:≥90 → tier90 含、tier60 不含 + session.add(_cust("c100", last_chat=_NOW - 100 * day_ms)) + # 历史消息恰好 5 条(=min_msgs,含边界)+ 沉默 45 天 → 入 tier30 + session.add(_cust("c5", last_chat=_NOW - 45 * day_ms)) + for peer, n in (("c75", 6), ("c100", 6), ("c5", 5)): + 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 - 110 * day_ms, + last_is_send=True)) + await session.commit() + + t30 = await metrics.churn(session, channel="wechat", kind="real", + tier=30, min_msgs=5, page=1, page_size=20) + assert {r["friend_id"] for r in t30["items"]} == {"c5"} + t60 = await metrics.churn(session, channel="wechat", kind="real", + tier=60, min_msgs=5, page=1, page_size=20) + assert {r["friend_id"] for r in t60["items"]} == {"c75"} + t90 = await metrics.churn(session, channel="wechat", kind="real", + tier=90, min_msgs=5, page=1, page_size=20) + assert {r["friend_id"] for r in t90["items"]} == {"c100"} + + await make_client(seed) + + +async def test_churn_handover_excludes_null_user(make_client, monkeypatch): + """user_id NULL(未映射归属)的客户不进待交接名单。""" + monkeypatch.setattr(metrics, "_now_ms", lambda: _NOW) + async def seed(session): + session.add(Staff(user_id="u9", name="离职的", is_on_job=False)) + session.add(_cust("c_hand", user_id="u9")) + session.add(_cust("c_nouser", user_id=None)) + await session.commit() + p = 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 p["items"]} == {"c_hand"} + + await make_client(seed) + + +def test_next_tier_bad_tier_raises_value_error(): + """非法 tier 报 ValueError(与 churn 的 kind 分支报错类型一致)。""" + with pytest.raises(ValueError): + metrics._next_tier(45) + assert metrics._next_tier(90) == 10_000 # 90 档=≥90 的开区间上限