diff --git a/.gitignore b/.gitignore index d19070e..e0c7c71 100644 --- a/.gitignore +++ b/.gitignore @@ -26,3 +26,6 @@ media/ *.sqlite *.sqlite3 .playwright-mcp/ + +# 部署过程产生的 git bundle 临时包 +*.bundle diff --git a/yunke-server/app/api/routers/sessions.py b/yunke-server/app/api/routers/sessions.py index ab1d6ef..ff90bec 100644 --- a/yunke-server/app/api/routers/sessions.py +++ b/yunke-server/app/api/routers/sessions.py @@ -1,10 +1,17 @@ -"""GET /api/sessions — 账号会话列表(好友+群,今日消息数+最近联系时间),全本地 message 聚合。 +"""GET /api/sessions — 账号会话列表(好友+群,今日消息数+最近联系时间)。 -镜像 staff_chat_stats 聚合模式(GROUP BY + case 计数 + Python 侧合并昵称/头像)。 -today_start 钉死 Asia/Shanghai(UTC+8)当日 0 点,不依赖服务器时区。 +**双路径(2026-09-03 会话列表超时修复)**: +- ``msg_daily_stats`` 有行 → 读日粒度汇总(每会话「今日行 + MAX(last_ts)」, + 毫秒级;旧实时全量聚合在重账号上实测 3~5s,128MB buffer pool 下更被推过 + 前端 15s 超时)。 +- 空表(部署回填前守卫)→ 回退旧实时聚合,行为不变。 + +today_start 钉死 Asia/Shanghai(UTC+8)当日 0 点,不依赖服务器时区;汇总层 +``day_key`` 同口径。 """ from __future__ import annotations +from collections import namedtuple from datetime import datetime from typing import Annotated from zoneinfo import ZoneInfo @@ -15,7 +22,8 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_session from app.api.schemas import SessionListOut, SessionOut -from app.db.models import ChatGroup, Customer, Message +from app.db.models import ChatGroup, Customer, Message, MsgDailyStats +from app.storage.summary import day_key, has_rows as _summary_has_rows router = APIRouter(prefix="/api", tags=["sessions"]) @@ -63,6 +71,51 @@ async def _group_rows(session: AsyncSession, owner: str, channel: str, today_sta return (await session.execute(stmt)).all() +# ── 汇总路径:msg_daily_stats 日行(namedtuple 与旧 Row 同形,端点主体零改动)── # + +_FriendRow = namedtuple("_FriendRow", ["talker_id", "last_chat_time", "today_count"]) +_GroupRow = namedtuple("_GroupRow", ["room_id", "last_chat_time", "today_count"]) + + +async def _summary_rows( + session: AsyncSession, owner: str, channel: str, kind: str, today: int +) -> dict[str, tuple[int | None, int]]: + """peer → (last_chat_time, today_count),来自日行聚合。 + + last = MAX(last_ts)(日行内时间单调,跨日最大即全史最后一条); + today = 当日行的 send+recv。 + """ + base = ( + MsgDailyStats.channel == channel, + MsgDailyStats.owner_account_id == owner, + MsgDailyStats.peer_kind == kind, + ) + last = { + peer: last_ts + for peer, last_ts in ( + await session.execute( + sa.select(MsgDailyStats.peer_id, sa.func.max(MsgDailyStats.last_ts)) + .where(*base) + .group_by(MsgDailyStats.peer_id) + ) + ).all() + } + today_counts = { + peer: int(cnt or 0) + for peer, cnt in ( + await session.execute( + sa.select( + MsgDailyStats.peer_id, + sa.func.coalesce(sa.func.sum(MsgDailyStats.send_cnt + MsgDailyStats.recv_cnt), 0), + ) + .where(*base, MsgDailyStats.day == today) + .group_by(MsgDailyStats.peer_id) + ) + ).all() + } + return {peer: (last_ts, today_counts.get(peer, 0)) for peer, last_ts in last.items()} + + async def _meta(session: AsyncSession, owner: str, channel: str, friend_ids: list[str], room_ids: list[str]): """查 customer/chat_group 表合并昵称/头像;好友行额外带 delete_flag(is_del 过滤用)。 @@ -105,8 +158,14 @@ async def list_sessions( ``pinned``:filehelper 会话置顶标记(对齐云客置顶区),其余好友/群恒 False。 """ today_start = _today_start_ms() - f_rows = await _friend_rows(session, owner_account_id, channel, today_start) - g_rows = await _group_rows(session, owner_account_id, channel, today_start) + if await _summary_has_rows(session): + f_map = await _summary_rows(session, owner_account_id, channel, "friend", day_key(today_start)) + g_map = await _summary_rows(session, owner_account_id, channel, "room", day_key(today_start)) + f_rows = [_FriendRow(talker_id=p, last_chat_time=t, today_count=c) for p, (t, c) in f_map.items()] + g_rows = [_GroupRow(room_id=p, last_chat_time=t, today_count=c) for p, (t, c) in g_map.items()] + else: + f_rows = await _friend_rows(session, owner_account_id, channel, today_start) + g_rows = await _group_rows(session, owner_account_id, channel, today_start) fnames, gnames = await _meta( session, owner_account_id, channel, [r.talker_id for r in f_rows], [r.room_id for r in g_rows] ) diff --git a/yunke-server/app/api/routers/staff_chat_stats.py b/yunke-server/app/api/routers/staff_chat_stats.py index cdbbf50..a02b1f4 100644 --- a/yunke-server/app/api/routers/staff_chat_stats.py +++ b/yunke-server/app/api/routers/staff_chat_stats.py @@ -1,12 +1,17 @@ -"""GET /api/staff-chat-stats — 微信员工 5 数字 + 客户明细（本地 message 表实时聚合）。 +"""GET /api/staff-chat-stats — 微信员工 5 数字 + 客户明细。 + +**双路径(2026-09-03 全时段窗 6.9s 超时修复)**:``msg_daily_stats`` 有行时 +读日粒度汇总(窗口对齐到自然日,毫秒级);空表回退旧实时聚合,行为不变。 +窗口语义随之统一为「上海自然日」(旧 SQL 日期函数跟随 DB 会话时区,UTC +容器下与今日数口径割裂——顺带修复)。 数据全本地：按 员工微信 wxid × ms-epoch 日期窗 聚合 1:1 私聊（room_id 过滤）， 不访问云客。5 汇总数从 customers 在 Python 侧派生（spec §5）。 -跨库日期提取（codebase 首例）镜像 app/storage/repos.py 的方言检测写法。 """ from __future__ import annotations import time +from collections import namedtuple from typing import Annotated import sqlalchemy as sa @@ -15,7 +20,8 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_session from app.api.schemas import StaffChatCustomerOut, StaffChatStatsOut, StaffChatStatsSummary -from app.db.models import Customer, Message +from app.db.models import Customer, Message, MsgDailyStats +from app.storage.summary import day_key, has_rows as _summary_has_rows router = APIRouter(prefix="/api", tags=["staff-chat-stats"]) @@ -81,6 +87,62 @@ async def _aggregate(session: AsyncSession, owner: str, start: int, end: int) -> return agg_rows, last_map +# ── 汇总路径:msg_daily_stats 日行(namedtuple 与旧 Row 同形)────────────── # + +_AggRow = namedtuple( + "_AggRow", ["talker_id", "send_count", "receive_count", "interaction_days", "last_time"] +) + + +async def _aggregate_summary( + session: AsyncSession, owner: str, start: int, end: int +) -> tuple[list, dict[str, bool | None]]: + """窗口 → [day_key(start), day_key(end)] 自然日区间,按日行 Python 聚合。 + + interaction_days = 窗内有消息的自然日数;last_time/last_is_send 取窗内 + 最后一条(last_ts 最大的日行)。窗口对齐自然日是文档化的语义变化 + (前端日期选择器本就按天取值)。 + """ + rows = ( + await session.execute( + sa.select( + MsgDailyStats.peer_id, + MsgDailyStats.day, + MsgDailyStats.send_cnt, + MsgDailyStats.recv_cnt, + MsgDailyStats.last_ts, + MsgDailyStats.last_is_send, + ).where( + MsgDailyStats.channel == "wechat", + MsgDailyStats.owner_account_id == owner, + MsgDailyStats.peer_kind == "friend", + MsgDailyStats.day >= day_key(start), + MsgDailyStats.day <= day_key(end), + ) + ) + ).all() + + per: dict[str, dict] = {} + for peer, _day, send, recv, last_ts, last_is in rows: + acc = per.setdefault( + peer, {"send": 0, "recv": 0, "days": 0, "last_ts": None, "last_is": None} + ) + acc["send"] += int(send or 0) + acc["recv"] += int(recv or 0) + acc["days"] += 1 + if acc["last_ts"] is None or (last_ts is not None and last_ts >= acc["last_ts"]): + acc["last_ts"] = last_ts + acc["last_is"] = last_is + + agg_rows = [ + _AggRow(talker_id=p, send_count=a["send"], receive_count=a["recv"], + interaction_days=a["days"], last_time=a["last_ts"]) + for p, a in per.items() + ] + last_map = {p: a["last_is"] for p, a in per.items()} + return agg_rows, last_map + + async def _customer_names(session: AsyncSession, owner: str, talker_ids: list[str]) -> dict[str, tuple[str, str, str]]: if not talker_ids: return {} @@ -114,7 +176,10 @@ async def get_staff_chat_stats( if start is None: start = end - 3 * _MS_PER_DAY # 默认近 3 天 - agg_rows, last_map = await _aggregate(session, owner_account_id, start, end) + if await _summary_has_rows(session): + agg_rows, last_map = await _aggregate_summary(session, owner_account_id, start, end) + else: + agg_rows, last_map = await _aggregate(session, owner_account_id, start, end) talker_ids = [r.talker_id for r in agg_rows] names = await _customer_names(session, owner_account_id, talker_ids) diff --git a/yunke-server/app/db/models.py b/yunke-server/app/db/models.py index b4a7b56..053b311 100644 --- a/yunke-server/app/db/models.py +++ b/yunke-server/app/db/models.py @@ -740,6 +740,46 @@ class SyncRun(Base): __table_args__ = (Index("ix_sync_run_entity_status", "entity_type", "status"),) +class MsgDailyStats(Base): + """日粒度会话统计:每个「渠道×账号×会话×自然日」一行(2026-09-03)。 + + sessions(staff-chat 会话列表)与 staff-chat-stats(5 数字弹窗)的预聚 + 合层——这两端点原先每次请求对 owner 全量 message 做 GROUP BY(重账号 + 5.2 万条实测 3~7s,128MB buffer pool 下更被推过前端 15s 超时)。日行由 + 写路径维护:``repos.upsert_messages`` 落库后自动重算本批触及的 + 「会话×天」;夜间 ``recompute_all`` 全量重建自愈(裸 UPDATE/回填脚本 + 等绕过钩子的写入由它兜底)。 + + 口径: + - ``day`` 一律 Asia/Shanghai 自然日(Python 计算,不用 SQL 日期函数—— + from_unixtime 跟随 MySQL 会话时区,UTC 容器下与今日数口径割裂)。 + - ``send_cnt``/``recv_cnt`` 仅计 is_send 真/假;NULL 两边都不计(对齐 + 旧聚合的 case 语义)。 + - ``last_ts``/``last_is_send`` 为该日最后一条消息的时间与方向。 + """ + + __tablename__ = "msg_daily_stats" + + channel: Mapped[str] = mapped_column(String(16), primary_key=True) + owner_account_id: Mapped[str] = mapped_column(String(128), primary_key=True) + # friend = 1:1 私聊(peer=talker_id) / room = 群(peer=room_id) + peer_kind: Mapped[str] = mapped_column(String(8), primary_key=True) + peer_id: Mapped[str] = mapped_column(String(128), primary_key=True) + # yyyymmdd,如 20260903(上海自然日) + day: Mapped[int] = mapped_column(Integer, primary_key=True) + send_cnt: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default=sa.text("0") + ) + recv_cnt: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default=sa.text("0") + ) + # 当日最后一条消息(ms);当天无有效方向计数的消息也存在(is_send=NULL) + last_ts: Mapped[int | None] = mapped_column(sa.BigInteger, nullable=True) + last_is_send: Mapped[bool | None] = mapped_column(Boolean, nullable=True) + + updated_at: Mapped[dt.datetime] = _updated_at() + + __all__ = [ "ChatGroup", "Customer", @@ -748,6 +788,7 @@ __all__ = [ "Media", "Message", "MomentsPost", + "MsgDailyStats", "PhoneCall", "Sms", "Staff", diff --git a/yunke-server/app/storage/repos.py b/yunke-server/app/storage/repos.py index 35733d0..0421bde 100644 --- a/yunke-server/app/storage/repos.py +++ b/yunke-server/app/storage/repos.py @@ -53,6 +53,7 @@ that one table. Multi-table routing is a LATER task (YAGNI for M1). from __future__ import annotations +import logging from typing import Any import sqlalchemy as sa @@ -74,6 +75,9 @@ from app.db.models import ( WechatAccount, WhatsappAccount, ) +from app.storage.summary import recompute_days, touched_day_keys + +_logger = logging.getLogger(__name__) __all__ = [ "upsert_chat_group", @@ -409,6 +413,20 @@ async def upsert_messages(session: AsyncSession, recs: list[dict[str, Any]]) -> """ for rec in recs: await _upsert_one(session, Message, rec) + + # 落库后维护日粒度汇总(msg_daily_stats):只重算本批触及的「会话×天」, + # 窄时间范围查询毫秒级。所有消息写入路径(页面抓取器微信/WA/群、CLI + # 手动同步)都汇到本函数,一处挂全。失败不阻塞消息写入——夜间 + # recompute_all 自愈兜底(2026-09-03 会话列表超时修复)。 + keys = touched_day_keys(recs) + if keys: + try: + await recompute_days(session, keys) + except Exception: + _logger.exception( + "daily-stats recompute failed (%d keys); nightly rebuild will heal", len(keys) + ) + return len(recs) diff --git a/yunke-server/app/storage/summary.py b/yunke-server/app/storage/summary.py new file mode 100644 index 0000000..6209f02 --- /dev/null +++ b/yunke-server/app/storage/summary.py @@ -0,0 +1,245 @@ +"""日粒度会话汇总(msg_daily_stats)的维护层。 + +设计要点(2026-09-03 会话列表超时修复): + +- **day 口径**:Asia/Shanghai 自然日,:func:`day_key` 纯 Python 计算—— + 不用 SQL 日期函数(from_unixtime 跟随 MySQL 会话时区,UTC 容器下与 + sessions 的上海零点口径割裂;SQLite/MySQL 方言也各自不同)。 +- **写路径钩子**::func:`repos.upsert_messages` 落库后调 + :func:`recompute_days` 重算本批触及的「会话×天」(窄时间范围查询, + 毫秒级)。页面抓取器/WA/群消息/CLI 手动同步全部路径最终都走 + upsert_messages,一处挂全。 +- **全量重建**::func:`recompute_all` 清空后逐会话重算——夜间自愈 + (02:00 full_sync_all.sh 追加一步)兼部署回填,幂等。 + +计数语义与旧实时聚合完全一致:is_send=True 计 send、False 计 recv、 +NULL 两边不计;timestamp 为 NULL 的消息不参与统计。 +""" + +from __future__ import annotations + +import datetime as dt +import logging +from zoneinfo import ZoneInfo + +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import Message, MsgDailyStats + +logger = logging.getLogger(__name__) + +_SH = ZoneInfo("Asia/Shanghai") + + +async def has_rows(session: AsyncSession) -> bool: + """汇总表是否已有数据(读端点的空表守卫:部署回填前自动回退旧实时聚合)。""" + return bool( + ( + await session.execute( + sa.select(sa.func.count()).select_from(MsgDailyStats) + ) + ).scalar_one() + ) + + +def day_key(ts_ms: int) -> int: + """ms-epoch → 上海自然日 yyyymmdd。""" + d = dt.datetime.fromtimestamp(ts_ms / 1000, tz=_SH) + return d.year * 10000 + d.month * 100 + d.day + + +def _day_bounds_ms(day: int) -> tuple[int, int]: + """上海自然日 → [当日 00:00, 次日 00:00) 的 ms 左闭右开区间。""" + y, m, d = day // 10000, (day // 100) % 100, day % 100 + start = dt.datetime(y, m, d, tzinfo=_SH) + return int(start.timestamp() * 1000), int( + (start + dt.timedelta(days=1)).timestamp() * 1000 + ) + + +def _kind_peer(rec: dict) -> tuple[str, str] | None: + """消息记录 → (peer_kind, peer_id);room 优先(群消息按群归账)。""" + room = rec.get("room_id") + if room: + return ("room", str(room)) + talker = rec.get("talker_id") + if talker: + return ("friend", str(talker)) + return None + + +def touched_day_keys( + recs: list[dict], +) -> set[tuple[str, str, str, str, int]]: + """从一批消息记录提取要重算的 (channel, owner, kind, peer, day) 集合。""" + out: set[tuple[str, str, str, str, int]] = set() + for rec in recs: + ts = rec.get("timestamp") + if not ts: + continue + kp = _kind_peer(rec) + if kp is None: + continue + out.add( + ( + str(rec.get("channel") or ""), + str(rec.get("owner_account_id") or ""), + kp[0], + kp[1], + day_key(int(ts)), + ) + ) + return out + + +def _peer_filter(kind: str, peer: str) -> list: + """kind/peer → message 表过滤条件(与 sessions/staff-chat-stats 口径一致)。""" + if kind == "room": + return [ + Message.room_id == peer, + Message.room_id.is_not(None), + Message.room_id != "", + ] + return [ + Message.talker_id == peer, + sa.or_(Message.room_id.is_(None), Message.room_id == ""), + ] + + +async def recompute_days( + session: AsyncSession, + keys: set[tuple[str, str, str, str, int]] | list[tuple[str, str, str, str, int]], +) -> int: + """重算给定 (channel, owner, kind, peer, day) 组合的统计行。 + + 幂等:先删旧行再按当前 message 真值重建(0 条消息的日子删行不留); + 同批消息跨页重放自然收敛。查询是窄时间范围(单会话单日),毫秒级。 + """ + n = 0 + for channel, owner, kind, peer, day in sorted(keys): + lo, hi = _day_bounds_ms(day) + rows = ( + await session.execute( + sa.select(Message.timestamp, Message.is_send).where( + Message.channel == channel, + Message.owner_account_id == owner, + *_peer_filter(kind, peer), + Message.timestamp.is_not(None), + Message.timestamp >= lo, + Message.timestamp < hi, + ) + ) + ).all() + send = recv = 0 + last_ts: int | None = None + last_is_send: bool | None = None + for ts, is_send in rows: + if is_send is True: + send += 1 + elif is_send is False: + recv += 1 + if last_ts is None or ts >= last_ts: + last_ts = int(ts) + last_is_send = is_send + await session.execute( + sa.delete(MsgDailyStats).where( + MsgDailyStats.channel == channel, + MsgDailyStats.owner_account_id == owner, + MsgDailyStats.peer_kind == kind, + MsgDailyStats.peer_id == peer, + MsgDailyStats.day == day, + ) + ) + if rows: + session.add( + MsgDailyStats( + channel=channel, + owner_account_id=owner, + peer_kind=kind, + peer_id=peer, + day=day, + send_cnt=send, + recv_cnt=recv, + last_ts=last_ts, + last_is_send=last_is_send, + ) + ) + n += 1 + return n + + +async def recompute_all(session: AsyncSession) -> int: + """全量重建 msg_daily_stats(夜间自愈 / 部署回填)。 + + 清空后按会话分组拉取全历史 (timestamp, is_send) 在 Python 侧按上海日 + 分桶——一次清空保证绕过钩子的写入(裸 UPDATE/回填脚本)也能被纠正。 + 返回重算的会话数。夜间 02:00 窗口跑,全库 86 万行量级预计分钟级。 + """ + await session.execute(sa.delete(MsgDailyStats)) + pair_rows = ( + await session.execute( + sa.select( + Message.channel, + Message.owner_account_id, + Message.room_id, + Message.talker_id, + ) + .where(Message.timestamp.is_not(None)) + .distinct() + ) + ).all() + + # 四列 DISTINCT 后同一群会按成员 talker 出现多行 → 归并成 (channel, + # owner, kind, peer) 去重,否则同主键日行会插两遍(IntegrityError)。 + pairs: set[tuple[str, str, str, str]] = set() + for channel, owner, room, talker in pair_rows: + if room: + pairs.add((channel, owner, "room", str(room))) + elif talker: + pairs.add((channel, owner, "friend", str(talker))) + + total_pairs = 0 + for channel, owner, kind, peer in sorted(pairs): + # 拉全会话消息按日重算(而非逐日窄查):全量重建时一次往返更省 + rows = ( + await session.execute( + sa.select(Message.timestamp, Message.is_send).where( + Message.channel == channel, + Message.owner_account_id == owner, + *_peer_filter(kind, peer), + ) + ) + ).all() + days: dict[int, list] = {} + for ts, is_send in rows: + days.setdefault(day_key(int(ts)), []).append((ts, is_send)) + for day, day_rows in days.items(): + send = recv = 0 + last_ts = None + last_is_send = None + for ts, is_send in day_rows: + if is_send is True: + send += 1 + elif is_send is False: + recv += 1 + if last_ts is None or ts >= last_ts: + last_ts = int(ts) + last_is_send = is_send + session.add( + MsgDailyStats( + channel=channel, + owner_account_id=owner, + peer_kind=kind, + peer_id=peer, + day=day, + send_cnt=send, + recv_cnt=recv, + last_ts=last_ts, + last_is_send=last_is_send, + ) + ) + total_pairs += 1 + if total_pairs % 500 == 0: + await session.flush() + return total_pairs diff --git a/yunke-server/app/sync/cli.py b/yunke-server/app/sync/cli.py index 2d7a8fa..e856184 100644 --- a/yunke-server/app/sync/cli.py +++ b/yunke-server/app/sync/cli.py @@ -89,6 +89,18 @@ def _build_parser() -> argparse.ArgumentParser: help="Suppress per-run JSON progress lines (errors still logged).", ) + # ``daily-stats`` --------------------------------------------------------- + p_stats = sub.add_parser( + "daily-stats", + help="Rebuild msg_daily_stats from the message table.", + description=( + "Full rebuild of the daily conversation-summary table " + "(msg_daily_stats). Used as the nightly self-heal step in " + "scripts/full_sync_all.sh and as the one-off deploy backfill " + "(see docs/DEPLOY.md). Idempotent." + ), + ) + # ``scheduler`` ----------------------------------------------------------- p_sched = sub.add_parser( "scheduler", @@ -274,6 +286,22 @@ def _print_sync_summary(entity: str, results: list[Any]) -> None: print(line, flush=True) +async def _cmd_daily_stats(args: argparse.Namespace) -> int: + """全量重建 msg_daily_stats(夜间自愈 / 部署回填,幂等)。""" + settings = get_settings() + engine, session_maker = await _init_db(settings.db_url) + try: + from app.storage.summary import recompute_all + + async with session_maker() as session: + pairs = await recompute_all(session) + await session.commit() + print(f"[daily-stats] rebuilt {pairs} conversation pairs") + return 0 + finally: + await engine.dispose() + + def main(argv: list[str] | None = None) -> int: """CLI entry point. Returns a process exit code.""" # Configure logging so the scheduler's JSON line + the orchestrator's @@ -292,6 +320,8 @@ def main(argv: list[str] | None = None) -> int: return asyncio.run(_cmd_sync(args)) if args.command == "scheduler": return asyncio.run(_cmd_scheduler(args)) + if args.command == "daily-stats": + return asyncio.run(_cmd_daily_stats(args)) except KeyboardInterrupt: # So `python -m app.sync.cli scheduler` exits cleanly on Ctrl-C. return 0 diff --git a/yunke-server/migrations/versions/a7d3e9f1c2b4_msg_daily_stats_table.py b/yunke-server/migrations/versions/a7d3e9f1c2b4_msg_daily_stats_table.py new file mode 100644 index 0000000..ccd7d1e --- /dev/null +++ b/yunke-server/migrations/versions/a7d3e9f1c2b4_msg_daily_stats_table.py @@ -0,0 +1,77 @@ +"""msg_daily_stats 日粒度会话汇总表(会话列表/5数字弹窗超时修复) + +Revision ID: a7d3e9f1c2b4 +Revises: c53f90b1e6a4 +Create Date: 2026-09-03 17:00:00.000000 + +内容 +---- +sessions 与 staff-chat-stats 原先每次请求对 owner 全量 message 实时 +GROUP BY——重账号 5.2 万条消息实测 3~5s(sessions)/6.9s(全时段 +staff-chat-stats),在服务器 128MB buffer pool 下进一步被推过前端 axios +15s 超时(ECONNABORTED,2026-09-03 用户报「聊天记录页数据多时请求失败」)。 + +本表是预聚合层:每个「渠道×账号×会话×上海自然日」一行(收/发计数、当日 +最后一条时间与方向)。维护:: + +* 写路径:``repos.upsert_messages`` 落库后自动重算本批触及的「会话×天」 + (所有抓取路径——页面抓取器微信/WA/群、CLI 手动同步——都汇到该函数)。 +* 读路径:两个端点在表有行时读它(毫秒级),空表回退旧实时聚合(部署 + 回填前的守卫)。 +* 自愈:夜间 full_sync_all.sh 的 ``daily-stats`` 步骤全量重建,纠正绕过 + 钩子的写入(裸 UPDATE/回填脚本);部署回填也用同一条命令 + (``python -m app.sync.cli daily-stats``)。 + +day 为 int yyyymmdd,口径一律 Asia/Shanghai(Python 计算,不用 SQL 日期 +函数——from_unixtime 跟随 DB 会话时区,与今日数口径割裂)。 + +幂等:``sa.inspect`` 先查同名表,存在即跳过。SQLite/MySQL 通用。 +""" +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + + +# revision identifiers, used by Alembic. +revision: str = "a7d3e9f1c2b4" +down_revision: str | Sequence[str] | None = "c53f90b1e6a4" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_TABLE = "msg_daily_stats" + + +def upgrade() -> None: + bind = op.get_bind() + insp = sa.inspect(bind) + if insp.has_table(_TABLE): + return + + op.create_table( + _TABLE, + sa.Column("channel", sa.String(length=16), primary_key=True), + sa.Column("owner_account_id", sa.String(length=128), primary_key=True), + sa.Column("peer_kind", sa.String(length=8), primary_key=True), + sa.Column("peer_id", sa.String(length=128), primary_key=True), + sa.Column("day", sa.Integer(), primary_key=True), + sa.Column( + "send_cnt", sa.Integer(), nullable=False, server_default=sa.text("0") + ), + sa.Column( + "recv_cnt", sa.Integer(), nullable=False, server_default=sa.text("0") + ), + sa.Column("last_ts", sa.BigInteger(), nullable=True), + sa.Column("last_is_send", sa.Boolean(), nullable=True), + sa.Column( + "updated_at", sa.DateTime(), nullable=False, server_default=sa.func.now() + ), + ) + + +def downgrade() -> None: + bind = op.get_bind() + insp = sa.inspect(bind) + if not insp.has_table(_TABLE): + return + op.drop_table(_TABLE) diff --git a/yunke-server/scripts/full_sync_all.sh b/yunke-server/scripts/full_sync_all.sh index 4f4fbba..3c4079b 100644 --- a/yunke-server/scripts/full_sync_all.sh +++ b/yunke-server/scripts/full_sync_all.sh @@ -45,30 +45,54 @@ echo "$(date) --- 9c/10 Moments (员工朋友圈) ---" $PY scripts/scrape_via_page.py --entity moments echo "$(date) --- 10/10 Media download (audio+images) ---" $PY scripts/scrape_via_page.py --entity media +# 11:会话日汇总全量重建(夜间自愈:绕过 upsert 钩子的写入由它纠正; +# sessions/staff-chat-stats 读路径依赖此表,2026-09-03) +echo "$(date) --- 11/11 Daily-stats rebuild ---" +$PY -m app.sync.cli daily-stats echo "==============================================" echo "$(date) === FULL RE-SYNC COMPLETE ===" echo "==============================================" -$PY -c " -import sqlite3 -con = sqlite3.connect('smoke.db') -c = con.cursor() -print() -print('=== FINAL DATA SUMMARY ===') -for label, sql in [ - ('wechat accounts', \"SELECT COUNT(*) FROM wechat_account\"), - ('whatsapp accounts', \"SELECT COUNT(*) FROM whatsapp_account\"), - ('wechat customers', \"SELECT COUNT(*) FROM customer WHERE channel='wechat'\"), - ('whatsapp customers', \"SELECT COUNT(*) FROM customer WHERE channel='whatsapp'\"), - ('wechat groups', \"SELECT COUNT(*) FROM chat_group WHERE channel='wechat'\"), - ('whatsapp groups', \"SELECT COUNT(*) FROM chat_group WHERE channel='whatsapp'\"), - ('wechat messages', \"SELECT COUNT(*) FROM message WHERE channel='wechat'\"), - ('whatsapp messages', \"SELECT COUNT(*) FROM message WHERE channel='whatsapp'\"), - ('voice calls', \"SELECT COUNT(*) FROM phone_call WHERE channel IN ('voice_call','wa_voice_call')\"), - ('voice clips', 'SELECT COUNT(*) FROM voice_clip'), - ('moments posts', 'SELECT COUNT(*) FROM moments_post'), - ('media files', 'SELECT COUNT(*) FROM media'), -]: - print(f' {label:22s} {c.execute(sql).fetchone()[0]:>6}') -" +# 终局数据快照:走 app 引擎(DB_URL 可移植,SQLite/MySQL 通吃;原 sqlite3 +# 直连 smoke.db 的写法在服务器 MySQL 上会炸,2026-09-03 修) +$PY - <<'PYEOF' +import asyncio + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import create_async_engine + +from app.config import get_settings + +TABLES = [ + ("wechat accounts", "SELECT COUNT(*) FROM wechat_account"), + ("whatsapp accounts", "SELECT COUNT(*) FROM whatsapp_account"), + ("wechat customers", "SELECT COUNT(*) FROM customer WHERE channel='wechat'"), + ("whatsapp customers", "SELECT COUNT(*) FROM customer WHERE channel='whatsapp'"), + ("wechat groups", "SELECT COUNT(*) FROM chat_group WHERE channel='wechat'"), + ("whatsapp groups", "SELECT COUNT(*) FROM chat_group WHERE channel='whatsapp'"), + ("wechat messages", "SELECT COUNT(*) FROM message WHERE channel='wechat'"), + ("whatsapp messages", "SELECT COUNT(*) FROM message WHERE channel='whatsapp'"), + ("daily stats rows", "SELECT COUNT(*) FROM msg_daily_stats"), + ("voice calls", "SELECT COUNT(*) FROM phone_call WHERE channel IN ('voice_call','wa_voice_call')"), + ("voice clips", "SELECT COUNT(*) FROM voice_clip"), + ("moments posts", "SELECT COUNT(*) FROM moments_post"), + ("media files", "SELECT COUNT(*) FROM media"), +] + + +async def main() -> None: + engine = create_async_engine(get_settings().db_url) + try: + async with engine.connect() as conn: + print() + print("=== FINAL DATA SUMMARY ===") + for label, sql in TABLES: + n = (await conn.execute(text(sql))).scalar_one() + print(f" {label:22s} {n:>6}") + finally: + await engine.dispose() + + +asyncio.run(main()) +PYEOF diff --git a/yunke-server/tests/test_api_summary_paths.py b/yunke-server/tests/test_api_summary_paths.py new file mode 100644 index 0000000..03b00ea --- /dev/null +++ b/yunke-server/tests/test_api_summary_paths.py @@ -0,0 +1,113 @@ +"""sessions / staff-chat-stats 的汇总表读路径判别测试(2026-09-03)。 + +两个端点在 msg_daily_stats 有行时必须改读汇总表(毫秒级),空表时回退旧 +实时聚合(部署回填前的守卫)。判别手法:**回填汇总后把 message 表删光**—— +旧实现会返回空,只有真正读汇总表的实现还能吐出完整数据。 + +另配对拍测试:同一批种子,「无汇总(旧路径)」与「有汇总(新路径)」的 +响应必须逐字段一致。 +""" + +from __future__ import annotations + +import time + +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import ChatGroup, Customer, Message +from app.storage.summary import recompute_all + +OWNER = "wx_owner_sum" +CH = "wechat" +_NOW = int(time.time() * 1000) +_MS_DAY = 86_400_000 + + +async def _seed(session: AsyncSession) -> None: + session.add_all([ + # t1:今天 1 发 1 收,2 天前还有一条 + Message(msg_id="s1", channel=CH, owner_account_id=OWNER, talker_id="t1", room_id=None, is_send=True, timestamp=_NOW, type="text", type_raw=1, text="x"), + Message(msg_id="s2", channel=CH, owner_account_id=OWNER, talker_id="t1", room_id=None, is_send=False, timestamp=_NOW - 1000, type="text", type_raw=1, text="y"), + Message(msg_id="s3", channel=CH, owner_account_id=OWNER, talker_id="t1", room_id=None, is_send=False, timestamp=_NOW - 2 * _MS_DAY, type="text", type_raw=1, text="old"), + # t2:只有 5 天前的消息(今日 0) + Message(msg_id="s4", channel=CH, owner_account_id=OWNER, talker_id="t2", room_id=None, is_send=True, timestamp=_NOW - 5 * _MS_DAY, type="text", type_raw=1, text="o2"), + # 群:今天 2 条 + Message(msg_id="s5", channel=CH, owner_account_id=OWNER, talker_id="m1", room_id="room1@chatroom", is_send=True, timestamp=_NOW, type="text", type_raw=1, text="g1"), + Message(msg_id="s6", channel=CH, owner_account_id=OWNER, talker_id="m2", room_id="room1@chatroom", is_send=False, timestamp=_NOW - 2000, type="text", type_raw=1, text="g2"), + Customer(friend_id="t1", channel=CH, owner_account_id=OWNER, friend_nickname="Alice", friend_remark=None), + Customer(friend_id="t2", channel=CH, owner_account_id=OWNER, friend_nickname="Bob", friend_remark=None), + ChatGroup(room_id="room1@chatroom", channel=CH, owner_account_id=OWNER, room_name="群一号"), + ]) + await session.commit() + + +async def _seed_with_stats(session: AsyncSession) -> None: + await _seed(session) + await recompute_all(session) + await session.commit() + + +async def _stats_then_wipe(session: AsyncSession) -> None: + await _seed_with_stats(session) + await session.execute(sa.delete(Message)) + await session.commit() + + +# ── sessions ─────────────────────────────────────────────────────────────── # + + +async def test_sessions_served_from_summary_when_populated(make_client): + """回填后删光 message:响应仍完整 → 数据来自汇总表(旧实现此刻返回空)。""" + wiped = await make_client(_stats_then_wipe) + r = await wiped.get("/api/sessions", params={"owner_account_id": OWNER, "channel": CH}) + assert r.status_code == 200 + body = r.json() + by = {(s["type"], s.get("talker_id") or s.get("room_id")): s for s in body["sessions"]} + assert set(by) == {("friend", "t1"), ("friend", "t2"), ("group", "room1@chatroom")} + assert by[("friend", "t1")]["today_count"] == 2 + assert by[("friend", "t1")]["last_chat_time"] == _NOW + assert by[("friend", "t2")]["today_count"] == 0 + assert by[("group", "room1@chatroom")]["today_count"] == 2 + assert by[("friend", "t1")]["name"] == "Alice" + + +async def test_sessions_parity_and_legacy_guard(make_client): + """对拍:无汇总(守卫回退旧路径)响应 == 有汇总(新路径)响应,且逐字段一致。""" + legacy = await make_client(_seed) + modern = await make_client(_seed_with_stats) + params = {"owner_account_id": OWNER, "channel": CH} + r1 = await legacy.get("/api/sessions", params=params) + r2 = await modern.get("/api/sessions", params=params) + assert r1.status_code == r2.status_code == 200 + assert r1.json() == r2.json() + + +# ── staff-chat-stats ─────────────────────────────────────────────────────── # + + +async def test_staff_chat_stats_served_from_summary(make_client): + """全时段窗口(旧实现实测 6.9s 的场景):回填后删光 message 仍出全量数字。""" + client = await make_client(_stats_then_wipe) + r = await client.get( + "/api/staff-chat-stats", + params={"owner_account_id": OWNER, "start": 0, "end": _NOW + _MS_DAY}, + ) + assert r.status_code == 200 + body = r.json() + by = {c["talker_id"]: c for c in body["customers"]} + assert set(by) == {"t1", "t2"} # 私聊口径,群不计 + assert by["t1"]["send_count"] == 1 and by["t1"]["receive_count"] == 2 + assert by["t1"]["last_time"] == _NOW + assert by["t2"]["send_count"] == 1 and by["t2"]["receive_count"] == 0 + + +async def test_staff_chat_stats_parity(make_client): + """宽窗口:旧路径与汇总路径响应一致。""" + legacy = await make_client(_seed) + modern = await make_client(_seed_with_stats) + params = {"owner_account_id": OWNER, "start": 0, "end": _NOW + _MS_DAY} + r1 = await legacy.get("/api/staff-chat-stats", params=params) + r2 = await modern.get("/api/staff-chat-stats", params=params) + assert r1.status_code == r2.status_code == 200 + assert r1.json() == r2.json() diff --git a/yunke-server/tests/test_summary_daily_stats.py b/yunke-server/tests/test_summary_daily_stats.py new file mode 100644 index 0000000..07f36b3 --- /dev/null +++ b/yunke-server/tests/test_summary_daily_stats.py @@ -0,0 +1,185 @@ +"""Tests for the msg_daily_stats 汇总层(日粒度会话统计)。 + +三个被钉死的行为: + +1. **日桶口径**:day 一律 Asia/Shanghai 自然日(Python 计算,不用 SQL 日期 + 函数——旧实现 from_unixtime 跟随 MySQL 会话时区,UTC 容器下和今日数 + 的上海口径不一致,这是顺带修复的旧病)。 +2. **写路径钩子**::func:`app.storage.repos.upsert_messages` 落库后自动 + 重算本批触及的「会话×天」行——页面抓取器/CLI 手动同步全部路径共用此 + 入口,一处挂全。幂等:同一批重放不产生重复/漂移。 +3. **全量重算**::func:`app.storage.summary.recompute_all` 从 message 表 + 重建全部日行(夜间自愈/部署回填用),幂等。 + +fixtures 用内存 SQLite(测试对 message 的所有写入都走 upsert_messages, +从而同时覆盖钩子行为)。 +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from zoneinfo import ZoneInfo + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlalchemy.pool import StaticPool + +from app.db.base import Base +from app.db.models import Message, MsgDailyStats +from app.storage.repos import upsert_messages +from app.storage.summary import day_key, recompute_all + +_SH = ZoneInfo("Asia/Shanghai") + + +def _sh(y, m, d, h, mi=0, s=0) -> int: + return int( + datetime(y, m, d, h, mi, s, tzinfo=_SH).timestamp() * 1000 + ) + + +def _utc(y, m, d, h, mi=0, s=0) -> int: + return int( + datetime(y, m, d, h, mi, s, tzinfo=timezone.utc).timestamp() * 1000 + ) + + +@pytest.fixture +async def session(): + engine = create_async_engine( + "sqlite+aiosqlite:///:memory:", + poolclass=StaticPool, + connect_args={"check_same_thread": False}, + ) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + maker = async_sessionmaker(engine, expire_on_commit=False) + async with maker() as s: + yield s + await engine.dispose() + + +# ── day_key:上海自然日 ──────────────────────────────────────────────────── # + + +def test_day_key_shanghai_calendar_day(): + # 上海 2026-09-03 09:30 → 20260903 + assert day_key(_sh(2026, 9, 3, 9, 30)) == 20260903 + + +def test_day_key_uses_shanghai_not_utc(): + # UTC 2026-09-02 16:30 = 上海 2026-09-03 00:30 → 必须归入 0903(旧 + # from_unixtime 在 UTC 容器里会归入 0902——口径修复点) + assert day_key(_utc(2026, 9, 2, 16, 30)) == 20260903 + + +def test_day_key_midnight_boundary(): + # 上海零点整属于新的一天;前一天 23:59:59 属于前一天 + assert day_key(_sh(2026, 9, 3, 0, 0, 0)) == 20260903 + assert day_key(_sh(2026, 9, 2, 23, 59, 59)) == 20260902 + + +# ── 写路径钩子:upsert_messages 自动维护日行 ────────────────────────────── # + + +def _msg(msg_id, ts, *, owner="wx_a", talker="t1", room=None, is_send=True): + return dict( + msg_id=msg_id, + channel="wechat", + owner_account_id=owner, + talker_id=talker if room is None else "room_member", + room_id=room, + is_send=is_send, + timestamp=ts, + type="text", + type_raw=1, + text="x", + ) + + +async def _stats_rows(session): + rows = (await session.execute(select(MsgDailyStats))).scalars().all() + return { + (r.peer_kind, r.peer_id, r.day): (r.send_cnt, r.recv_cnt, r.last_ts, r.last_is_send) + for r in rows + } + + +async def test_upsert_messages_maintains_daily_rows(session): + await upsert_messages( + session, + [ + _msg("m1", _sh(2026, 9, 3, 10, 0), is_send=True), + _msg("m2", _sh(2026, 9, 3, 11, 0), is_send=False), + _msg("m3", _sh(2026, 9, 3, 11, 30), is_send=True), + ], + ) + await session.commit() + rows = await _stats_rows(session) + assert set(rows) == {("friend", "t1", 20260903)} + assert rows[("friend", "t1", 20260903)] == (2, 1, _sh(2026, 9, 3, 11, 30), True) + + +async def test_upsert_messages_splits_days_and_kinds(session): + await upsert_messages( + session, + [ + _msg("d1", _sh(2026, 9, 2, 23, 59), talker="t1"), + _msg("d2", _sh(2026, 9, 3, 0, 1), talker="t1"), + _msg("g1", _sh(2026, 9, 3, 9, 0), room="room1@chatroom", is_send=None), + ], + ) + await session.commit() + rows = await _stats_rows(session) + # 跨上海零点的两条消息分属两天;群消息单独一行(room 维度) + assert set(rows) == { + ("friend", "t1", 20260902), + ("friend", "t1", 20260903), + ("room", "room1@chatroom", 20260903), + } + # is_send=NULL 的消息:两个方向都不计(对齐旧聚合 case 语义),但占用行 + assert rows[("room", "room1@chatroom", 20260903)] == (0, 0, _sh(2026, 9, 3, 9, 0), None) + + +async def test_upsert_messages_idempotent(session): + batch = [ + _msg("m1", _sh(2026, 9, 3, 10, 0)), + _msg("m2", _sh(2026, 9, 3, 11, 0), is_send=False), + ] + await upsert_messages(session, batch) + await session.commit() + await upsert_messages(session, batch) # 同批重放(跨页重复是常态) + await session.commit() + rows = await _stats_rows(session) + assert len(rows) == 1 + assert rows[("friend", "t1", 20260903)] == (1, 1, _sh(2026, 9, 3, 11, 0), False) + + +async def test_upsert_messages_null_timestamp_skipped(session): + await upsert_messages(session, [_msg("n1", None)]) + await session.commit() + assert await _stats_rows(session) == {} + + +# ── 全量重算(夜间自愈/部署回填)────────────────────────────────────────── # + + +async def test_recompute_all_rebuilds_and_is_idempotent(session): + # 先写一批消息(顺带产生日行),再篡改一条日行模拟漂移,全量重算应修复 + await upsert_messages( + session, + [_msg("m1", _sh(2026, 9, 3, 10, 0)), _msg("m2", _sh(2026, 9, 3, 11, 0), is_send=False)], + ) + await session.commit() + bad = (await session.execute(select(MsgDailyStats))).scalars().one() + bad.send_cnt = 999 # 漂移 + await session.commit() + + n = await recompute_all(session) + assert n >= 1 + rows = await _stats_rows(session) + assert rows[("friend", "t1", 20260903)] == (1, 1, _sh(2026, 9, 3, 11, 0), False) + + await recompute_all(session) # 幂等(重建后是全新 ORM 对象,比字段值不比对象) + assert await _stats_rows(session) == rows