feat(management): 首响事实表跨日精确算法——跨夜串不丢/串末条归属日/同源日聚合

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-09 12:53:33 +08:00
parent 373fb266a6
commit 0b0dd0808b
2 changed files with 305 additions and 0 deletions
@@ -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))