diff --git a/yunke-server/app/storage/staff_summary.py b/yunke-server/app/storage/staff_summary.py new file mode 100644 index 0000000..221972d --- /dev/null +++ b/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)) diff --git a/yunke-server/app/storage/summary.py b/yunke-server/app/storage/summary.py index a0b4f05..2bef64e 100644 --- a/yunke-server/app/storage/summary.py +++ b/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 diff --git a/yunke-server/tests/test_staff_summary.py b/yunke-server/tests/test_staff_summary.py new file mode 100644 index 0000000..737df7d --- /dev/null +++ b/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]