feat(analytics): 首响时长+回复率数据管道——消息串算法/日汇总扩展列/员工聚合

批1(A1+A2):管理层要看「每个人的响应质量」。 - 迁移 d9e0f1a2b3c4:msg_daily_stats/stat_daily_staff 各加 cust_threads/threads_answered/fr_sum_ms(NOT NULL server_default 0, 幂等守卫照 9b0d0c133035 惯例;models.py 双声明同步) - summary.py _thread_stats:串=客户连续消息不被员工消息打断的段, is_send=None 不参与串判定;fr=员工首条回复−串末条客户消息; 日切近似 v1(日末未闭合串计 threads 不计 answered,口径写进 docstring) - recompute_days/recompute_all 都先按 ts 排序再算串(乱序防护) - staff_summary 聚合三列按账号→员工累加(fr_cnt≡threads_answered 不单列) - 测试 +10:串算法矩阵 9 例+staff 聚合传递;全量 589 绿 - 本地 86 万行回填验证:daily-stats 60s/staff 10s,rollup 总和与明细完全一致,随机 5 行 SQL 独立复算全对 Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-07 14:13:09 +08:00
parent 39791b2a49
commit feb3b98665
6 changed files with 359 additions and 5 deletions
@@ -776,6 +776,10 @@ class MsgDailyStats(Base):
    - ``send_cnt``/``recv_cnt`` 仅计 is_send 真/假;NULL 两边都不计(对齐
      旧聚合的 case 语义)。
    - ``last_ts``/``last_is_send`` 为该日最后一条消息的时间与方向。
