feat(analytics): stat_daily_staff 员工日汇总维护层——rollup派生/幂等/判别法

Co-Authored-By: Claude Code <noreply@anthropic.com>

yunke-server/app/storage/staff_summary.py

@@ -0,0 +1,131 @@
+"""stat_daily_staff(员工×日)维护层——rollup-of-rollup,不挂写钩子。
+
+刷新策略(spec §3.1):4h 调度刷当日+滚动 3 天(staff_summary.rolling_days),
+夜间 full_sync 全量自愈(recompute_all),部署回填 CLI 同款。空表时读端点
+回退直接聚合 msg_daily_stats(毫秒级)——部署即用,沿用 sessions 的
+has_rows 回退模式。
+"""
+from __future__ import annotations
+
+import datetime as dt
+import logging
+from collections.abc import Iterable
+from zoneinfo import ZoneInfo
+
+import sqlalchemy as sa
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models import MsgDailyStats, StatDailyStaff, WechatAccount, WhatsappAccount
+
+logger = logging.getLogger(__name__)
+
+_SH = ZoneInfo("Asia/Shanghai")
+
+
+async def has_rows(session: AsyncSession) -> bool:
+    """汇总表是否已有数据(读端点空表守卫;表不存在与空表同视)。"""
+    try:
+        return bool(
+            (
+                await session.execute(
+                    sa.select(sa.func.count()).select_from(StatDailyStaff)
+                )
+            ).scalar_one()
+        )
+    except (sa.exc.OperationalError, sa.exc.ProgrammingError):
+        return False
+
+
+async def account_map(session: AsyncSession, channel: str) -> dict[str, str]:
+    """owner_account_id → user_id(按渠道选账号表;映射不上的不在 dict)。"""
+    if channel == "whatsapp":
+        stmt = sa.select(WhatsappAccount.uid, WhatsappAccount.user_id).where(
+            WhatsappAccount.user_id.is_not(None)
+        )
+    else:
+        stmt = sa.select(WechatAccount.wx_id, WechatAccount.user_id).where(
+            WechatAccount.user_id.is_not(None)
+        )
+    return {owner: uid for owner, uid in (await session.execute(stmt)).all() if uid}
+
+
+def rolling_days(today: int, window: int = 3) -> list[int]:
+    """当日 + 前 window-1 天的 day 键(4h job 的滚动刷新窗)。"""
+    d = _day_from_key(today)
+    return [_key_from_day(d - dt.timedelta(days=i)) for i in range(window - 1, -1, -1)]
+
+
+def _day_from_key(day: int) -> dt.date:
+    return dt.date(day // 10000, (day // 100) % 100, day % 100)
+
+
+def _key_from_day(d: dt.date) -> int:
+    return d.year * 10000 + d.month * 100 + d.day
+
+
+async def recompute_days(
+    session: AsyncSession, days: Iterable[int], channels: tuple[str, ...] = ("wechat", "whatsapp")
+) -> int:
+    """重算给定 day×channel 的员工行:删旧插新,幂等。
+
+    返回**有数据行的 day×channel 组合数**(空渠道不计——与测试断言的
+    「天数」语义一致,seed 只有 wechat 日行时返回 1)。
+    单日日行仅数百~两千,拉回 Python 侧按 user_id 聚合——COUNT(DISTINCT
+    peer) 必须在「账号合并到员工」之后算(同人多账号 distinct peer 跨账号
+    去重),写 SQL 反而绕且踩 only_full_group_by。
+    """
+    n = 0
+    for day in sorted(set(days)):
+        for channel in channels:
+            amap = await account_map(session, channel)
+            rows = (
+                await session.execute(
+                    sa.select(
+                        MsgDailyStats.owner_account_id,
+                        MsgDailyStats.peer_kind,
+                        MsgDailyStats.peer_id,
+                        MsgDailyStats.send_cnt,
+                        MsgDailyStats.recv_cnt,
+                    ).where(
+                        MsgDailyStats.day == day, MsgDailyStats.channel == channel
+                    )
+                )
+            ).all()
+            await session.execute(
+                sa.delete(StatDailyStaff).where(
+                    StatDailyStaff.day == day, StatDailyStaff.channel == channel
+                )
+            )
+            if not rows:
+                continue
+            agg: dict[str, dict] = {}
+            for owner, kind, peer, s, r in rows:
+                uid = amap.get(owner, owner)  # 映射不上按 owner 原值(读侧标注)
+                a = agg.setdefault(
+                    uid, {"sent": 0, "recv": 0, "friends": set(), "rooms": set()}
+                )
+                a["sent"] += int(s or 0)
+                a["recv"] += int(r or 0)
+                (a["friends"] if kind == "friend" else a["rooms"]).add(peer)
+            for uid, a in sorted(agg.items()):
+                session.add(
+                    StatDailyStaff(
+                        day=day,
+                        channel=channel,
+                        user_id=uid,
+                        msg_sent=a["sent"],
+                        msg_recv=a["recv"],
+                        active_friends=len(a["friends"]),
+                        active_rooms=len(a["rooms"]),
+                    )
+                )
+            n += 1
+    return n
+
+
+async def recompute_all(session: AsyncSession) -> int:
+    """全量重建(夜间自愈 / 部署回填,幂等)。"""
+    days = (
+        await session.execute(sa.select(MsgDailyStats.day).distinct())
+    ).scalars().all()
+    return await recompute_days(session, list(days))

yunke-server/app/storage/summary.py

@@ -249,3 +249,7 @@ async def recompute_all(session: AsyncSession) -> int:
         if total_pairs % 500 == 0:
             await session.flush()
     return total_pairs
+
+
+# 公共别名(2026-09-04):analytics 层复用日界计算,避免各处复制时区逻辑。
+day_bounds_ms = _day_bounds_ms

yunke-server/tests/test_staff_summary.py

@@ -0,0 +1,116 @@
+"""stat_daily_staff 维护层:派生/幂等/判别法/未映射兜底。
+
+判别法(9-03 先例):rollup 只读 msg_daily_stats——回填后删光 message 明细,
+员工汇总重算仍出完整数据。
+"""
+from __future__ import annotations
+
+import datetime as dt
+
+from app.db.models import Message, MsgDailyStats, StatDailyStaff, WechatAccount
+from app.storage import staff_summary
+
+
+def _mk_msg(msg_id: str, ts: int, is_send: bool | None) -> Message:
+    return Message(
+        msg_id=msg_id,
+        channel="wechat",
+        owner_account_id="wx1",
+        talker_id="t1",
+        timestamp=ts,
+        is_send=is_send,
+        text="x",
+        fetched_at=dt.datetime(2026, 9, 4, 12, 0),
+    )
+
+
+async def _seed_pair(session) -> None:
+    """两账号同一员工(wx1/wx2→u1) + 未映射账号 wx3,20260904 各有消息。"""
+    session.add(WechatAccount(wx_id="wx1", user_id="u1", user_name="张三"))
+    session.add(WechatAccount(wx_id="wx2", user_id="u1", user_name="张三"))
+    rows = [
+        MsgDailyStats(channel="wechat", owner_account_id="wx1", peer_kind="friend",
+                      peer_id="t1", day=20260904, send_cnt=10, recv_cnt=5,
+                      last_ts=1756900000000, last_is_send=True),
+        MsgDailyStats(channel="wechat", owner_account_id="wx1", peer_kind="room",
+                      peer_id="r1", day=20260904, send_cnt=3, recv_cnt=0,
+                      last_ts=1756900000000, last_is_send=True),
+        MsgDailyStats(channel="wechat", owner_account_id="wx2", peer_kind="friend",
+                      peer_id="t1", day=20260904, send_cnt=2, recv_cnt=1,
+                      last_ts=1756900001000, last_is_send=False),
+        MsgDailyStats(channel="wechat", owner_account_id="wx3", peer_kind="friend",
+                      peer_id="t9", day=20260904, send_cnt=1, recv_cnt=0,
+                      last_ts=1756900002000, last_is_send=True),
+    ]
+    session.add_all(rows)
+    await session.commit()
+
+
+async def test_recompute_aggregates_by_user(make_client):
+    """两账号同人合并(sent=10+3+2=15,recv=5+0+1=6;t1 在 wx1/wx2 都聊 → 合并后 distinct friend={t1} + room={r1});wx3 未映射按原值。"""
+    async def seed(session):
+        await _seed_pair(session)
+        await staff_summary.recompute_days(session, [20260904])
+        await session.commit()
+        from sqlalchemy import select
+
+        rows = (await session.execute(select(StatDailyStaff))).scalars().all()
+        by_uid = {r.user_id: r for r in rows}
+        assert set(by_uid) == {"u1", "wx3"}  # wx3 映射不上按 owner 原值
+        u1 = by_uid["u1"]
+        assert u1.msg_sent == 15 and u1.msg_recv == 6
+        assert u1.active_friends == 1  # t1 跨账号去重
+        assert u1.active_rooms == 1
+        assert by_uid["wx3"].active_friends == 1
+
+    await make_client(seed)
+
+
+async def test_recompute_idempotent(make_client):
+    async def seed(session):
+        await _seed_pair(session)
+        await staff_summary.recompute_days(session, [20260904])
+        await staff_summary.recompute_days(session, [20260904])  # 再跑不翻倍
+        await session.commit()
+        from sqlalchemy import func, select
+
+        n = (
+            await session.execute(
+                select(func.count()).select_from(StatDailyStaff)
+            )
+        ).scalar_one()
+        assert n == 2  # u1 + wx3 各一行
+
+    await make_client(seed)
+
+
+async def test_discriminant_survives_message_delete(make_client):
+    """判别法:rollup 回填后删光 message,重算结果不变(只依赖 msg_daily_stats)。"""
+    async def seed(session):
+        await _seed_pair(session)
+        session.add(_mk_msg("m1", 1756900000000, True))
+        await session.commit()
+        await staff_summary.recompute_days(session, [20260904])
+        await session.commit()
+        from sqlalchemy import delete, select
+
+        await session.execute(delete(Message))
+        await staff_summary.recompute_days(session, [20260904])  # 自愈重算
+        await session.commit()
+        rows = (await session.execute(select(StatDailyStaff))).scalars().all()
+        by_uid = {r.user_id: r for r in rows}
+        assert by_uid["u1"].msg_sent == 15  # 数据不丢
+
+    await make_client(seed)
+
+
+async def test_recompute_all_and_rolling(make_client):
+    async def seed(session):
+        await _seed_pair(session)
+        n = await staff_summary.recompute_all(session)
+        await session.commit()
+        assert n == 1  # 全库只有一个「有数据的 day×channel」
+
+    await make_client(seed)
+    days = staff_summary.rolling_days(20260904, window=3)
+    assert days == [20260902, 20260903, 20260904]