From 5b6a9bd35b35246c4ccab114ebdda06c6bc4d79b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Fri, 4 Sep 2026 16:28:39 +0800 Subject: [PATCH] =?UTF-8?q?fix(analytics):=20recompute=5Fall=20=E5=85=A8?= =?UTF-8?q?=E8=A1=A8=E6=B8=85=E7=A9=BA=E8=87=AA=E6=84=88+whatsapp/?= =?UTF-8?q?=E5=A4=9A=E6=97=A5/=E7=A9=BA=E6=97=A5=E6=B5=8B=E8=AF=95+?= =?UTF-8?q?=E6=98=A0=E5=B0=84=E7=BC=93=E5=AD=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- yunke-server/app/storage/staff_summary.py | 17 ++++- yunke-server/tests/test_staff_summary.py | 93 ++++++++++++++++++++++- 2 files changed, 106 insertions(+), 4 deletions(-) diff --git a/yunke-server/app/storage/staff_summary.py b/yunke-server/app/storage/staff_summary.py index 221972d..53e6347 100644 --- a/yunke-server/app/storage/staff_summary.py +++ b/yunke-server/app/storage/staff_summary.py @@ -75,9 +75,12 @@ async def recompute_days( 去重),写 SQL 反而绕且踩 only_full_group_by。 """ n = 0 + # 映射按渠道取一次缓存(而非 day×channel 每轮重查)——全量重建时 + # ~600 次重复小表全查收敛为每渠道 1 次,且整个重建用同一份映射快照。 + maps = {ch: await account_map(session, ch) for ch in channels} for day in sorted(set(days)): for channel in channels: - amap = await account_map(session, channel) + amap = maps[channel] rows = ( await session.execute( sa.select( @@ -106,7 +109,10 @@ async def recompute_days( ) a["sent"] += int(s or 0) a["recv"] += int(r or 0) - (a["friends"] if kind == "friend" else a["rooms"]).add(peer) + if kind == "friend": # 显式分支:未来新 kind 跳过,不误入 rooms + a["friends"].add(peer) + elif kind == "room": + a["rooms"].add(peer) for uid, a in sorted(agg.items()): session.add( StatDailyStaff( @@ -124,7 +130,12 @@ async def recompute_days( async def recompute_all(session: AsyncSession) -> int: - """全量重建(夜间自愈 / 部署回填,幂等)。""" + """全量重建(夜间自愈 / 部署回填,幂等)。 + + 一次清空保证 msg_daily_stats 中已消失的日期不留陈旧行(夜间 + summary.recompute_all 清空重建 msg_daily_stats 后,本层同步自愈)。 + """ + await session.execute(sa.delete(StatDailyStaff)) days = ( await session.execute(sa.select(MsgDailyStats.day).distinct()) ).scalars().all() diff --git a/yunke-server/tests/test_staff_summary.py b/yunke-server/tests/test_staff_summary.py index 737df7d..ab7996b 100644 --- a/yunke-server/tests/test_staff_summary.py +++ b/yunke-server/tests/test_staff_summary.py @@ -7,7 +7,13 @@ from __future__ import annotations import datetime as dt -from app.db.models import Message, MsgDailyStats, StatDailyStaff, WechatAccount +from app.db.models import ( + Message, + MsgDailyStats, + StatDailyStaff, + WechatAccount, + WhatsappAccount, +) from app.storage import staff_summary @@ -46,6 +52,91 @@ async def _seed_pair(session) -> None: await session.commit() +async def _seed_wa_and_days(session) -> None: + """wechat 20260903/20260904 两天 + whatsapp 20260904 一天(uid→wa_u1)。""" + session.add(WechatAccount(wx_id="wx1", user_id="u1", user_name="张三")) + session.add(WhatsappAccount(uid="wa1@x", user_id="wa_u1", user_name="李四")) + rows = [ + MsgDailyStats(channel="wechat", owner_account_id="wx1", peer_kind="friend", + peer_id="t1", day=20260903, send_cnt=4, recv_cnt=2, + last_ts=1756800000000, last_is_send=True), + 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="whatsapp", owner_account_id="wa1@x", peer_kind="friend", + peer_id="p1", day=20260904, send_cnt=7, recv_cnt=3, + last_ts=1756900005000, last_is_send=True), + ] + session.add_all(rows) + await session.commit() + + +async def test_recompute_multi_day_and_whatsapp(make_client): + """行按 (day, channel, user_id) 正确分键:wechat 两天各一行 + whatsapp 一行; + WA 走 WhatsappAccount.uid→user_id 映射(而非 wechat 账号表)。""" + async def seed(session): + await _seed_wa_and_days(session) + n = await staff_summary.recompute_days(session, [20260903, 20260904]) + await session.commit() + assert n == 3 # wechat×2天 + whatsapp×1天(空 day×channel 不计) + from sqlalchemy import select + + rows = (await session.execute(select(StatDailyStaff))).scalars().all() + keys = {(r.day, r.channel, r.user_id) for r in rows} + assert keys == { + (20260903, "wechat", "u1"), + (20260904, "wechat", "u1"), + (20260904, "whatsapp", "wa_u1"), + } + wa = next(r for r in rows if r.channel == "whatsapp") + assert wa.msg_sent == 7 and wa.msg_recv == 3 and wa.active_friends == 1 + + await make_client(seed) + + +async def test_stale_day_cleanup(make_client): + """陈旧行清理:msg_daily_stats 里已消失的日期,重算后 stat 行必须消失。 + + - recompute_all 全表清空自愈:day 从 msg_daily_stats 整体消失(夜间 + summary.recompute_all 清空重建的典型场景)→ distinct days 不含它, + 只有全表清空能救; + - recompute_days 空日 delete-before-continue:day 仍在但某渠道行删光 + → 该 day×channel 的 stat 行删除,同日 wechat 行保留。 + """ + async def seed(session): + from sqlalchemy import delete, select + + await _seed_wa_and_days(session) + await staff_summary.recompute_all(session) # stat 先有 3 行 + await session.commit() + + # 场景 1:20260903 整天从 msg_daily_stats 消失 → recompute_all 自愈 + await session.execute( + delete(MsgDailyStats).where(MsgDailyStats.day == 20260903) + ) + n = await staff_summary.recompute_all(session) + await session.commit() + assert n == 2 # 只剩 20260904 的 wechat + whatsapp 两个 day×channel + days = set( + (await session.execute(select(StatDailyStaff.day))).scalars().all() + ) + assert days == {20260904} # 20260903 陈旧行被清 + + # 场景 2:20260904 的 whatsapp 行删光 → 空日重算删行,wechat 保留 + await session.execute( + delete(MsgDailyStats).where(MsgDailyStats.channel == "whatsapp") + ) + await staff_summary.recompute_days(session, [20260904]) + await session.commit() + keys = { + (r.day, r.channel) + for r in (await session.execute(select(StatDailyStaff))).scalars().all() + } + assert keys == {(20260904, "wechat")} # whatsapp 行消失,wechat 不动 + + await make_client(seed) + + 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):