fix(analytics): recompute_all 全表清空自愈+whatsapp/多日/空日测试+映射缓存

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-09-04 16:28:39 +08:00
parent e3bb97bdf3
commit 5b6a9bd35b
2 changed files with 106 additions and 4 deletions
+14 -3
View File
@@ -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()
+92 -1
View File
@@ -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):