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 _day_rows(
    session: AsyncSession, channel: str, days: Iterable[int]
) -> list[Any]:
    return (
        await session.execute(
            sa.select(
                MsgDailyStats.day,
                MsgDailyStats.peer_kind,
                MsgDailyStats.peer_id,
                MsgDailyStats.send_cnt,
                MsgDailyStats.recv_cnt,
            ).where(
                MsgDailyStats.channel == channel, MsgDailyStats.day.in_(list(days))
            )
        )
    ).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, days: list[int]
) -> int:
    """首次沟通客户数:peer 按渠道全史 MIN(day) 落窗(跨员工归一,不跨渠道)。"""
    rows = (
        await session.execute(
            sa.select(MsgDailyStats.peer_id, sa.func.min(MsgDailyStats.day))
            .where(
                MsgDailyStats.channel == channel,
                MsgDailyStats.peer_kind == "friend",
            )
            .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, days: list[int]
) -> tuple[int, int]:
    """活跃员工 (distinct 数, 未映射账号数);stat_daily_staff 空表回退 msg_daily_stats+映射。

    未映射口径(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):
        uids = set(
            (
                await session.execute(
                    sa.select(StatDailyStaff.user_id).where(
                        StatDailyStaff.channel == channel,
                        StatDailyStaff.day.in_(days),
                    )
                )
            ).scalars()
        )
        mapped = set((await account_map(session, channel)).values())
        return len(uids), len(uids - mapped)
    amap = await account_map(session, channel)
    owners = {o for (o,) in (
        await session.execute(
            sa.select(MsgDailyStats.owner_account_id).where(
                MsgDailyStats.channel == channel, MsgDailyStats.day.in_(days)
            )
        )
    ).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) -> dict | None:
    """E0-lite 异常警示:昨日总量 vs 前 7 日均值偏离>50%,明细到员工。"""
    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:
        rows = (
            await session.execute(
                sa.select(
                    StatDailyStaff.day, StatDailyStaff.user_id,
                    sa.func.coalesce(StatDailyStaff.msg_sent + StatDailyStaff.msg_recv, 0),
                ).where(
                    StatDailyStaff.channel == channel,
                    StatDailyStaff.day.in_([yday] + prev7),
                )
            )
        ).all()
    else:
        amap = await account_map(session, 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(
                    MsgDailyStats.channel == channel,
                    MsgDailyStats.day.in_([yday] + prev7),
                )
                .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, 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)。

    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)
    rows = (
        await session.execute(
            sa.select(
                MsgDailyStats.owner_account_id,
                MsgDailyStats.peer_kind,
                MsgDailyStats.peer_id,
                MsgDailyStats.send_cnt,
                MsgDailyStats.recv_cnt,
            ).where(
                MsgDailyStats.channel == channel, MsgDailyStats.day.in_(days)
            )
        )
    ).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:
        rollup = (
            await session.execute(
                sa.select(
                    StatDailyStaff.user_id,
                    sa.func.coalesce(
                        sa.func.sum(
                            StatDailyStaff.msg_sent + StatDailyStaff.msg_recv
                        ),
                        0,
                    ),
                )
                .where(
                    StatDailyStaff.channel == channel,
                    StatDailyStaff.day.in_(days),
                )
                .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:
            prev_rows = (
                await session.execute(
                    sa.select(
                        StatDailyStaff.user_id,
                        sa.func.coalesce(
                            sa.func.sum(
                                StatDailyStaff.msg_sent + StatDailyStaff.msg_recv
                            ),
                            0,
                        ),
                    )
                    .where(
                        StatDailyStaff.channel == channel,
                        StatDailyStaff.day.in_(prev_days),
                    )
                    .group_by(StatDailyStaff.user_id)
                )
            ).all()
            prev_total = {uid: int(v) for uid, v in prev_rows}
        else:
            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(
                        MsgDailyStats.channel == channel,
                        MsgDailyStats.day.in_(prev_days),
                    )
                    .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) -> dict:
    """预警行动入口计数(看板顶部卡片):复用已测试的查询路径,各取 total。"""
    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 = (
        await session.execute(
            sa.select(sa.func.count()).select_from(SensitiveHit).where(
                SensitiveHit.channel == channel,
                SensitiveHit.disposition == "pending",
            )
        )
    ).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, scope: str, trend_days: int = 30
) -> dict[str, Any]:
    """A1:卡片 + 环比(仅周/月,同期对齐) + 近 N 天趋势 + 异常警示。

    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)
    t_rows = await _day_rows(session, channel, t_days)
    by_day: dict[int, dict] = {d: {"msg": 0, "cust": set()} for d in t_days}
    for d, kind, p, s, r in t_rows:
        by_day[d]["msg"] += int(s or 0) + int(r or 0)
        if kind == "friend":
            by_day[d]["cust"].add(p)
    # 通话趋势: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
    trend = [
        {"day": d, "msg": v["msg"], "active_customers": len(v["cust"]),
         "call_cnt": call_by_day.get(d, 0)}
        for d, v in by_day.items()
    ]

    return {
        "scope": scope,
        "channel": 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),
    }


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, 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。

    active_days=历史活跃天数(distinct 自然日;peer 跨账号转移同日多行只计一日)。
    """
    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
                )
            )
        )
        # 决胜列 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 = (
            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(
                    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}


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

    候选资格看「最新日行」(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 = (
        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.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": 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}