diff --git a/yunke-server/app/analytics/metrics.py b/yunke-server/app/analytics/metrics.py new file mode 100644 index 0000000..2f5ff46 --- /dev/null +++ b/yunke-server/app/analytics/metrics.py @@ -0,0 +1,291 @@ +"""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), + } diff --git a/yunke-server/tests/test_analytics_overview.py b/yunke-server/tests/test_analytics_overview.py new file mode 100644 index 0000000..41b3608 --- /dev/null +++ b/yunke-server/tests/test_analytics_overview.py @@ -0,0 +1,91 @@ +"""A1 总览指标:卡片/首沟通口径/环比(仅周月)/异常警示。""" +from __future__ import annotations + +from app.analytics import metrics +from app.db.models import ( + MsgDailyStats, + PhoneCall, + StatDailyStaff, + WechatAccount, +) + +_TODAY = 20260904 # 测试统一用固定 day 键,monkeypatch today_key + + +async def _seed(session): + session.add(WechatAccount(wx_id="wx1", user_id="u1", user_name="张三")) + # 昨日(20260903):u1 90 条;前 7 日均值(20260827-0902):每天 200 → 偏离 -55%>50% 触发 + # (阈值是严格 >0.5——正好 -50% 不触发,seed 用 90) + for day, sent in [(20260827, 200), (20260828, 200), (20260829, 200), + (20260830, 200), (20260831, 200), (20260901, 200), + (20260902, 200), (20260903, 90)]: + session.add(StatDailyStaff(day=day, channel="wechat", user_id="u1", + msg_sent=sent, msg_recv=0, + active_friends=5, active_rooms=1)) + # 当日 rollup 行(active_staff 卡片读 stat_daily_staff,必须有当日行) + session.add(StatDailyStaff(day=20260904, channel="wechat", user_id="u1", + msg_sent=15, msg_recv=7, + active_friends=1, active_rooms=1)) + # 当日 + 首沟通:t1 当日首现(首沟通=1),t2 早在 20260827 就有(不算) + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t1", day=20260904, + send_cnt=10, recv_cnt=5, last_ts=1788500000000, + last_is_send=True)) + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t2", day=20260827, + send_cnt=3, recv_cnt=0, last_ts=1787800000000, + last_is_send=True)) + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="room", peer_id="r1", day=20260904, + send_cnt=7, recv_cnt=0, last_ts=1788500000000, + last_is_send=True)) + # 通话时间戳必须是 2026-09-04 当日(1756950000000 是 2025 年——坑) + session.add(PhoneCall(call_id="c1", channel="voice_call", + start_call_time=1788500000000, call_duration=60)) + await session.commit() + + +async def test_overview_today(make_client, monkeypatch): + monkeypatch.setattr(metrics, "today_key", lambda: _TODAY) + async def seed(session): + await _seed(session) + out = await metrics.overview(session, channel="wechat", scope="today") + assert out["cards"]["msg_total"] == 22 # (10+5)+(7+0) + assert out["cards"]["active_customers"] == 1 # peer_kind=friend 去重,群不计 + assert out["cards"]["active_staff"] == 1 + assert out["cards"]["call_cnt"] == 1 + assert out["cards"]["first_contact"] == 1 # t1 全史 MIN(day)=今日;t2 不是 + assert out["cards"].get("mom_total_delta") is None # 今日无环比 + + await make_client(seed) + + +async def test_anomaly_yesterday_vs_prev7(make_client, monkeypatch): + monkeypatch.setattr(metrics, "today_key", lambda: _TODAY) + async def seed(session): + await _seed(session) + a = await metrics.anomaly(session, channel="wechat") + assert a is not None + assert a["total_delta_pct"] <= -0.5 # 90 vs 均值 200 = -55%(严格 >50% 触发) + staff = {s["user_id"]: s for s in a["staff"]} + assert staff["u1"]["yesterday"] == 90 + assert staff["u1"]["prev7_mean"] == 200.0 + + await make_client(seed) + + +async def test_overview_fallback_when_rollup_empty(make_client, monkeypatch): + """stat_daily_staff 空表 → 员工卡片回退实时聚合 msg_daily_stats(部署即用)。""" + monkeypatch.setattr(metrics, "today_key", lambda: _TODAY) + async def seed(session): + session.add(WechatAccount(wx_id="wx1", user_id="u1")) + session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="t1", day=20260904, + send_cnt=2, recv_cnt=1, last_ts=1788500000000, + last_is_send=True)) + await session.commit() + out = await metrics.overview(session, channel="wechat", scope="today") + assert out["cards"]["active_staff"] == 1 + assert out["cards"]["msg_total"] == 3 + + await make_client(seed)