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

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

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

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 MsgDailyStats, PhoneCall, 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]:
    """紧邻的前一个等长窗口(环比;今日 scope 不调用)。"""
    n = len(days)
    last = _day_date(days[0]) - dt.timedelta(days=1)
    return [
        _date_key(last - dt.timedelta(days=i))
        for i in range(n - 1, -1, -1)
    ]


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, channel_scope_days: list[int]) -> int:
    """phone_call 落窗计数(3k 行小表;start_call_time ms 与 day 键边界换算)。"""
    if not channel_scope_days:
        return 0
    lo, _hi = day_bounds_ms(channel_scope_days[0])
    hi_day_start, _ = day_bounds_ms(channel_scope_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]
) -> int:
    """活跃员工 distinct user_id;stat_daily_staff 空表回退 msg_daily_stats+映射。"""
    if await _staff_rows(session):
        val = (
            await session.execute(
                sa.select(sa.func.count(sa.func.distinct(StatDailyStaff.user_id))).where(
                    StatDailyStaff.channel == channel,
                    StatDailyStaff.day.in_(days),
                )
            )
        ).scalar_one()
        return int(val)
    amap = await account_map(session, channel)
    owners = {amap.get(o, o) for (o,) in (
        await session.execute(
            sa.select(MsgDailyStats.owner_account_id).where(
                MsgDailyStats.channel == channel, MsgDailyStats.day.in_(days)
            )
        )
    ).all()}
    return len(owners)


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 顺序
    day_rows: dict[str, dict[int, int]] = {}
    for day, uid, v in rows:
        day_rows.setdefault(uid, {})[day] = 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),
            })
    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 天趋势 + 异常警示。"""
    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"})

    cards: dict[str, Any] = {
        "msg_total": msg_total,
        "active_customers": active_customers,
        "active_staff": await _active_staff(session, channel, days),
        "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),
    }