diff --git a/yunke-server/app/storage/first_reply_facts.py b/yunke-server/app/storage/first_reply_facts.py new file mode 100644 index 0000000..df7fa48 --- /dev/null +++ b/yunke-server/app/storage/first_reply_facts.py @@ -0,0 +1,164 @@ +"""first_reply_fact 维护层(管理域,批次A)——跨日精确串算法。 + +区别于 app.storage.summary._thread_stats 的日切近似(v2 §1.1 #1): +按会话**全历史**计算,跨夜串(客户 23:00 → 员工次日 9:00 回)不再丢失。 +**不挂写钩子**(v2 §3.1):由回填 CLI / 4h 滚动 / 夜间全量重算,写钩子 +零改动。只处理 1:1(friend)会话——群串语义不成立。 + +compute_conversation 同时输出每日串聚合(threads/answered/fr_sum_ms, +按串末条归属日)——employee_metric_daily 的串类列与 fact 行**同源** +(回复率分子分母一致,v2 §1.1 #5)。 +""" +from __future__ import annotations + +import logging +from collections import namedtuple + +import sqlalchemy as sa +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import FirstReplyFact, Message, MsgDailyStats +from app.storage.summary import day_key + +logger = logging.getLogger(__name__) + +# 一个被回串的首响样本;day = 串末条客户消息所在上海自然日 +FactSample = namedtuple("FactSample", "thread_end_ts reply_ts fr_ms day") +# 某会话某日的串聚合(分子分母同源) +DayAgg = namedtuple("DayAgg", "threads answered fr_sum_ms") + + +class _DayAcc: + __slots__ = ("answered", "fr_sum_ms", "threads") + + def __init__(self) -> None: + self.threads = 0 + self.answered = 0 + self.fr_sum_ms = 0 + + +def compute_conversation(rows: list) -> tuple[list[FactSample], dict[int, DayAgg]]: + """跨日精确串算法。rows = [(ts, is_send)] 该会话全历史,**须按 ts 升序**。 + + - 串 = 客户连续消息(is_send=False)不被员工消息(is_send=True)打断的 + 段;is_send=None 跳过(不打断也不起串)——与 _thread_stats 判定一致。 + - 每个被回串一个样本:fr_ms = 员工首条回复 ts − 串**末条**客户消息 ts, + 归属 day = 串末条所在上海自然日(跨夜串不再丢失)。 + - 会话末尾仍未闭合的串:计入当日 threads(回复率分母),不产生样本。 + """ + facts: list[FactSample] = [] + acc: dict[int, _DayAcc] = {} + in_thread = False + end_ts: int | None = None + for ts, is_send in rows: + if is_send is False: + in_thread = True + end_ts = int(ts) + elif is_send is True: + 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)) + a = acc.setdefault(d, _DayAcc()) + a.threads += 1 + a.answered += 1 + a.fr_sum_ms += fr + in_thread = False + end_ts = None + # is_send=None:跳过 + if in_thread and end_ts is not None: # 会话末尾未闭合串 + a = acc.setdefault(day_key(end_ts), _DayAcc()) + a.threads += 1 + return facts, {d: DayAgg(a.threads, a.answered, a.fr_sum_ms) + for d, a in acc.items()} + + +async def load_conversation( + session: AsyncSession, channel: str, owner: str, peer: str +) -> list: + """拉取该 1:1 会话全历史 (timestamp, is_send),Python 侧排序(串算法 + 需严格时序——与 summary.recompute_days 同惯例,避免依赖 SQL 排序)。 + + 公有命名:employee_metrics 宽表生成复用同一加载口径(同源原则)。""" + rows = ( + await session.execute( + sa.select(Message.timestamp, Message.is_send).where( + Message.channel == channel, + Message.owner_account_id == owner, + Message.talker_id == peer, + sa.or_(Message.room_id.is_(None), Message.room_id == ""), + Message.timestamp.is_not(None), + ) + ) + ).all() + return sorted(rows, key=lambda r: r[0]) + + +async def recompute_pairs( + session: AsyncSession, pairs: list[tuple[str, str, str]] +) -> int: + """重算给定 friend 会话的 fact 行:按会话 delete+insert,幂等。 + + pairs = [(channel, owner_account_id, peer_id)]。返回会话数。 + """ + for channel, owner, peer in sorted(pairs): + rows = await load_conversation(session, channel, owner, peer) + facts, _agg = compute_conversation(rows) + await session.execute( + sa.delete(FirstReplyFact).where( + FirstReplyFact.channel == channel, + FirstReplyFact.owner_account_id == owner, + FirstReplyFact.peer_id == peer, + ) + ) + for f in facts: + session.add( + FirstReplyFact( + channel=channel, + owner_account_id=owner, + peer_id=peer, + thread_end_ts=f.thread_end_ts, + reply_ts=f.reply_ts, + day=f.day, + fr_ms=f.fr_ms, + ) + ) + return len(pairs) + + +async def all_friend_pairs(session: AsyncSession) -> list[tuple[str, str, str]]: + """全部 1:1 会话(friend 行 distinct)——全量回填用。""" + rows = ( + await session.execute( + sa.select( + MsgDailyStats.channel, + MsgDailyStats.owner_account_id, + MsgDailyStats.peer_id, + ).where(MsgDailyStats.peer_kind == "friend") + ) + ).all() + return sorted({(r[0], r[1], r[2]) for r in rows}) + + +async def pairs_active_in_days( + session: AsyncSession, days +) -> list[tuple[str, str, str]]: + """窗口内有消息的 friend 会话(4h 滚动重算用;批次F 接调度)。""" + rows = ( + await session.execute( + sa.select( + MsgDailyStats.channel, + MsgDailyStats.owner_account_id, + MsgDailyStats.peer_id, + ).where( + MsgDailyStats.peer_kind == "friend", + MsgDailyStats.day.in_(list(days)), + ) + ) + ).all() + return sorted({(r[0], r[1], r[2]) for r in rows}) + + +async def recompute_all(session: AsyncSession) -> int: + """全量重建(回填/夜间自愈,幂等)。""" + return await recompute_pairs(session, await all_friend_pairs(session)) diff --git a/yunke-server/tests/test_first_reply_facts.py b/yunke-server/tests/test_first_reply_facts.py new file mode 100644 index 0000000..fb75504 --- /dev/null +++ b/yunke-server/tests/test_first_reply_facts.py @@ -0,0 +1,141 @@ +"""first_reply_fact 跨日精确串算法判别测试(批次A Task3)。 + +四个核心判据(v2 §1.1 #1/#5、§9): +1. 跨夜串必须产生样本(日切口径丢失的场景); +2. 多日未回串只计一次、归属末条日(日切会按天重复计——分母膨胀); +3. 同源:threads_answered ≡ fact 行数(按归属日); +4. is_send=None 不参与串判定(与 _thread_stats 一致)。 +""" +import datetime as dt +from zoneinfo import ZoneInfo + +import sqlalchemy as sa + +from app.db.models import FirstReplyFact, Message, MsgDailyStats +from app.storage.first_reply_facts import ( + all_friend_pairs, + compute_conversation, + recompute_pairs, +) + +_SH = ZoneInfo("Asia/Shanghai") + + +def _ts(*args) -> int: + return int(dt.datetime(*args, tzinfo=_SH).timestamp() * 1000) + + +# ---------------------------- 纯函数 ---------------------------- + +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)] + ) + assert len(facts) == 1 + assert facts[0].fr_ms == 10 * 3600 * 1000 + assert facts[0].day == 20260901 + assert agg[20260901].answered == 1 + + +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)] + ) + assert facts == [] + assert 20260901 not in agg + assert agg[20260902].threads == 1 and agg[20260902].answered == 0 + + +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)] + ) + assert len(facts) == 2 + assert facts[0].fr_ms == 25 * 60 * 1000 + assert agg[20260901].threads == 2 and agg[20260901].answered == 2 + + +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)] + ) + 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)] + ) + assert facts == [] and agg == {} + + +def test_unsorted_rows_raise_or_documented(): + """调用方契约:rows 须升序。乱序输入行为未定义——文档断言 + 防御排序 + 在 load_conversation 里做(与 summary.py 惯例一致)。""" + # 纯函数不做防御(性能);落库路径负责排序。此处仅锁定契约注释。 + import app.storage.first_reply_facts as m + assert "升序" in m.compute_conversation.__doc__ + + +# ---------------------------- 落库 ---------------------------- + +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, + owner_account_id=owner, talker_id=peer, + timestamp=ts, is_send=is_send, type="text", + text="x")) + session.add(MsgDailyStats(channel=channel, owner_account_id=owner, + peer_kind="friend", peer_id=peer, + day=20260901, send_cnt=1, recv_cnt=1)) + await session.commit() + + +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)]) + 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 + # 幂等:重跑不重复 + await recompute_pairs(s, [("wechat", "wx1", "p1")]) + await s.commit() + assert len((await s.execute(sa.select(FirstReplyFact))).scalars().all()) == 1 + + +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)]) + await recompute_pairs(s, [("wechat", "wx1", "p1")]) + await s.commit() + s.add(Message(msg_id="p1-new", channel="wechat", owner_account_id="wx1", + talker_id="p1", timestamp=_ts(2026, 9, 1, 12, 0), + is_send=False, type="text", text="y")) + await s.commit() + await recompute_pairs(s, [("wechat", "wx1", "p1")]) + await s.commit() + rows = (await s.execute(sa.select(FirstReplyFact))).scalars().all() + assert len(rows) == 1 # 新串未回,只剩旧样本 + + +async def test_all_friend_pairs_only_friend(make_session): + s = make_session + s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="room", peer_id="r1", day=20260901)) + s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", + peer_kind="friend", peer_id="p1", day=20260901)) + await s.commit() + assert await all_friend_pairs(s) == [("wechat", "wx1", "p1")]