yunke/yunke-server/app/analytics/metrics.py

"""A1/A3/B2 指标函数——纯查询,口径速查表见 spec §8。

时区:day 键一律 Asia/Shanghai 自然日(Python 侧算,禁 SQL 日期函数)。
周起始=周一(scope=week)。
"""
from __future__ import annotations

import calendar
import datetime as dt
from collections.abc import Iterable
from typing import Any
from zoneinfo import ZoneInfo

import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.models import (
    Customer,
    Message,
    MsgDailyStats,
    PhoneCall,
    SensitiveHit,
    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

_SH = ZoneInfo("Asia/Shanghai")


def _date_key(d: dt.date) -> int:
    return d.year * 10000 + d.month * 100 + d.day


def today_key(now: dt.datetime | None = None) -> int:
    """上海自然日 yyyymmdd(测试 monkeypatch 点)。"""
    return _date_key((now or dt.datetime.now(_SH)).date())


def _day_date(day: int) -> dt.date:
    return dt.date(day // 10000, (day // 100) % 100, day % 100)


def scope_day_keys(scope: str, today: int) -> list[int]:
    """today={今};week=本周一..今天;month=本月1号..今天(含端点)。"""
    end = _day_date(today)
    if scope == "today":
        start = end
    elif scope == "week":
        start = end - dt.timedelta(days=end.weekday())  # 周一=0
    elif scope == "month":
        start = end.replace(day=1)
    else:
        raise ValueError(f"bad scope: {scope!r}")
    out, cur = [], start
    while cur <= end:
        out.append(_date_key(cur))
        cur += dt.timedelta(days=1)
    return out


def _prev_scope_days(scope: str, days: list[int]) -> list[int]:
    """同期对齐环比窗(与 days 等长逐日对齐;今日 scope 不调用)。

    week=各日-7天(本周一..今 vs 上周一..上周同日);month=上月同日
    (短月钳到上月末,如 3/31→2/28)。同期对齐而非滚动紧邻窗——后者
    周会跨周末、月取上月尾巴,管理者会误读成「上周/上月」(spec §8)。
    """
    if scope == "week":
        return [_date_key(_day_date(d) - dt.timedelta(days=7)) for d in days]
    if scope == "month":
        out = []
        for d in days:
            cur = _day_date(d)
            y, m = cur.year, cur.month - 1
            if m == 0:
                y, m = y - 1, 12
            out.append(
                _date_key(dt.date(y, m, min(cur.day, calendar.monthrange(y, m)[1])))
            )
        return out
    raise ValueError(f"bad scope: {scope!r}")


async def _account_map(session: AsyncSession, channel: str | None) -> dict[str, str]:
    """channel=None(全平台)→ 两渠道账号映射合并;owner 账号格式跨渠道不碰撞。"""
    if channel is not None:
        return await account_map(session, channel)
    merged = await account_map(session, "wechat")
    merged.update(await account_map(session, "whatsapp"))
    return merged


async def _day_rows(
    session: AsyncSession, channel: str | None, days: Iterable[int]
) -> list[Any]:
    # channel=None(全平台)→ 不加渠道条件,两渠道行合并(peer_id 跨渠道不碰撞)
    conds = [MsgDailyStats.day.in_(list(days))]
    if channel is not None:
        conds.append(MsgDailyStats.channel == channel)
    return (
        await session.execute(
            sa.select(
                MsgDailyStats.day,
                MsgDailyStats.peer_kind,
                MsgDailyStats.peer_id,
                MsgDailyStats.send_cnt,
                MsgDailyStats.recv_cnt,
            ).where(*conds)
        )
    ).all()


async def _call_count(session: AsyncSession, days: list[int]) -> int:
    """phone_call 落窗计数(3k 行小表;start_call_time ms 与 day 键边界换算)。

    通话不按 messaging channel 过滤(spec §8:含全部 call-type)。
    """
    if not days:
        return 0
    lo, _hi = day_bounds_ms(days[0])
    hi_day_start, _ = day_bounds_ms(days[-1])
    hi = hi_day_start + 86_400_000  # 含末日全天
    val = (
        await session.execute(
            sa.select(sa.func.count()).select_from(PhoneCall).where(
                PhoneCall.start_call_time.is_not(None),
                PhoneCall.start_call_time >= lo,
                PhoneCall.start_call_time < hi,
            )
        )
    ).scalar_one()
    return int(val)


async def _first_contact(
    session: AsyncSession, channel: str | None, days: list[int]
) -> int:
    """首次沟通客户数:peer 按渠道全史 MIN(day) 落窗(跨员工归一,不跨渠道)。

    channel=None(全平台)按 peer 全史 MIN(day)(peer 跨渠道不碰撞,直接
    跨渠道合并首现日)。
    """
    conds = [MsgDailyStats.peer_kind == "friend"]
    if channel is not None:
        conds.append(MsgDailyStats.channel == channel)
    rows = (
        await session.execute(
            sa.select(MsgDailyStats.peer_id, sa.func.min(MsgDailyStats.day))
            .where(*conds)
            .group_by(MsgDailyStats.peer_id)
        )
    ).all()
    dayset = set(days)
    return sum(1 for _peer, first in rows if first in dayset)


async def _active_staff(
    session: AsyncSession, channel: str | None, days: list[int]
) -> tuple[int, int]:
    """活跃员工 (distinct 数, 未映射账号数);stat_daily_staff 空表回退 msg_daily_stats+映射。

    channel=None(全平台):distinct user_id 跨渠道去重(同人两渠道各活跃
    算 1);映射取两渠道合并——看板 all 口径「跨渠道重复计数不拆」即此。

    未映射口径(spec §8「未映射账号按账号计并标注」):
    - rollup:当窗 distinct user_id 不在 account_map 值集(=在册员工 user_id
      全集)——recompute 以 amap.get(owner, owner) 落表,映射不上的 owner
      以原值混入 user_id 列,值集之外的即未映射;
    - 回退:当窗 owner 中 amap.get(o, o)==o(映射不上,按账号原值计入
      活跃数)的计未映射——与 rollup 口径一致(映射不上的账号原样出现)。
    """
    if await _staff_rows(session):
        conds = [StatDailyStaff.day.in_(days)]
        if channel is not None:
            conds.append(StatDailyStaff.channel == channel)
        uids = set(
            (
                await session.execute(
                    sa.select(StatDailyStaff.user_id).where(*conds)
                )
            ).scalars()
        )
        mapped = set((await _account_map(session, channel)).values())
        return len(uids), len(uids - mapped)
    amap = await _account_map(session, channel)
    conds = [MsgDailyStats.day.in_(days)]
    if channel is not None:
        conds.append(MsgDailyStats.channel == channel)
    owners = {o for (o,) in (
        await session.execute(
            sa.select(MsgDailyStats.owner_account_id).where(*conds)
        )
    ).all()}
    unmapped = sum(1 for o in owners if o not in amap)
    return len({amap.get(o, o) for o in owners}), unmapped


def anomaly_threshold() -> float:
    return 0.5  # 昨日 vs 前 7 日均值偏离阈值,v1 固定


async def anomaly(session: AsyncSession, channel: str | None) -> dict | None:
    """E0-lite 异常警示:昨日总量 vs 前 7 日均值偏离>50%,明细到员工。

    channel=None(全平台):rollup/回退查询去渠道条件后,同 user_id 跨渠道
    行在逐日累加阶段自然合并(user 级总量/明细跨渠道口径)。
    """
    today = today_key()
    yesterday = _day_date(today) - dt.timedelta(days=1)

    yday = _date_key(yesterday)
    prev7 = [_date_key(yesterday - dt.timedelta(days=i)) for i in range(1, 8)]

    use_rollup = await _staff_rows(session)
    if use_rollup:
        conds = [StatDailyStaff.day.in_([yday] + prev7)]
        if channel is not None:
            conds.append(StatDailyStaff.channel == channel)
        rows = (
            await session.execute(
                sa.select(
                    StatDailyStaff.day, StatDailyStaff.user_id,
                    sa.func.coalesce(StatDailyStaff.msg_sent + StatDailyStaff.msg_recv, 0),
                ).where(*conds)
            )
        ).all()
    else:
        amap = await _account_map(session, channel)
        conds = [MsgDailyStats.day.in_([yday] + prev7)]
        if channel is not None:
            conds.append(MsgDailyStats.channel == channel)
        raw = (
            await session.execute(
                sa.select(
                    MsgDailyStats.day, MsgDailyStats.owner_account_id,
                    sa.func.coalesce(
                        sa.func.sum(MsgDailyStats.send_cnt + MsgDailyStats.recv_cnt), 0
                    ),
                )
                .where(*conds)
                .group_by(MsgDailyStats.day, MsgDailyStats.owner_account_id)
            )
        ).all()
        rows = [(d, amap.get(o, o), v) for d, o, v in raw]

    # 逐 user 显式算(yesterday 量 + prev7 均值),不猜 rows 顺序。
    # 累加而非覆盖:回退路径同人(多账号映射到同一 user_id)会有多行/日。
    day_rows: dict[str, dict[int, int]] = {}
    for day, uid, v in rows:
        by_day = day_rows.setdefault(uid, {})
        by_day[day] = by_day.get(day, 0) + int(v)

    staff_out = []
    totals_y, totals_p = 0, 0.0
    for uid, by_day in day_rows.items():
        y = by_day.get(yday, 0)
        pvals = [by_day.get(d, 0) for d in prev7]
        pmean = sum(pvals) / len(pvals) if pvals else 0.0
        totals_y += y
        totals_p += pmean
        if pmean > 0 and abs(y - pmean) / pmean > anomaly_threshold():
            staff_out.append({
                "user_id": uid,
                "yesterday": y,
                "prev7_mean": round(pmean, 1),
            })
        elif pmean == 0 and y > 0:
            # 新员工:前 7 日无量、昨日有量——昨日新增的大异常源,不能从明细消失
            staff_out.append({
                "user_id": uid,
                "yesterday": y,
                "prev7_mean": 0.0,
                "new": True,
            })
    if totals_p <= 0:
        return None
    delta = (totals_y - totals_p) / totals_p
    if abs(delta) > anomaly_threshold():
        return {
            "total_delta_pct": round(delta, 3),
            "yesterday_total": totals_y,
            "prev7_mean": round(totals_p, 1),
            "staff": sorted(staff_out, key=lambda s: -s["yesterday"]),
        }
    return None


async def staff_ranking(
    session: AsyncSession, *, channel: str | None, scope: str,
) -> list[dict]:
    """员工活跃排行(A2 核心子集,2026-09-07 管理者反馈补强)。

    口径:msg_total=stat_daily_staff 聚合(空表回退 msg_daily_stats+映射,
    同 _active_staff 模式);call_cnt=phone_call 按 user_id 落窗直查(3k 行
    小表);环比=与 _prev_scope_days 同期对比(None=前窗无数据);名字取
    staff.name,无 staff 行/未映射账号回退 user_id 原值标 unmapped。
    排序 msg_total desc + user_id asc 决胜。返回全部行(员工 ~41)。

    channel=None(全平台):查询去渠道条件+映射合并,同 user 跨渠道量
    自然合到一行(员工排行本就是 user 级)。

    active_friends/active_rooms 两路径统一从 msg_daily_stats 算窗内
    COUNT(DISTINCT peer)(映射合并到员工后,同 recompute_days 模式):
    stat_daily_staff 的 active_* 是**当日快照**,跨日 SUM 会重复计数、
    MAX 只是峰值,皆非「窗内活跃去重数」——rollup 只供 msg_total,
    避免两路径语义分叉。
    """
    days = scope_day_keys(scope, today_key())
    amap = await _account_map(session, channel)

    # 活跃客户/群 + 回退 msg_total:落窗拉行,Python 侧映射后按 user 去重
    # (COUNT(DISTINCT peer) 必须在「账号合并到员工」之后——同人多账号
    # 跨账号去重,写 SQL 反而绕且踩 only_full_group_by)
    conds = [MsgDailyStats.day.in_(days)]
    if channel is not None:
        conds.append(MsgDailyStats.channel == channel)
    rows = (
        await session.execute(
            sa.select(
                MsgDailyStats.owner_account_id,
                MsgDailyStats.peer_kind,
                MsgDailyStats.peer_id,
                MsgDailyStats.send_cnt,
                MsgDailyStats.recv_cnt,
            ).where(*conds)
        )
    ).all()
    act: dict[str, dict[str, set]] = {}
    msg_fallback: dict[str, int] = {}
    for owner, kind, peer, s, r in rows:
        uid = amap.get(owner, owner)
        a = act.setdefault(uid, {"friends": set(), "rooms": set()})
        if kind == "friend":  # 显式分支:未来新 kind 跳过,不误入 rooms
            a["friends"].add(peer)
        elif kind == "room":
            a["rooms"].add(peer)
        msg_fallback[uid] = msg_fallback.get(uid, 0) + int(s or 0) + int(r or 0)

    use_rollup = await _staff_rows(session)
    if use_rollup:
        conds = [StatDailyStaff.day.in_(days)]
        if channel is not None:
            conds.append(StatDailyStaff.channel == channel)
        rollup = (
            await session.execute(
                sa.select(
                    StatDailyStaff.user_id,
                    sa.func.coalesce(
                        sa.func.sum(
                            StatDailyStaff.msg_sent + StatDailyStaff.msg_recv
                        ),
                        0,
                    ),
                )
                .where(*conds)
                .group_by(StatDailyStaff.user_id)
            )
        ).all()
        msg_total = {uid: int(v) for uid, v in rollup}
    else:
        msg_total = msg_fallback

    # 通话:按 user_id 落窗直查(窗口换算同 _call_count 写法)
    calls: dict[str, int] = {}
    if days:
        lo, _ = day_bounds_ms(days[0])
        hi_start, _ = day_bounds_ms(days[-1])
        call_rows = (
            await session.execute(
                sa.select(PhoneCall.user_id, sa.func.count())
                .where(
                    PhoneCall.user_id.is_not(None),
                    PhoneCall.start_call_time.is_not(None),
                    PhoneCall.start_call_time >= lo,
                    PhoneCall.start_call_time < hi_start + 86_400_000,
                )
                .group_by(PhoneCall.user_id)
            )
        ).all()
        calls = {uid: int(v) for uid, v in call_rows}

    # 环比:同期对齐前窗(仅周/月;今日 vs 昨日全天会误导,与 overview 同口径)
    prev_total: dict[str, int] = {}
    if scope in ("week", "month"):
        prev_days = _prev_scope_days(scope, days)
        if use_rollup:
            conds = [StatDailyStaff.day.in_(prev_days)]
            if channel is not None:
                conds.append(StatDailyStaff.channel == channel)
            prev_rows = (
                await session.execute(
                    sa.select(
                        StatDailyStaff.user_id,
                        sa.func.coalesce(
                            sa.func.sum(
                                StatDailyStaff.msg_sent + StatDailyStaff.msg_recv
                            ),
                            0,
                        ),
                    )
                    .where(*conds)
                    .group_by(StatDailyStaff.user_id)
                )
            ).all()
            prev_total = {uid: int(v) for uid, v in prev_rows}
        else:
            conds = [MsgDailyStats.day.in_(prev_days)]
            if channel is not None:
                conds.append(MsgDailyStats.channel == channel)
            prev_rows = (
                await session.execute(
                    sa.select(
                        MsgDailyStats.owner_account_id,
                        sa.func.coalesce(
                            sa.func.sum(
                                MsgDailyStats.send_cnt + MsgDailyStats.recv_cnt
                            ),
                            0,
                        ),
                    )
                    .where(*conds)
                    .group_by(MsgDailyStats.owner_account_id)
                )
            ).all()
            for owner, v in prev_rows:
                uid = amap.get(owner, owner)  # 同人多账号累加,不覆盖
                prev_total[uid] = prev_total.get(uid, 0) + int(v)

    # 名字/未映射:staff 几十行全拉;映射值集=在册员工 user_id 全集
    staff_names = {
        uid: name
        for uid, name in (
            await session.execute(sa.select(Staff.user_id, Staff.name))
        ).all()
    }
    mapped_uids = set(amap.values())

    out = []
    for uid in sorted(set(msg_total) | set(act) | set(calls),
                      key=lambda u: (-msg_total.get(u, 0), u)):
        cur, prev = msg_total.get(uid, 0), prev_total.get(uid, 0)
        out.append({
            "user_id": uid,
            "name": staff_names.get(uid) or uid,
            "unmapped": uid not in staff_names and uid not in mapped_uids,
            "msg_total": cur,
            "active_friends": len(act[uid]["friends"]) if uid in act else 0,
            "active_rooms": len(act[uid]["rooms"]) if uid in act else 0,
            "call_cnt": calls.get(uid, 0),
            "msg_delta": round((cur - prev) / prev, 3) if cur and prev else None,
        })
    return out


async def action_counts(session: AsyncSession, *, channel: str | None) -> dict:
    """预警行动入口计数(看板顶部卡片):复用已测试的查询路径,各取 total。

    channel=None(全平台):三计数全部去渠道过滤(含 sensitive_pending)。
    """
    un = await unreplied(session, channel=channel, hours=2, page=1, page_size=1)
    ch = await churn(session, channel=channel, kind="real", tier=30, min_msgs=5,
                     page=1, page_size=1)
    pending_conds = [SensitiveHit.disposition == "pending"]
    if channel is not None:
        pending_conds.append(SensitiveHit.channel == channel)
    pending = (
        await session.execute(
            sa.select(sa.func.count()).select_from(SensitiveHit).where(*pending_conds)
        )
    ).scalar_one()
    return {
        "unreplied_2h": int(un["total"]),
        "churn_real_30d": int(ch["total"]),
        "sensitive_pending": int(pending),
    }


async def overview(
    session: AsyncSession, *, channel: str | None, scope: str, trend_days: int = 30
) -> dict[str, Any]:
    """A1:卡片 + 环比(仅周/月,同期对齐) + 近 N 天趋势 + 异常警示。

    channel=None(全平台,看板 v2 默认):所有指标跨渠道合并——msg/活跃客户
    直接合计(peer_id 跨渠道不碰撞);活跃员工按 user 去重(跨渠道重复计数
    不拆);通话量本就是 call-type 无渠道语义。此时追加:
    - by_channel:微信/WhatsApp 两渠道的 msg_total/active_customers/
      first_contact 拆分(卡片下小字;活跃员工/通话量不拆,口径不成立);
    - trend_by_channel:两渠道各自趋势序列(仅 all 时出现;单渠道不加键)。
    trend 在 all 时=两渠道合计(单系列口径保持可用)。
    cards.unmapped_accounts=活跃数里映射不上的账号数(spec §8 按账号计并标注)。
    """
    if trend_days < 1 or trend_days > 90:
        raise ValueError("trend_days must be 1..90")
    today = today_key()
    days = scope_day_keys(scope, today)
    rows = await _day_rows(session, channel, days)
    msg_total = sum(int(s or 0) + int(r or 0) for _d, _k, _p, s, r in rows)
    active_customers = len({p for _d, kind, p, _s, _r in rows if kind == "friend"})

    active_staff, unmapped = await _active_staff(session, channel, days)
    cards: dict[str, Any] = {
        "msg_total": msg_total,
        "active_customers": active_customers,
        "active_staff": active_staff,
        "unmapped_accounts": unmapped,
        "call_cnt": await _call_count(session, days),
        "first_contact": await _first_contact(session, channel, days),
    }

    # 环比仅周/月,同期对齐(今日 vs 昨日全天会误导,spec §8)
    if scope in ("week", "month"):
        prev_days = _prev_scope_days(scope, days)
        prev_rows = await _day_rows(session, channel, prev_days)
        prev_total = sum(int(s or 0) + int(r or 0) for _d, _k, _p, s, r in prev_rows)
        cards["prev_msg_total"] = prev_total
        cards["msg_total_delta"] = (
            round((msg_total - prev_total) / prev_total, 3) if prev_total else None
        )

    # 趋势:近 trend_days 天 每日 消息量/活跃客户/通话量
    end = _day_date(today)
    t_days = []
    cur = end - dt.timedelta(days=trend_days - 1)
    while cur <= end:
        t_days.append(_date_key(cur))
        cur += dt.timedelta(days=1)
    # 通话趋势:phone_call 拉窗内 (start_call_time) Python 侧按日分桶(3k 行)
    lo, _ = day_bounds_ms(t_days[0])
    hi_day, _ = day_bounds_ms(t_days[-1])
    calls = (
        await session.execute(
            sa.select(PhoneCall.start_call_time).where(
                PhoneCall.start_call_time.is_not(None),
                PhoneCall.start_call_time >= lo,
                PhoneCall.start_call_time < hi_day + 86_400_000,
            )
        )
    ).scalars().all()
    call_by_day: dict[int, int] = {}
    for ts in calls:
        k = _date_key(dt.datetime.fromtimestamp(ts / 1000, _SH).date())
        call_by_day[k] = call_by_day.get(k, 0) + 1

    def _trend_points(rows: list[Any], call_by: dict[int, int]) -> list[dict]:
        by_day: dict[int, dict] = {d: {"msg": 0, "cust": set()} for d in t_days}
        for d, kind, p, s, r in rows:
            by_day[d]["msg"] += int(s or 0) + int(r or 0)
            if kind == "friend":
                by_day[d]["cust"].add(p)
        return [
            {"day": d, "msg": v["msg"], "active_customers": len(v["cust"]),
             "call_cnt": call_by.get(d, 0)}
            for d, v in by_day.items()
        ]

    t_rows = await _day_rows(session, channel, t_days)
    trend = _trend_points(t_rows, call_by_day)

    out: dict[str, Any] = {
        "scope": scope,
        "channel": "all" if channel is None else channel,
        "days": days,
        "cards": cards,
        "trend": trend,
        "anomaly": await anomaly(session, channel),
        "staff_ranking": await staff_ranking(session, channel=channel, scope=scope),
        "actions": await action_counts(session, channel=channel),
    }
    if channel is None:
        # 拆分小字/双线趋势的数据源:复用同查询路径按渠道各跑一遍(小表毫秒级)。
        # 渠道序列 call_cnt 恒 0——通话是 call-type 无渠道语义不拆,前端 all
        # 模式通话量画总趋势单线。
        out["by_channel"] = {}
        out["trend_by_channel"] = {}
        for ch in ("wechat", "whatsapp"):
            ch_rows = await _day_rows(session, ch, days)
            out["by_channel"][ch] = {
                "msg_total": sum(
                    int(s or 0) + int(r or 0) for _d, _k, _p, s, r in ch_rows
                ),
                "active_customers": len(
                    {p for _d, kind, p, _s, _r in ch_rows if kind == "friend"}
                ),
                "first_contact": await _first_contact(session, ch, days),
            }
            out["trend_by_channel"][ch] = _trend_points(
                await _day_rows(session, ch, t_days), {}
            )
    return out


def _now_ms() -> int:
    return int(dt.datetime.now(_SH).timestamp() * 1000)


def _next_tier(tier: int) -> int:
    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 | None, 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。

    channel=None(全平台):去渠道过滤,两渠道客户合并(peer_id 跨渠道不
    碰撞,历史消息半连接按 peer 聚合同样成立);items[].channel 保留原值
    供前端跳转分流。

    active_days=历史活跃天数(distinct 自然日;peer 跨账号转移同日多行只计一日)。
    """
    day_ms = 86_400_000
    now = _now_ms()
    base = [
        sa.or_(Customer.delete_flag.is_(None), Customer.delete_flag == 0),
    ]
    if channel is not None:
        base.append(Customer.channel == channel)
    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 语义正确):半连接聚合
        hist_conds = [MsgDailyStats.peer_kind == "friend"]
        if channel is not None:
            hist_conds.append(MsgDailyStats.channel == channel)
        base.append(
            Customer.friend_id.in_(
                sa.select(MsgDailyStats.peer_id)
                .where(*hist_conds)
                .group_by(MsgDailyStats.peer_id)
                .having(
                    sa.func.coalesce(
                        sa.func.sum(MsgDailyStats.send_cnt + MsgDailyStats.recv_cnt), 0
                    ) >= min_msgs
                )
            )
        )
        # 决胜列 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()]
    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 下语义天然等价);
        # NULL 尾部密集(离职员工的未激活客户),必须带 friend_id 决胜列防翻页重复
        order = [Customer.last_chat_time.desc().nulls_last(),
                 Customer.friend_id.asc()]
    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 聚合;
    # 活跃天数按 distinct 自然日计——peer 跨账号同日多行不重复计)
    items = []
    if rows:
        ids = [c.friend_id for c in rows]
        agg_conds = [MsgDailyStats.peer_kind == "friend",
                     MsgDailyStats.peer_id.in_(ids)]
        if channel is not None:
            agg_conds.append(MsgDailyStats.channel == channel)
        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(sa.distinct(MsgDailyStats.day)),
                )
                .where(*agg_conds)
                .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}


_VALID_TAIL_TYPES = ("text", "image", "voice", "video", "file", "link")

# 候选量阈值:≤此值决赛圈全量点查(total 精确);超过则按页点查(total=候选数)
_FINALIZE_ALL_MAX = 500


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 | None, hours: int, page: int,
    page_size: int,
) -> dict:
    """B2 该回未回(spec §3.4)。候选=汇总表最新日行;决赛圈点查类型/预览/通话豁免。

    channel=None(全平台,看板 v2 计数用):候选去渠道条件,窗口 partition
    本含 channel 列——(渠道,owner,peer) 三元组各自取最新日行,跨渠道
    天然正确;决赛圈点查以 (owner,peer) 定位(peer 跨渠道不碰撞),回显
    channel 取尾部消息实值供前端跳转分流。

    候选资格看「最新日行」(row_number 先于方向/超时过滤——员工次日已回的
    旧未回日不复活);决赛圈再点查尾部消息:须员工未回(is_send 非 True——
    stats/message 漂移时以 message 实尾为准,写钩子吞重算失败是已记录状态;
    is_send=None 保留=与统计同向保守)且有效类型(排除 unknown/divider 等),
    剔除已删/拉黑客户,通话豁免(该客户 phone 任一晚于最后消息的通话=已回应,
    不限定通话员工——客户被回应过即可,spec:尾部=max(最后消息,最后通话))。
    无 customer 行的 peer(voice-only)保留在清单——豁免不可判定,保守方向
    是有意为之。

    total 口径:候选 ≤ _FINALIZE_ALL_MAX 时决赛圈全过滤后分页(total 精确);
    超过则按候选分页(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_conds = [
        MsgDailyStats.peer_kind == "friend",
        MsgDailyStats.last_ts.is_not(None),
    ]
    if channel is not None:
        sub_conds.append(MsgDailyStats.channel == channel)
    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(*sub_conds)
    ).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)
        msg_conds = [
            Message.owner_account_id == owner,
            Message.talker_id == peer,
            sa.or_(Message.room_id.is_(None), Message.room_id == ""),
            Message.timestamp.is_not(None),
        ]
        if channel is not None:
            msg_conds.append(Message.channel == channel)
        last_msg = (
            await session.execute(
                sa.select(Message).where(*msg_conds)
                .order_by(Message.timestamp.desc()).limit(1)
            )
        ).scalar_one_or_none()
        if last_msg is None or last_msg.is_send 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": last_msg.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)。
        # 真预算:≤_FINALIZE_ALL_MAX 分支=全部候选点查(≤500 候选 ×~3 查询
        # ≈1500 顺序往返,毫秒级/个);>500 分支=仅本页 ≤100 候选。豁免查询
        # phone_call.call_number 无索引(3k 行小表扫;量大时批量化 IN 或补
        # 索引——后续优化不在本次)
        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) <= _FINALIZE_ALL_MAX:
        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}