From 58d274096d0ad3ff2caf118613c01b83fff08f3b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Wed, 9 Sep 2026 13:18:08 +0800 Subject: [PATCH] =?UTF-8?q?fix(management):=20first=5Freply=5Ffact=20?= =?UTF-8?q?=E5=90=8C=E6=AF=AB=E7=A7=92=E5=B9=B6=E5=88=97=E4=B8=BB=E9=94=AE?= =?UTF-8?q?=E5=86=B2=E7=AA=81+=E5=AE=9A=E5=BA=8F=E6=8A=96=E5=8A=A8?= =?UTF-8?q?=E2=80=94=E2=80=94reply=5Fmsg=5Fid=20=E8=BF=9B=20PK/=E6=8C=89(t?= =?UTF-8?q?s,msg=5Fid)=E6=8E=92=E5=BA=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 评审修正(代码质量评审 Critical#1 + Important#2/#3): - reply_msg_id 进复合 PK(模型+migration b7c8d9e0f1a2 原地改,本地 down/up 重建,表空未部署):同毫秒 F,T,F,T 交换的两个样本 (thread_end_ts, reply_ts) 完全并列不再撞主键(原会中断整个 recompute_all 事务);兼作下钻证据定位回复消息。 - load_conversation 增查 msg_id 并按 (ts, msg_id) 排序, compute_conversation 协议改 3 元组——SQL 等键顺序未定义,原实现 同毫秒 [F,T] 归属随查询抖动(0/1 answered 摇摆)。 - recompute_pairs 每 500 会话显式 flush+进度日志(内存上界对齐 summary.recompute_all 惯例,不再依赖隐式 autoflush)。 - 测试:旧 9 个更新协议;新增同毫秒 F,T,F,T 回归/并列确定性/ 空会话幂等清空,fr_sum 总量断言并入双串测试; test_management_models 构造补新主键列。 Co-Authored-By: Claude Code --- yunke-server/app/db/models.py | 3 + yunke-server/app/storage/first_reply_facts.py | 59 +++++++++--- .../b7c8d9e0f1a2_management_phase1_tables.py | 3 + yunke-server/tests/test_first_reply_facts.py | 89 ++++++++++++++++--- yunke-server/tests/test_management_models.py | 5 +- 5 files changed, 129 insertions(+), 30 deletions(-) diff --git a/yunke-server/app/db/models.py b/yunke-server/app/db/models.py index d9cbe28..7849632 100644 --- a/yunke-server/app/db/models.py +++ b/yunke-server/app/db/models.py @@ -1021,6 +1021,9 @@ class FirstReplyFact(Base): peer_id: Mapped[str] = mapped_column(String(128), primary_key=True) thread_end_ts: Mapped[int] = mapped_column(BigInteger, primary_key=True) reply_ts: Mapped[int] = mapped_column(BigInteger, primary_key=True) + # 回复消息 msg_id:同毫秒 F,T,F,T 交换的 (thread_end_ts, reply_ts) 并列 + # 去重(一条回复恰好闭合一个串);兼作下钻证据定位回复消息 + reply_msg_id: Mapped[str] = mapped_column(String(64), primary_key=True) # 串末条所在上海自然日 yyyymmdd(冗余物化,窗口/趋势索引) day: Mapped[int] = mapped_column(Integer, nullable=False) fr_ms: Mapped[int] = mapped_column(BigInteger, nullable=False) diff --git a/yunke-server/app/storage/first_reply_facts.py b/yunke-server/app/storage/first_reply_facts.py index df7fa48..6005107 100644 --- a/yunke-server/app/storage/first_reply_facts.py +++ b/yunke-server/app/storage/first_reply_facts.py @@ -8,11 +8,17 @@ compute_conversation 同时输出每日串聚合(threads/answered/fr_sum_ms, 按串末条归属日)——employee_metric_daily 的串类列与 fact 行**同源** (回复率分子分母一致,v2 §1.1 #5)。 + +时序确定性:同毫秒并列(ts 相同的 F/T 消息)按 msg_id 定序(wechat +msgSvrId 服务端单调,近似真实到达序)——SQL 等键返回顺序未定义,不做 +定序会导致串归属随查询抖动;fact PK 含 reply_msg_id,同毫秒 F,T,F,T +交换的两个样本不撞主键。 """ from __future__ import annotations import logging from collections import namedtuple +from collections.abc import Iterable import sqlalchemy as sa from sqlalchemy.ext.asyncio import AsyncSession @@ -22,8 +28,9 @@ from app.storage.summary import day_key logger = logging.getLogger(__name__) -# 一个被回串的首响样本;day = 串末条客户消息所在上海自然日 -FactSample = namedtuple("FactSample", "thread_end_ts reply_ts fr_ms day") +# 一个被回串的首响样本;day = 串末条客户消息所在上海自然日, +# reply_msg_id = 闭合该串的员工回复消息 msg_id(PK 并列去重 + 下钻证据) +FactSample = namedtuple("FactSample", "thread_end_ts reply_ts fr_ms day reply_msg_id") # 某会话某日的串聚合(分子分母同源) DayAgg = namedtuple("DayAgg", "threads answered fr_sum_ms") @@ -38,19 +45,23 @@ class _DayAcc: def compute_conversation(rows: list) -> tuple[list[FactSample], dict[int, DayAgg]]: - """跨日精确串算法。rows = [(ts, is_send)] 该会话全历史,**须按 ts 升序**。 + """跨日精确串算法。rows = [(ts, is_send, msg_id)] 该会话全历史, + **须按 (ts, msg_id) 升序**(load_conversation 负责排序;同毫秒并列以 + msg_id 定序,见模块 docstring)。 - 串 = 客户连续消息(is_send=False)不被员工消息(is_send=True)打断的 段;is_send=None 跳过(不打断也不起串)——与 _thread_stats 判定一致。 - 每个被回串一个样本:fr_ms = 员工首条回复 ts − 串**末条**客户消息 ts, - 归属 day = 串末条所在上海自然日(跨夜串不再丢失)。 + 归属 day = 串末条所在上海自然日(跨夜串不再丢失);reply_msg_id = + 该回复消息 msg_id——一条回复恰好闭合一个串,同毫秒 F,T,F,T 交换的 + (thread_end_ts, reply_ts) 并列靠它去重,兼作下钻证据。 - 会话末尾仍未闭合的串:计入当日 threads(回复率分母),不产生样本。 """ facts: list[FactSample] = [] acc: dict[int, _DayAcc] = {} in_thread = False end_ts: int | None = None - for ts, is_send in rows: + for ts, is_send, msg_id in rows: if is_send is False: in_thread = True end_ts = int(ts) @@ -58,7 +69,7 @@ def compute_conversation(rows: list) -> tuple[list[FactSample], dict[int, DayAgg if in_thread and end_ts is not None: fr = int(ts) - end_ts d = day_key(end_ts) - facts.append(FactSample(end_ts, int(ts), fr, d)) + facts.append(FactSample(end_ts, int(ts), fr, d, msg_id)) a = acc.setdefault(d, _DayAcc()) a.threads += 1 a.answered += 1 @@ -76,13 +87,19 @@ def compute_conversation(rows: list) -> tuple[list[FactSample], dict[int, DayAgg async def load_conversation( session: AsyncSession, channel: str, owner: str, peer: str ) -> list: - """拉取该 1:1 会话全历史 (timestamp, is_send),Python 侧排序(串算法 - 需严格时序——与 summary.recompute_days 同惯例,避免依赖 SQL 排序)。 + """拉取该 1:1 会话全历史 (timestamp, is_send, msg_id),Python 侧按 + (ts, msg_id) 排序——串算法需严格时序且同毫秒并列须定序(SQL 等键顺序 + 未定义,不排序会随查询抖动;与 summary.recompute_days 同惯例,避免 + 依赖 SQL 排序)。friend 过滤口径与 summary._peer_filter 的 friend 分支 + 同源:talker_id 原值精确匹配 + room_id 为空(peer_id 与 talker_id 同源 + 同形态,不做截断/归一化)。 公有命名:employee_metrics 宽表生成复用同一加载口径(同源原则)。""" rows = ( await session.execute( - sa.select(Message.timestamp, Message.is_send).where( + sa.select( + Message.timestamp, Message.is_send, Message.msg_id + ).where( Message.channel == channel, Message.owner_account_id == owner, Message.talker_id == peer, @@ -91,7 +108,7 @@ async def load_conversation( ) ) ).all() - return sorted(rows, key=lambda r: r[0]) + return sorted(rows, key=lambda r: (r[0], r[2])) async def recompute_pairs( @@ -99,8 +116,12 @@ async def recompute_pairs( ) -> int: """重算给定 friend 会话的 fact 行:按会话 delete+insert,幂等。 - pairs = [(channel, owner_account_id, peer_id)]。返回会话数。 + pairs = [(channel, owner_account_id, peer_id)]。返回会话数。每 500 个 + 会话 flush 一次约束 session 内存(与 summary.recompute_all 同惯例—— + 全量回填百万级 fact 行不能攒到最后),同时打进度日志(回填分钟级, + 可观测)。commit 由调用方负责。 """ + n = 0 for channel, owner, peer in sorted(pairs): rows = await load_conversation(session, channel, owner, peer) facts, _agg = compute_conversation(rows) @@ -119,15 +140,25 @@ async def recompute_pairs( peer_id=peer, thread_end_ts=f.thread_end_ts, reply_ts=f.reply_ts, + reply_msg_id=f.reply_msg_id, day=f.day, fr_ms=f.fr_ms, ) ) - return len(pairs) + n += 1 + if n % 500 == 0: + await session.flush() + logger.info("first_reply_facts: 已重算 %d 个会话", n) + return n async def all_friend_pairs(session: AsyncSession) -> list[tuple[str, str, str]]: - """全部 1:1 会话(friend 行 distinct)——全量回填用。""" + """全部 1:1 会话(friend 行 distinct)——全量回填用。 + + 依赖 msg_daily_stats 已回填(先 stats 后 facts):friend 会话清单取自 + stats 表而非 message 全表 distinct——写路径钩子保证 stats 常新,代价 + 远低于扫 message。 + """ rows = ( await session.execute( sa.select( @@ -141,7 +172,7 @@ async def all_friend_pairs(session: AsyncSession) -> list[tuple[str, str, str]]: async def pairs_active_in_days( - session: AsyncSession, days + session: AsyncSession, days: Iterable[int] ) -> list[tuple[str, str, str]]: """窗口内有消息的 friend 会话(4h 滚动重算用;批次F 接调度)。""" rows = ( diff --git a/yunke-server/migrations/versions/b7c8d9e0f1a2_management_phase1_tables.py b/yunke-server/migrations/versions/b7c8d9e0f1a2_management_phase1_tables.py index 5326c73..2070c31 100644 --- a/yunke-server/migrations/versions/b7c8d9e0f1a2_management_phase1_tables.py +++ b/yunke-server/migrations/versions/b7c8d9e0f1a2_management_phase1_tables.py @@ -26,6 +26,9 @@ _TABLES = [ sa.Column("peer_id", sa.String(length=128), primary_key=True), sa.Column("thread_end_ts", sa.BigInteger(), primary_key=True), sa.Column("reply_ts", sa.BigInteger(), primary_key=True), + # 同毫秒 F,T,F,T 交换的 (thread_end_ts, reply_ts) 并列去重(一条回复 + # 恰好闭合一个串,reply_msg_id 唯一定位);兼作下钻证据定位回复消息 + sa.Column("reply_msg_id", sa.String(length=64), primary_key=True), sa.Column("day", sa.Integer(), nullable=False), sa.Column("fr_ms", sa.BigInteger(), nullable=False), )), diff --git a/yunke-server/tests/test_first_reply_facts.py b/yunke-server/tests/test_first_reply_facts.py index fb75504..af23726 100644 --- a/yunke-server/tests/test_first_reply_facts.py +++ b/yunke-server/tests/test_first_reply_facts.py @@ -5,6 +5,10 @@ 2. 多日未回串只计一次、归属末条日(日切会按天重复计——分母膨胀); 3. 同源:threads_answered ≡ fact 行数(按归属日); 4. is_send=None 不参与串判定(与 _thread_stats 一致)。 + +评审修正补判据: +5. 同毫秒 F,T,F,T 交换 → reply_msg_id 进 PK 去重,不撞主键(原崩溃回归); +6. 同毫秒并列按 (ts, msg_id) 定序,结果确定不随 SQL 等键序抖动。 """ import datetime as dt from zoneinfo import ZoneInfo @@ -30,7 +34,8 @@ def _ts(*args) -> int: def test_overnight_thread_recovered(): """客户 23:00 发→员工次日 9:00 回:fr_ms≈10h,归属 D1(v2 §1.1 #1)。""" facts, agg = compute_conversation( - [(_ts(2026, 9, 1, 23, 0), False), (_ts(2026, 9, 2, 9, 0), True)] + [(_ts(2026, 9, 1, 23, 0), False, "f1"), + (_ts(2026, 9, 2, 9, 0), True, "t1")] ) assert len(facts) == 1 assert facts[0].fr_ms == 10 * 3600 * 1000 @@ -41,8 +46,8 @@ def test_overnight_thread_recovered(): def test_multiday_open_thread_counts_once_at_end_day(): """客户消息跨 D1/D2 无回复:1 个串归属 D2;日切口径会 D1、D2 各计 1。""" facts, agg = compute_conversation( - [(_ts(2026, 9, 1, 10, 0), False), (_ts(2026, 9, 1, 20, 0), False), - (_ts(2026, 9, 2, 9, 0), False)] + [(_ts(2026, 9, 1, 10, 0), False, "f1"), (_ts(2026, 9, 1, 20, 0), False, "f2"), + (_ts(2026, 9, 2, 9, 0), False, "f3")] ) assert facts == [] assert 20260901 not in agg @@ -52,26 +57,27 @@ def test_multiday_open_thread_counts_once_at_end_day(): def test_reply_closes_thread_and_opens_new(): """员工回复后客户再发=新串;串内多条客户消息取末条为 end。""" facts, agg = compute_conversation( - [(_ts(2026, 9, 1, 10, 0), False), (_ts(2026, 9, 1, 10, 5), False), - (_ts(2026, 9, 1, 10, 30), True), # 回复1:fr=30-10:05=25min - (_ts(2026, 9, 1, 11, 0), False), (_ts(2026, 9, 1, 12, 0), True)] - ) + [(_ts(2026, 9, 1, 10, 0), False, "f1"), (_ts(2026, 9, 1, 10, 5), False, "f2"), + (_ts(2026, 9, 1, 10, 30), True, "t1"), # 回复1:fr=30-10:05=25min + (_ts(2026, 9, 1, 11, 0), False, "f3"), + (_ts(2026, 9, 1, 12, 0), True, "t2")]) # 回复2:fr=12:00-11:00=60min assert len(facts) == 2 assert facts[0].fr_ms == 25 * 60 * 1000 assert agg[20260901].threads == 2 and agg[20260901].answered == 2 + assert agg[20260901].fr_sum_ms == 85 * 60 * 1000 # 25min + 60min def test_unknown_direction_ignored(): facts, _agg = compute_conversation( - [(_ts(2026, 9, 1, 10, 0), False), (_ts(2026, 9, 1, 10, 10), None), - (_ts(2026, 9, 1, 10, 20), True)] + [(_ts(2026, 9, 1, 10, 0), False, "f1"), (_ts(2026, 9, 1, 10, 10), None, "u1"), + (_ts(2026, 9, 1, 10, 20), True, "t1")] ) assert len(facts) == 1 # None 不打断串:仍视为对 10:00 串的回复 def test_staff_message_outside_thread_is_noop(): facts, agg = compute_conversation( - [(_ts(2026, 9, 1, 9, 0), True), (_ts(2026, 9, 1, 10, 0), True)] + [(_ts(2026, 9, 1, 9, 0), True, "t1"), (_ts(2026, 9, 1, 10, 0), True, "t2")] ) assert facts == [] and agg == {} @@ -84,12 +90,29 @@ def test_unsorted_rows_raise_or_documented(): assert "升序" in m.compute_conversation.__doc__ +def test_tie_break_by_msg_id_deterministic(): + """同毫秒 [F(m1), T(m2)]:按 (ts, msg_id) 定序 F 在前 → 1 个被回串; + 同一 multiset 无论入参顺序,排序后结果一致(SQL 等键序未定义,不能 + 随查询抖动)。""" + same = _ts(2026, 9, 1, 10, 0) + rows = [(same, False, "m1"), (same, True, "m2")] + ordered = sorted(rows, key=lambda r: (r[0], r[2])) + assert ordered[0][2] == "m1" # F(m1) 先于 T(m2) + facts, agg = compute_conversation(ordered) + assert len(facts) == 1 and facts[0].reply_msg_id == "m2" + assert agg[20260901].answered == 1 + shuffled = [rows[1], rows[0]] # 同一 multiset,入参顺序相反 + facts2, agg2 = compute_conversation( + sorted(shuffled, key=lambda r: (r[0], r[2]))) + assert facts2 == facts and agg2 == agg + + # ---------------------------- 落库 ---------------------------- async def _seed_conversation(session, *, channel="wechat", owner="wx1", peer="p1", rows): - for i, (ts, is_send) in enumerate(rows): - session.add(Message(msg_id=f"{peer}-{i}", channel=channel, + for ts, is_send, msg_id in rows: + session.add(Message(msg_id=msg_id, channel=channel, owner_account_id=owner, talker_id=peer, timestamp=ts, is_send=is_send, type="text", text="x")) @@ -102,23 +125,43 @@ async def _seed_conversation(session, *, channel="wechat", owner="wx1", async def test_recompute_pairs_persists_and_idempotent(make_session): s = make_session await _seed_conversation( - s, rows=[(_ts(2026, 9, 1, 23, 0), False), (_ts(2026, 9, 2, 9, 0), True)]) + s, rows=[(_ts(2026, 9, 1, 23, 0), False, "p1-f0"), + (_ts(2026, 9, 2, 9, 0), True, "p1-t1")]) n = await recompute_pairs(s, [("wechat", "wx1", "p1")]) await s.commit() assert n == 1 rows = (await s.execute(sa.select(FirstReplyFact))).scalars().all() assert len(rows) == 1 and rows[0].fr_ms == 10 * 3600 * 1000 + assert rows[0].reply_msg_id == "p1-t1" # 幂等:重跑不重复 await recompute_pairs(s, [("wechat", "wx1", "p1")]) await s.commit() assert len((await s.execute(sa.select(FirstReplyFact))).scalars().all()) == 1 +async def test_same_ms_ftft_two_distinct_facts(make_session): + """同毫秒 F,T,F,T 交换:两个样本 (thread_end_ts, reply_ts) 完全并列 + ——reply_msg_id 进 PK 去重后恰好 2 行(评审复现的主键冲突回归)。""" + s = make_session + same = _ts(2026, 9, 1, 10, 0) + await _seed_conversation(s, rows=[ + (same, False, "m1"), (same, True, "m2"), + (same, False, "m3"), (same, True, "m4"), + ]) + n = await recompute_pairs(s, [("wechat", "wx1", "p1")]) + await s.commit() + assert n == 1 + rows = (await s.execute(sa.select(FirstReplyFact))).scalars().all() + assert len(rows) == 2 + assert {r.reply_msg_id for r in rows} == {"m2", "m4"} + + async def test_recompute_pairs_replaces_stale(make_session): """会话追加消息后重算,旧样本被替换(delete+insert 语义)。""" s = make_session await _seed_conversation( - s, rows=[(_ts(2026, 9, 1, 10, 0), False), (_ts(2026, 9, 1, 11, 0), True)]) + s, rows=[(_ts(2026, 9, 1, 10, 0), False, "p1-f0"), + (_ts(2026, 9, 1, 11, 0), True, "p1-t1")]) await recompute_pairs(s, [("wechat", "wx1", "p1")]) await s.commit() s.add(Message(msg_id="p1-new", channel="wechat", owner_account_id="wx1", @@ -131,6 +174,24 @@ async def test_recompute_pairs_replaces_stale(make_session): assert len(rows) == 1 # 新串未回,只剩旧样本 +async def test_recompute_pairs_empty_conversation(make_session): + """stats 行在但 message 无合格行(空会话/全被过滤):删旧行不插新—— + delete+insert 语义对空会话同样成立(幂等清空)。""" + s = make_session + s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="p1", day=20260901)) + # 预置历史残留 fact,重算后应被清掉 + s.add(FirstReplyFact(channel="wechat", owner_account_id="wx1", peer_id="p1", + thread_end_ts=_ts(2026, 9, 1, 10, 0), + reply_ts=_ts(2026, 9, 1, 11, 0), reply_msg_id="stale", + day=20260901, fr_ms=3600 * 1000)) + await s.commit() + n = await recompute_pairs(s, [("wechat", "wx1", "p1")]) + await s.commit() + assert n == 1 + assert (await s.execute(sa.select(FirstReplyFact))).scalars().all() == [] + + async def test_all_friend_pairs_only_friend(make_session): s = make_session s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", diff --git a/yunke-server/tests/test_management_models.py b/yunke-server/tests/test_management_models.py index 03c51b4..ee2df83 100644 --- a/yunke-server/tests/test_management_models.py +++ b/yunke-server/tests/test_management_models.py @@ -21,7 +21,7 @@ async def test_first_reply_fact_roundtrip(make_session): s = make_session s.add(FirstReplyFact(channel="wechat", owner_account_id="wx1", peer_id="p1", thread_end_ts=1_700_000_000_000, - reply_ts=1_700_000_600_000, + reply_ts=1_700_000_600_000, reply_msg_id="m1", day=20260901, fr_ms=600_000)) await s.commit() val = (await s.execute(sa.select(FirstReplyFact.fr_ms))).scalar_one() @@ -31,7 +31,8 @@ async def test_first_reply_fact_roundtrip(make_session): async def test_first_reply_fact_unique_sample(make_session): s = make_session kwargs = {"channel": "wechat", "owner_account_id": "wx1", "peer_id": "p1", - "thread_end_ts": 1, "reply_ts": 2, "day": 20260901, "fr_ms": 1} + "thread_end_ts": 1, "reply_ts": 2, "reply_msg_id": "m1", + "day": 20260901, "fr_ms": 1} s.add(FirstReplyFact(**kwargs)) await s.commit() s.add(FirstReplyFact(**kwargs))