+    - 首响/回复率三列(2026-09-07 批1)按「消息串」口径,见
+      :func:`app.storage.summary._thread_stats`:串=客户连续消息不被员工
+      消息打断的段(is_send=NULL 不参与串判定);日切近似——日末未闭合
+      串计 threads 不计 answered,跨夜的串在次日段重新起算。
    """

    __tablename__ = "msg_daily_stats"
@@ -796,6 +800,18 @@ class MsgDailyStats(Base):
    # 当日最后一条消息(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)
+    # 当日客户串数(串=客户连续消息不被员工消息打断的段;NULL 方向不参与)
+    cust_threads: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    # 被员工回复的串数 ≡ 首响样本数 fr_cnt(fr_cnt=threads_answered 不单列)
+    threads_answered: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    # 首响样本时长和(ms):展示用首响时长 = fr_sum_ms / threads_answered
+    fr_sum_ms: Mapped[int] = mapped_column(
+        sa.BigInteger, nullable=False, default=0, server_default=sa.text("0")
+    )

    updated_at: Mapped[dt.datetime] = _updated_at()

@@ -829,6 +845,17 @@ class StatDailyStaff(Base):
    active_rooms: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=sa.text("0")
    )
+    # 首响/回复率三列(2026-09-07 批1):msg_daily_stats 同义列按账号→员工
+    # 累加。回复率=Σanswered/Σthreads;首响时长=Σfr_sum_ms/Σanswered。
+    cust_threads: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    threads_answered: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default=sa.text("0")
+    )
+    fr_sum_ms: Mapped[int] = mapped_column(
+        sa.BigInteger, nullable=False, default=0, server_default=sa.text("0")
+    )
    updated_at: Mapped[dt.datetime] = _updated_at()


@@ -89,6 +89,10 @@ async def recompute_days(
                        MsgDailyStats.peer_id,
                        MsgDailyStats.send_cnt,
                        MsgDailyStats.recv_cnt,
+                        # 首响/回复率三列(2026-09-07 批1):同义聚合累加
+                        MsgDailyStats.cust_threads,
+                        MsgDailyStats.threads_answered,
+                        MsgDailyStats.fr_sum_ms,
                    ).where(
                        MsgDailyStats.day == day, MsgDailyStats.channel == channel
                    )
@@ -102,13 +106,25 @@ async def recompute_days(
            if not rows:
                continue
            agg: dict[str, dict] = {}
-            for owner, kind, peer, s, r in rows:
+            for owner, kind, peer, s, r, t, a_ns, fr in rows:
                uid = amap.get(owner, owner)  # 映射不上按 owner 原值(读侧标注)
                a = agg.setdefault(
-                    uid, {"sent": 0, "recv": 0, "friends": set(), "rooms": set()}
+                    uid,
+                    {
+                        "sent": 0,
+                        "recv": 0,
+                        "friends": set(),
+                        "rooms": set(),
+                        "threads": 0,
+                        "answered": 0,
+                        "fr": 0,
+                    },
                )
                a["sent"] += int(s or 0)
                a["recv"] += int(r or 0)
+                a["threads"] += int(t or 0)
+                a["answered"] += int(a_ns or 0)
+                a["fr"] += int(fr or 0)
                if kind == "friend":  # 显式分支:未来新 kind 跳过,不误入 rooms
                    a["friends"].add(peer)
                elif kind == "room":
@@ -123,6 +139,9 @@ async def recompute_days(
                        msg_recv=a["recv"],
                        active_friends=len(a["friends"]),
                        active_rooms=len(a["rooms"]),
+                        cust_threads=a["threads"],
+                        threads_answered=a["answered"],
+                        fr_sum_ms=a["fr"],
                    )
                )
            n += 1
@@ -14,6 +14,17 @@

 计数语义与旧实时聚合完全一致:is_send=True 计 send、False 计 recv、
 NULL 两边不计;timestamp 为 NULL 的消息不参与统计。
+
+首响/回复率三列(2026-09-07 批1,A1+A2)——「消息串」口径(与 8-24
+spec 对齐,定案):
+
+- **串**:同会话内客户连续消息(is_send=False)不被员工消息(is_send=True)
+  打断的段;is_send=None(unknown)不参与串判定(跳过——不打断也不起串)。
+- **首响样本**:每个被回串一个样本 fr_ms = 员工首条回复 ts − 串末条客户
+  消息 ts。fr_cnt ≡ threads_answered(被回串数=样本数),不单列。
+- **日切近似**(v1):单日内计算,日末未闭合的串计 threads 不计 answered,
+  跨夜的串在次日段重新起算——8-24 spec 的 24h 窗口径的简化,读端拼均值
+  (fr_sum_ms / threads_answered)不受影响,仅跨夜串的归属日有偏差。
 """

 from __future__ import annotations
@@ -113,6 +124,36 @@ def _peer_filter(kind: str, peer: str) -> list:
    ]


