From 52af5315981bf589111f96a1b6c798394457cf60 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Fri, 4 Sep 2026 19:00:15 +0800 Subject: [PATCH] =?UTF-8?q?feat(analytics):=20B2=20=E8=AF=A5=E5=9B=9E?= =?UTF-8?q?=E6=9C=AA=E5=9B=9E=E2=80=94=E2=80=94=E5=B0=BE=E9=83=A8=E5=88=A4?= =?UTF-8?q?=E5=AE=9A/=E6=9C=89=E6=95=88=E7=B1=BB=E5=9E=8B/=E9=80=9A?= =?UTF-8?q?=E8=AF=9D=E8=B1=81=E5=85=8D/=E5=B7=B2=E5=88=A0=E5=89=94?= =?UTF-8?q?=E9=99=A4?= 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 | 149 ++++++++++++++++- yunke-server/app/api/routers/analytics.py | 14 ++ .../tests/test_analytics_churn_unreplied.py | 151 +++++++++++++++++- 3 files changed, 312 insertions(+), 2 deletions(-) diff --git a/yunke-server/app/analytics/metrics.py b/yunke-server/app/analytics/metrics.py index 85ebbcf..ddb5d4b 100644 --- a/yunke-server/app/analytics/metrics.py +++ b/yunke-server/app/analytics/metrics.py @@ -14,7 +14,14 @@ from zoneinfo import ZoneInfo import sqlalchemy as sa from sqlalchemy.ext.asyncio import AsyncSession -from app.db.models import Customer, MsgDailyStats, PhoneCall, Staff, StatDailyStaff +from app.db.models import ( + Customer, + Message, + 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 @@ -449,3 +456,143 @@ async def churn( "active_days": d, }) return {"items": items, "total": int(total), "page": page, "page_size": page_size} + + +_VALID_TAIL_TYPES = ("text", "image", "voice", "video", "file", "link") + + +def _norm_phone(p: str | None) -> str | None: + """归一:+86/空格/横线剥掉(spec B1 同款规则)。""" + if not p: + return None + out = p.replace(" ", "").replace("-", "") + for pre in ("+86", "86"): + if out.startswith(pre) and len(out) > len(pre): + out = out[len(pre):] + break + return out or None + + +async def unreplied( + session: AsyncSession, *, channel: str, hours: int, page: int, page_size: int, +) -> dict: + """B2 该回未回(spec §3.4)。候选=汇总表最新日行;决赛圈点查类型/预览/通话豁免。 + + 候选资格看「最新日行」(row_number 先于方向/超时过滤——员工次日已回的 + 旧未回日不复活);决赛圈再点查尾部消息有效类型(排除 unknown/divider 等)、 + 剔除已删/拉黑客户、通话豁免(该客户 phone 任一晚于最后消息的通话=已回应, + 不限定通话员工——客户被回应过即可,spec:尾部=max(最后消息,最后通话))。 + + total 口径:候选 ≤500 时决赛圈全过滤后分页(total 精确);>500 时按候选 + 分页(total=候选数,注释口径——决赛圈过滤可能让个别页少几行)。 + """ + now = _now_ms() + cutoff = now - hours * 3600 * 1000 + + # 候选集:每 (channel,owner,peer) 最新日行(row_number 双方言:SQLite + # 3.25+/MySQL 8)。方向=客户 且 last_ts ≤ cutoff 的过滤作用在 rn==1 行上 + rn = ( + sa.func.row_number() + .over( + partition_by=( + MsgDailyStats.channel, + MsgDailyStats.owner_account_id, + MsgDailyStats.peer_id, + ), + order_by=MsgDailyStats.day.desc(), + ) + .label("rn") + ) + sub = ( + sa.select( + MsgDailyStats.owner_account_id.label("owner"), + MsgDailyStats.peer_id.label("peer"), + MsgDailyStats.last_ts.label("ts"), + MsgDailyStats.last_is_send.label("last_is_send"), + rn, + ) + .where( + MsgDailyStats.channel == channel, + MsgDailyStats.peer_kind == "friend", + MsgDailyStats.last_ts.is_not(None), + ) + ).subquery() + cand = ( + await session.execute( + sa.select(sub.c.owner, sub.c.peer, sub.c.ts) + .where( + sub.c.rn == 1, + sub.c.last_is_send.is_(False), + sub.c.ts <= cutoff, + ) + .order_by(sub.c.ts.asc(), sub.c.peer.asc(), sub.c.owner.asc()) + ) + ).all() + + async def _finalist(owner: str, peer: str, ts: int) -> dict | None: + # 决赛圈点查:尾部消息类型+预览(走 ix_message_channel_owner_talker_time) + last_msg = ( + await session.execute( + sa.select(Message).where( + Message.channel == channel, + Message.owner_account_id == owner, + Message.talker_id == peer, + sa.or_(Message.room_id.is_(None), Message.room_id == ""), + Message.timestamp.is_not(None), + ).order_by(Message.timestamp.desc()).limit(1) + ) + ).scalar_one_or_none() + if last_msg is None or last_msg.type not in _VALID_TAIL_TYPES: + return None + cust = await session.get(Customer, peer) + if cust is not None and ( + cust.delete_flag == 1 or cust.is_block is True + ): + return None + # 通话豁免:该客户手机号有「晚于最后消息」的通话 → 已回应。 + # customer 侧归一后与 call_number 精确等值——call_number 若带 +86 + # 前缀变体则不中,v1 可接受(spec B1 同款规则)。 + phone = _norm_phone(cust.phone if cust else None) + if phone: + call_ts = ( + await session.execute( + sa.select(sa.func.max(PhoneCall.start_call_time)).where( + PhoneCall.call_number == phone, + ) + ) + ).scalar_one_or_none() + if call_ts and call_ts > ts: + return None + text = (last_msg.text or "").replace("\n", " ") + return { + "peer_id": peer, + "channel": channel, + "owner_account_id": owner, + "last_ts": ts, + "wait_ms": now - ts, + "preview": text[:50], + "msg_id": last_msg.msg_id, + } + + async def _finalize(rows: list[Any]) -> list[dict]: + # 顺序逐个 await——AsyncSession 不能并发执行查询(禁 asyncio.gather); + # 每页 ≤100 个点查、每个毫秒级,顺序总耗时百毫秒级可接受 + out = [] + for o, p, t in rows: + r = await _finalist(o, p, t) + if r is not None: + out.append(r) + return out + + if len(cand) <= 500: + items = await _finalize(cand) + # keep 候选顺序(ts asc, peer asc)——顺序循环天然保序 + total = len(items) + start = (page - 1) * page_size + return {"items": items[start : start + page_size], "total": total, + "page": page, "page_size": page_size} + # 大候选量:按候选分页,total=候选数(口径注释——个别页决赛圈过滤后少几行) + start = (page - 1) * page_size + items = await _finalize(cand[start : start + page_size]) + return {"items": items, "total": len(cand), "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 0efc6aa..feb34e1 100644 --- a/yunke-server/app/api/routers/analytics.py +++ b/yunke-server/app/api/routers/analytics.py @@ -55,3 +55,17 @@ async def churn_list( session, channel=channel, kind=type, tier=tier, min_msgs=min_msgs, page=page, page_size=page_size, ) + + +@router.get("/unreplied") +async def unreplied_list( + session: SessionDep, + channel: ChannelParam = "wechat", + hours: int = Query(2, ge=1, le=168), + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), +) -> dict[str, Any]: + """B2 该回未回清单(spec §3.4)。hours=超时阈值(默认 2h)。""" + return await metrics_svc.unreplied( + session, channel=channel, hours=hours, 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 index 8e1814c..ad47bd7 100644 --- a/yunke-server/tests/test_analytics_churn_unreplied.py +++ b/yunke-server/tests/test_analytics_churn_unreplied.py @@ -1,10 +1,12 @@ """A3 三分法流失 + B2 该回未回。""" from __future__ import annotations +import datetime as dt + import pytest from app.analytics import metrics -from app.db.models import Customer, MsgDailyStats, Staff +from app.db.models import Customer, Message, MsgDailyStats, PhoneCall, Staff _NOW = 1756950000000 # 相对时间基准(绝对日期无关紧要,断言全部相对此值自洽) @@ -134,3 +136,150 @@ def test_next_tier_bad_tier_raises_value_error(): with pytest.raises(ValueError): metrics._next_tier(45) assert metrics._next_tier(90) == 10_000 # 90 档=≥90 的开区间上限 + + +_VALID_TYPES = ("text", "image", "voice", "video", "file", "link") + + +async def _seed_unreplied(session): + now = _NOW + h3 = 3 * 3600 * 1000 + # 该回未回:t1 客户最后一条(is_send=False,text)3h 前 + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t1", day=20260904, + send_cnt=5, recv_cnt=3, last_ts=now - h3, + last_is_send=False)) + session.add(Message(msg_id="t1m", channel="wechat", owner_account_id="wx1", + talker_id="t1", timestamp=now - h3, is_send=False, + type="text", text="在吗,方案发我一下", + fetched_at=dt.datetime(2026, 9, 4, 12, 0))) + # 员工已回 → 不在清单 + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t2", day=20260904, + send_cnt=5, recv_cnt=3, last_ts=now - h3, + last_is_send=True)) + # 客户最后一条是 unknown 类型 → 无效,不出清单 + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t3", day=20260904, + send_cnt=5, recv_cnt=3, last_ts=now - h3, + last_is_send=False)) + session.add(Message(msg_id="t3m", channel="wechat", owner_account_id="wx1", + talker_id="t3", timestamp=now - h3, is_send=False, + type="unknown", text=None, + fetched_at=dt.datetime(2026, 9, 4, 12, 0))) + # 尾部 0.5h < 2h 阈值 → 不出 + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t4", day=20260904, + send_cnt=5, recv_cnt=3, last_ts=now - 1800_000, + last_is_send=False)) + session.add(Message(msg_id="t4m", channel="wechat", owner_account_id="wx1", + talker_id="t4", timestamp=now - 1800_000, is_send=False, + type="text", text="刚发的", + fetched_at=dt.datetime(2026, 9, 4, 12, 0))) + # 通话豁免:最后消息 5h 前没回,但 4h 前员工打过电话 → 视为已回应 + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t5", day=20260904, + send_cnt=5, recv_cnt=3, last_ts=now - 5 * 3600 * 1000, + last_is_send=False)) + session.add(Message(msg_id="t5m", channel="wechat", owner_account_id="wx1", + talker_id="t5", timestamp=now - 5 * 3600 * 1000, + is_send=False, type="text", text="hello", + fetched_at=dt.datetime(2026, 9, 4, 12, 0))) + session.add(_cust("t5", last_chat=now, user_id="u1")) + (await session.get(Customer, "t5")).phone = "13800000005" + session.add(PhoneCall(call_id="pc1", channel="voice_call", + user_id="u1", call_number="13800000005", + start_call_time=now - 4 * 3600 * 1000)) + # 已删好友 → 剔除 + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t6", day=20260904, + send_cnt=5, recv_cnt=3, last_ts=now - h3, + last_is_send=False)) + session.add(Message(msg_id="t6m", channel="wechat", owner_account_id="wx1", + talker_id="t6", timestamp=now - h3, is_send=False, + type="text", text="已删我的", + fetched_at=dt.datetime(2026, 9, 4, 12, 0))) + session.add(_cust("t6", last_chat=now - h3, delete=1)) + await session.commit() + + +async def test_unreplied(make_client, monkeypatch): + monkeypatch.setattr(metrics, "_now_ms", lambda: _NOW) + async def seed(session): + await _seed_unreplied(session) + out = await metrics.unreplied(session, channel="wechat", hours=2, + page=1, page_size=20) + ids = [r["peer_id"] for r in out["items"]] + assert ids == ["t1"] # t2已回/t3无效类型/t4未超时/t5通话豁免/t6已删 + row = out["items"][0] + assert row["wait_ms"] >= 3 * 3600 * 1000 + assert "方案" in row["preview"] + + await make_client(seed) + + +async def test_unreplied_call_not_by_owner_still_exempts(make_client, monkeypatch): + """通话豁免语义=『该客户被回应过』——不强制通话是同一员工打的(spec:尾部=max(最后消息,最后通话))。""" + monkeypatch.setattr(metrics, "_now_ms", lambda: _NOW) + async def seed(session): + now = _NOW + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t7", day=20260904, + send_cnt=1, recv_cnt=1, last_ts=now - 6 * 3600 * 1000, + last_is_send=False)) + session.add(Message(msg_id="t7m", channel="wechat", owner_account_id="wx1", + talker_id="t7", timestamp=now - 6 * 3600 * 1000, + is_send=False, type="text", text="hi", + fetched_at=dt.datetime(2026, 9, 4, 12, 0))) + session.add(_cust("t7", last_chat=now)) + (await session.get(Customer, "t7")).phone = "13900000007" + # 通话是别的员工(user_id=u_other)打的,晚于最后消息 + session.add(PhoneCall(call_id="pc2", channel="voice_call", + user_id="u_other", call_number="13900000007", + start_call_time=now - 2 * 3600 * 1000)) + out = await metrics.unreplied(session, channel="wechat", hours=2, + page=1, page_size=20) + assert all(r["peer_id"] != "t7" for r in out["items"]) + + await make_client(seed) + + +async def test_unreplied_latest_day_row_wins(make_client, monkeypatch): + """资格判定看「最新日行」:更早的未回日不能复活。 + + 前日客户未回(已远超 2h),但今日员工回了(最新日行 last_is_send=True, + 会话尾部=员工)→ 不在清单。若候选过滤先于 row_number(在窗口子查询内 + 过滤),旧日行会被复活成候选——此测试钉死正确语义。 + """ + monkeypatch.setattr(metrics, "_now_ms", lambda: _NOW) + async def seed(session): + now = _NOW + h3 = 3 * 3600 * 1000 + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t8", day=20260903, + send_cnt=0, recv_cnt=3, last_ts=now - 30 * 3600 * 1000, + last_is_send=False)) + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t8", day=20260904, + send_cnt=5, recv_cnt=0, last_ts=now - h3, + last_is_send=True)) + session.add(Message(msg_id="t8m", channel="wechat", owner_account_id="wx1", + talker_id="t8", timestamp=now - h3, is_send=True, + type="text", text="已经回了", + fetched_at=dt.datetime(2026, 9, 4, 12, 0))) + out = await metrics.unreplied(session, channel="wechat", hours=2, + page=1, page_size=20) + assert all(r["peer_id"] != "t8" for r in out["items"]) + + await make_client(seed) + + +async def test_unreplied_endpoint_422(make_client): + async def seed(session): + ... + + client = await make_client(seed) + r = await client.get("/api/analytics/unreplied", params={"hours": 0}) + assert r.status_code == 422 + r2 = await client.get("/api/analytics/unreplied", params={"channel": "sms"}) + assert r2.status_code == 422