From e725c7032a6bf188c3c3f43a737e0707f79e664b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Wed, 9 Sep 2026 14:45:58 +0800 Subject: [PATCH] =?UTF-8?q?docs(management):=20=E6=89=B9=E6=AC=A1A?= =?UTF-8?q?=E8=AE=A1=E5=88=92Task6=E4=BB=A3=E7=A0=81=E5=9D=97=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E8=AF=84=E5=AE=A1=E4=BF=AE=E6=AD=A3=E2=80=94=E2=80=94?= =?UTF-8?q?hist=E6=97=B6=E7=82=B9=E8=AF=AD=E4=B9=89/=E9=98=B2=E5=BE=A1?= =?UTF-8?q?=E6=80=A7flush?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 与 ad3beb7 实际实现对齐:day_msgs+cum 累积数组(hist=≤as_of,历史日沉睡门槛 不被未来消息虚增)、每日循环末 flush、accumulate import --- ...agement-console-batch-a-data-foundation.md | 24 ++++++++++++------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/docs/superpowers/plans/2026-09-08-management-console-batch-a-data-foundation.md b/docs/superpowers/plans/2026-09-08-management-console-batch-a-data-foundation.md index 3de0d68..abbee74 100644 --- a/docs/superpowers/plans/2026-09-08-management-console-batch-a-data-foundation.md +++ b/docs/superpowers/plans/2026-09-08-management-console-batch-a-data-foundation.md @@ -1627,6 +1627,7 @@ cd yunke-server && .venv/Scripts/python.exe -m pytest tests/test_employee_metric ```python import math # (文件头补充) from bisect import bisect_right +from itertools import accumulate from app.analytics.lifecycle import classify from app.db.models import Customer, EmployeeMetricDaily, MsgDailyStats @@ -1657,8 +1658,11 @@ async def _lifecycle_daily( ) -> dict[int, dict[str, dict[str, int]]]: """逐日 {day: {user: {state: count}}}——customer 当前归属 + 互动日 sweep。 - 数据源:msg_daily_stats friend 行全历史(互动日/条数)。16k 客户 × 窗口 - 天数 bisect,夜间全量约秒级;4h tick 只算当日+3 天。 + 数据源:msg_daily_stats friend 行全历史(互动日/条数)。hist 为 ≤as_of + 的历史消息数(时点语义:与 last 同一 bisect 位取按日累积数组 cum[idx], + as_of 之后的消息不虚增历史日的沉睡门槛——Task6 评审修正)。16k 客户 × + 窗口天数 bisect+classify,夜间全量实测见批次A核实文档;4h tick 只算 + 当日+3 天。 """ rows = ( await session.execute( @@ -1671,12 +1675,13 @@ async def _lifecycle_daily( ) ) ).all() - days_map: dict[str, set[int]] = {} - cnt: dict[str, int] = {} + # peer → {day: 消息条数};days_list 与累积数组 cum 对齐: + # cum[i] = 该客户 ≤days_list[i] 的历史消息数(hist 时点语义) + day_msgs: dict[str, dict[int, int]] = {} for peer, day, snd, rcv in rows: - days_map.setdefault(peer, set()).add(day) - cnt[peer] = cnt.get(peer, 0) + int(snd or 0) + int(rcv or 0) - interaction = {p: sorted(ds) for p, ds in days_map.items()} + dm = day_msgs.setdefault(peer, {}) + dm[day] = dm.get(day, 0) + int(snd or 0) + int(rcv or 0) + interaction = {p: sorted(dm) for p, dm in day_msgs.items()} custs = ( await session.execute( @@ -1699,14 +1704,14 @@ async def _lifecycle_daily( if uid is None: continue first = days_list[0] - hist = cnt.get(peer, 0) + cum = list(accumulate(day_msgs[peer][d] for d in days_list)) for d in days: idx = bisect_right(days_list, d) - 1 if idx < 0: continue last = days_list[idx] prev = days_list[idx - 1] if idx > 0 else None - state = classify(first, last, prev, d, hist) + state = classify(first, last, prev, d, cum[idx]) if state: out[d].setdefault(uid, {}).setdefault(state, 0) out[d][uid][state] += 1 @@ -1832,6 +1837,7 @@ async def recompute_metric_days( ) ) written += 1 + await session.flush() # 防御性:防 autoflush=False 配置下 ORM 对象堆积(Task6 评审修正) return written ```