docs(management): 批次A计划Task6代码块同步评审修正——hist时点语义/防御性flush

与 ad3beb7 实际实现对齐:day_msgs+cum 累积数组(hist=≤as_of,历史日沉睡门槛
不被未来消息虚增)、每日循环末 flush、accumulate import
This commit is contained in:
浅念
2026-09-09 14:45:58 +08:00
parent ad3beb7625
commit e725c7032a
@@ -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
```