feat(analytics): B2 该回未回——尾部判定/有效类型/通话豁免/已删剔除

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

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}

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

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