+def _thread_stats(rows: list) -> tuple[int, int, int]:
+    """消息串口径三值:(cust_threads, threads_answered, fr_sum_ms)。
+
+    - 串 = 客户连续消息(is_send=False)不被员工消息(is_send=True)打断
+      的段;is_send=None(unknown)不参与串判定——跳过,不打断也不起串。
+    - 每个被回串一个首响样本 fr_ms = 员工首条回复 ts − 串**末条**客户消息
+      ts;fr_sum_ms 为样本时长和(展示用首响时长 = fr_sum_ms/answered)。
+    - 日切近似 v1:日末未闭合的串计 threads 不计 answered(跨夜的串在次日
+      段重新起算——8-24 spec 的 24h 窗口径的简化)。
+    - rows 须先按 ts 升序(调用方 sorted;SQL 无 ORDER BY,乱序会错算串)。
+    """
+    threads = answered = 0
+    fr_sum = 0
+    in_thread = False
+    thread_last_ts: int | None = None
+    for ts, is_send in rows:
+        if is_send is False:
+            if not in_thread:
+                in_thread = True
+                threads += 1
+            thread_last_ts = int(ts)
+        elif is_send is True:
+            if in_thread and thread_last_ts is not None:
+                answered += 1
+                fr_sum += int(ts) - thread_last_ts
+                in_thread = False
+        # is_send=None:跳过(不参与串判定)
+    return threads, answered, fr_sum
+
+
 async def recompute_days(
    session: AsyncSession,
    keys: set[tuple[str, str, str, str, int]] | list[tuple[str, str, str, str, int]],
@@ -140,7 +181,10 @@ async def recompute_days(
        send = recv = 0
        last_ts: int | None = None
        last_is_send: bool | None = None
-        for ts, is_send in rows:
+        # 串算法需要严格时序:先排序(SQL 无 ORDER BY,旧代码遍历中 last_ts
+        # 用 >= 兜乱序;排序后该兜底语义不变——max ts 与并列末条方向一致)。
+        ordered = sorted(rows, key=lambda r: r[0])
+        for ts, is_send in ordered:
            if is_send is True:
                send += 1
            elif is_send is False:
@@ -148,6 +192,7 @@ async def recompute_days(
            if last_ts is None or ts >= last_ts:
                last_ts = int(ts)
                last_is_send = is_send
+        threads, answered, fr_sum = _thread_stats(ordered)
        await session.execute(
            sa.delete(MsgDailyStats).where(
                MsgDailyStats.channel == channel,
@@ -169,6 +214,9 @@ async def recompute_days(
                    recv_cnt=recv,
                    last_ts=last_ts,
                    last_is_send=last_is_send,
+                    cust_threads=threads,
+                    threads_answered=answered,
+                    fr_sum_ms=fr_sum,
                )
            )
        n += 1
@@ -224,7 +272,8 @@ async def recompute_all(session: AsyncSession) -> int:
            send = recv = 0
            last_ts = None
            last_is_send = None
-            for ts, is_send in day_rows:
+            ordered = sorted(day_rows, key=lambda r: r[0])  # 串算法需严格时序
+            for ts, is_send in ordered:
                if is_send is True:
                    send += 1
                elif is_send is False:
@@ -232,6 +281,7 @@ async def recompute_all(session: AsyncSession) -> int:
                if last_ts is None or ts >= last_ts:
                    last_ts = int(ts)
                    last_is_send = is_send
+            threads, answered, fr_sum = _thread_stats(ordered)
            session.add(
                MsgDailyStats(
                    channel=channel,
@@ -243,6 +293,9 @@ async def recompute_all(session: AsyncSession) -> int:
                    recv_cnt=recv,
                    last_ts=last_ts,
                    last_is_send=last_is_send,
+                    cust_threads=threads,
+                    threads_answered=answered,
+                    fr_sum_ms=fr_sum,
                )
            )
        total_pairs += 1
@@ -0,0 +1,62 @@
+"""首响时长+回复率聚合列(msg_daily_stats / stat_daily_staff 各 3 列)
+
+Revision ID: d9e0f1a2b3c4
+Revises: c3d4e5f6a7b8
+Create Date: 2026-09-07
+
+数据分析板块批1(A1+A2,2026-09-07):管理层要看「每个人的响应质量」。
+两表各加 cust_threads / threads_answered / fr_sum_ms——消息串算法见
+app/storage/summary.py(口径与 8-24 spec 对齐,日切近似 v1)。
+fr_cnt ≡ threads_answered(被回串数=样本数),不单列。
+
+幂等:sa.inspect 先查列存在则跳过(照 9b0d0c133035 惯例,SQLite/MySQL
+通用);downgrade 对称删列。模型侧 models.py 同步双声明,防 alembic
+autogenerate 反 DROP。NOT NULL + server_default 0:旧行/旧构造不传也不炸。
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision: str = "d9e0f1a2b3c4"
+down_revision: str | Sequence[str] | None = "c3d4e5f6a7b8"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_COLUMNS: tuple[tuple[str, str, sa.types.TypeEngine], ...] = (
+    ("msg_daily_stats", "cust_threads", sa.Integer()),
+    ("msg_daily_stats", "threads_answered", sa.Integer()),
+    ("msg_daily_stats", "fr_sum_ms", sa.BigInteger()),
+    ("stat_daily_staff", "cust_threads", sa.Integer()),
+    ("stat_daily_staff", "threads_answered", sa.Integer()),
+    ("stat_daily_staff", "fr_sum_ms", sa.BigInteger()),
+)
+
+
+def _has_column(bind, table: str, column: str) -> bool:
+    insp = sa.inspect(bind)
+    if not insp.has_table(table):
+        return False
+    return column in {c["name"] for c in insp.get_columns(table)}
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    bind = op.get_bind()
+    for table, column, type_ in _COLUMNS:
+        if _has_column(bind, table, column):
+            continue  # 幂等:已加过则跳过
+        op.add_column(
+            table,
+            sa.Column(column, type_, nullable=False, server_default=sa.text("0")),
+        )
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    bind = op.get_bind()
+    for table, column, _ in _COLUMNS:
+        if _has_column(bind, table, column):
+            op.drop_column(table, column)
@@ -205,3 +205,44 @@ async def test_recompute_all_and_rolling(make_client):
    await make_client(seed)
    days = staff_summary.rolling_days(20260904, window=3)
    assert days == [20260902, 20260903, 20260904]
+
+
+async def test_recompute_carries_thread_stats(make_client):
+    """首响/回复率三列(2026-09-07 批1)按账号→员工累加传递;未映射账号同样保留。"""
+
+    async def seed(session):
+        session.add(WechatAccount(wx_id="wx1", user_id="u1", user_name="张三"))
+        session.add(WechatAccount(wx_id="wx2", user_id="u1", user_name="张三"))
+        rows = [
+            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,
+                          cust_threads=4, threads_answered=3, fr_sum_ms=600_000),
+            MsgDailyStats(channel="wechat", owner_account_id="wx2", peer_kind="room",
+                          peer_id="r1", day=20260904, send_cnt=2, recv_cnt=1,
+                          last_ts=1756900001000, last_is_send=False,
+                          cust_threads=2, threads_answered=1, fr_sum_ms=240_000),
+            # 旧数据行(迁移前):三列 server_default 兜 0,聚合不炸
+            MsgDailyStats(channel="wechat", owner_account_id="wx3", peer_kind="friend",
+                          peer_id="t9", day=20260904, send_cnt=1, recv_cnt=0,
+                          last_ts=1756900002000, last_is_send=True),
+        ]
+        session.add_all(rows)
+        await session.commit()
+        await staff_summary.recompute_days(session, [20260904])
+        await session.commit()
+        from sqlalchemy import select
+
+        by_uid = {
+            r.user_id: r
+            for r in (await session.execute(select(StatDailyStaff))).scalars().all()
+        }
+        u1 = by_uid["u1"]  # wx1+wx2 合并:threads=6, answered=4, fr_sum=840000
+        assert u1.cust_threads == 6
+        assert u1.threads_answered == 4
+        assert u1.fr_sum_ms == 840_000
+        assert by_uid["wx3"].cust_threads == 0  # 未传三列 → server_default 0
+        assert by_uid["wx3"].threads_answered == 0
+        assert by_uid["wx3"].fr_sum_ms == 0
+
+    await make_client(seed)
@@ -1,6 +1,6 @@
 """Tests for the msg_daily_stats 汇总层(日粒度会话统计)。

-三个被钉死的行为:
+四个被钉死的行为:

 1. **日桶口径**:day 一律 Asia/Shanghai 自然日(Python 计算,不用 SQL 日期
   函数——旧实现 from_unixtime 跟随 MySQL 会话时区,UTC 容器下和今日数
@@ -10,6 +10,8 @@
   入口,一处挂全。幂等:同一批重放不产生重复/漂移。
 3. **全量重算**::func:`app.storage.summary.recompute_all` 从 message 表
   重建全部日行(夜间自愈/部署回填用),幂等。
+4. **消息串口径**(2026-09-07 批1 首响/回复率):cust_threads/
+   threads_answered/fr_sum_ms 三列的串算法矩阵——见下方测试组。

 fixtures 用内存 SQLite(测试对 message 的所有写入都走 upsert_messages,
 从而同时覆盖钩子行为)。
@@ -183,3 +185,153 @@ async def test_recompute_all_rebuilds_and_is_idempotent(session):

    await recompute_all(session)  # 幂等(重建后是全新 ORM 对象,比字段值不比对象)
    assert await _stats_rows(session) == rows
+
+
+# ── 消息串口径:首响/回复率三列(2026-09-07 批1)────────────────────────── #
+#
+# 口径(定案):串=客户连续消息(is_send=False)不被员工消息(is_send=True)
+# 打断的段;is_send=None 不参与串判定(跳过,不打断也不起串)。每个被回串
+# 一个样本 fr_ms=员工首条回复 ts−串末条客户消息 ts。日切近似:日末未闭合
+# 串计 threads 不计 answered,跨夜的串在次日段重新起算。
+
+
+async def _thread_rows(session):
+    rows = (await session.execute(select(MsgDailyStats))).scalars().all()
+    return {
+        (r.peer_kind, r.peer_id, r.day): (r.cust_threads, r.threads_answered, r.fr_sum_ms)
+        for r in rows
+    }
+
+
+async def _seed_thread(session, msgs) -> dict:
+    """写一批消息(走钩子→recompute_days),返回三列断言字典。"""
+    await upsert_messages(session, msgs)
+    await session.commit()
+    return await _thread_rows(session)
+
+
+async def test_thread_single_answered(session):
+    """单串被回:fr=回复−串首条客户消息(此处串只有一条)。"""
+    rows = await _seed_thread(
+        session,
+        [
+            _msg("c1", _sh(2026, 9, 3, 10, 0), is_send=False),
+            _msg("s1", _sh(2026, 9, 3, 10, 5)),
+        ],
+    )
+    # (cust_threads, threads_answered, fr_sum_ms)
+    assert rows[("friend", "t1", 20260903)] == (1, 1, 5 * 60 * 1000)
+
+
+async def test_thread_fr_takes_last_customer_msg(session):
+    """多条客户消息串内:fr 取「末条」客户消息(非首条)。"""
+    rows = await _seed_thread(
+        session,
+        [
+            _msg("c1", _sh(2026, 9, 3, 10, 0), is_send=False),
+            _msg("c2", _sh(2026, 9, 3, 10, 2), is_send=False),
+            _msg("c3", _sh(2026, 9, 3, 10, 7), is_send=False),
+            _msg("s1", _sh(2026, 9, 3, 10, 9)),
+        ],
+    )
+    assert rows[("friend", "t1", 20260903)] == (1, 1, 2 * 60 * 1000)  # 10:09−10:07
+
+
+async def test_two_threads_only_one_answered(session):
+    """两串只回一串:threads=2/answered=1,fr 只含被回串。"""
+    rows = await _seed_thread(
+        session,
+        [
+            _msg("c1", _sh(2026, 9, 3, 10, 0), is_send=False),
+            _msg("s1", _sh(2026, 9, 3, 10, 5)),
+            _msg("c2", _sh(2026, 9, 3, 11, 0), is_send=False),  # 新串,日末未闭合
+        ],
+    )
+    assert rows[("friend", "t1", 20260903)] == (2, 1, 5 * 60 * 1000)
+
+
+async def test_thread_not_broken_by_unknown(session):
+    """unknown(is_send=None)消息夹在串中:不打断串,fr 仍取末条客户消息。"""
+    rows = await _seed_thread(
+        session,
+        [
+            _msg("c1", _sh(2026, 9, 3, 10, 0), is_send=False),
+            _msg("u1", _sh(2026, 9, 3, 10, 1), is_send=None),
+            _msg("c2", _sh(2026, 9, 3, 10, 2), is_send=False),
+            _msg("s1", _sh(2026, 9, 3, 10, 5)),
+        ],
+    )
+    assert rows[("friend", "t1", 20260903)] == (1, 1, 3 * 60 * 1000)  # 10:05−10:02
+
+
+async def test_thread_open_at_day_end_counts_unanswered(session):
+    """日末未闭合串:计 threads 不计 answered(日切近似,fr 不含它)。"""
+    rows = await _seed_thread(
+        session,
+        [
+            _msg("c1", _sh(2026, 9, 3, 10, 0), is_send=False),
+            _msg("s1", _sh(2026, 9, 3, 10, 5)),
+            _msg("c2", _sh(2026, 9, 3, 23, 50), is_send=False),  # 日末未回
+        ],
+    )
+    assert rows[("friend", "t1", 20260903)] == (2, 1, 5 * 60 * 1000)
+
+
+async def test_staff_msg_first_is_not_a_thread(session):
+    """员工先发不算串:开场白式员工消息不产生 threads/answered。"""
+    rows = await _seed_thread(
+        session,
+        [
+            _msg("s0", _sh(2026, 9, 3, 9, 0)),
+            _msg("c1", _sh(2026, 9, 3, 10, 0), is_send=False),
+        ],
+    )
+    assert rows[("friend", "t1", 20260903)] == (1, 0, 0)
+
+
+async def test_thread_closed_then_reopened_and_multi_fr(session):
+    """闭合后再起串+多样本:fr_sum 累加各被回串的样本时长。"""
+    rows = await _seed_thread(
+        session,
+        [
+            _msg("c1", _sh(2026, 9, 3, 10, 0), is_send=False),
+            _msg("s1", _sh(2026, 9, 3, 10, 5)),  # 串1被回,fr=5min
+            _msg("s2", _sh(2026, 9, 3, 10, 8)),  # 串已闭合,追加员工消息无效
+            _msg("c2", _sh(2026, 9, 3, 11, 0), is_send=False),
+            _msg("c3", _sh(2026, 9, 3, 11, 4), is_send=False),
+            _msg("s3", _sh(2026, 9, 3, 11, 10)),  # 串2被回,fr=6min(11:10−11:04)
+        ],
+    )
+    assert rows[("friend", "t1", 20260903)] == (2, 2, 11 * 60 * 1000)
+
+
+async def test_thread_cross_day_restarts_next_day(session):
+    """日切近似:23:50 的串在当日计 open;次日段客户续发+被回在次日重新起算。"""
+    rows = await _seed_thread(
+        session,
+        [
+            _msg("c1", _sh(2026, 9, 3, 23, 50), is_send=False),  # 当日未回
+            _msg("c2", _sh(2026, 9, 4, 9, 0), is_send=False),  # 次日新串(同串跨夜近似)
+            _msg("s1", _sh(2026, 9, 4, 9, 3)),
+        ],
+    )
+    assert rows[("friend", "t1", 20260903)] == (1, 0, 0)
+    assert rows[("friend", "t1", 20260904)] == (1, 1, 3 * 60 * 1000)  # 9:03−9:00
+
+
+async def test_recompute_all_carries_thread_stats(session):
+    """全量重算路径(逐会话全史分桶)同样计算三值,且与钩子路径一致。"""
+    msgs = [
+        _msg("c1", _sh(2026, 9, 3, 10, 0), is_send=False),
+        _msg("u1", _sh(2026, 9, 3, 10, 1), is_send=None),
+        _msg("c2", _sh(2026, 9, 3, 10, 2), is_send=False),
+        _msg("s1", _sh(2026, 9, 3, 10, 5)),
+        _msg("c3", _sh(2026, 9, 3, 11, 0), is_send=False),
+    ]
+    await upsert_messages(session, msgs)
+    await session.commit()
+    hooked = await _thread_rows(session)
+
+    await recompute_all(session)
+    assert await _thread_rows(session) == hooked
+    assert hooked[("friend", "t1", 20260903)] == (2, 1, 3 * 60 * 1000)