diff --git a/yunke-server/app/db/models.py b/yunke-server/app/db/models.py index 8215dd6..cfad16f 100644 --- a/yunke-server/app/db/models.py +++ b/yunke-server/app/db/models.py @@ -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() diff --git a/yunke-server/app/storage/staff_summary.py b/yunke-server/app/storage/staff_summary.py index 53e6347..c136bbb 100644 --- a/yunke-server/app/storage/staff_summary.py +++ b/yunke-server/app/storage/staff_summary.py @@ -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 diff --git a/yunke-server/app/storage/summary.py b/yunke-server/app/storage/summary.py index 2bef64e..24e5667 100644 --- a/yunke-server/app/storage/summary.py +++ b/yunke-server/app/storage/summary.py @@ -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 diff --git a/yunke-server/migrations/versions/d9e0f1a2b3c4_fr_reply_rate_columns.py b/yunke-server/migrations/versions/d9e0f1a2b3c4_fr_reply_rate_columns.py new file mode 100644 index 0000000..00f10de --- /dev/null +++ b/yunke-server/migrations/versions/d9e0f1a2b3c4_fr_reply_rate_columns.py @@ -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) diff --git a/yunke-server/tests/test_staff_summary.py b/yunke-server/tests/test_staff_summary.py index ab7996b..393e515 100644 --- a/yunke-server/tests/test_staff_summary.py +++ b/yunke-server/tests/test_staff_summary.py @@ -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) diff --git a/yunke-server/tests/test_summary_daily_stats.py b/yunke-server/tests/test_summary_daily_stats.py index 07f36b3..10e5ab2 100644 --- a/yunke-server/tests/test_summary_daily_stats.py +++ b/yunke-server/tests/test_summary_daily_stats.py @@ -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)