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

_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 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),
    }


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}