xinyouhe/yunke
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(management): 员工指标宽表——1:1口径/串类与fact同源/生命周期列/幽灵过滤/纯生命周期日出行

Browse Source 
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-09-09 14:19:25 +08:00
parent a38170d205
commit 2bb4cee76d
2 changed files with 328 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yunke-server/app/storage/employee_metrics.py
+216 -1
View File
@@ -12,12 +12,28 @@
from __future__ import annotations
import logging
import math
from bisect import bisect_right
from collections.abc import Iterable
import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import EmployeeCallDaily, Message, PhoneCall, Staff
from app.analytics.lifecycle import classify
from app.db.models import (
Customer,
EmployeeCallDaily,
EmployeeMetricDaily,
Message,
MsgDailyStats,
PhoneCall,
Staff,
)
from app.storage.first_reply_facts import (
compute_conversation,
load_conversation,
pairs_active_in_days,
)
from app.storage.staff_summary import account_map
from app.storage.summary import day_bounds_ms
@@ -131,3 +147,202 @@ async def recompute_call_days(session: AsyncSession, days: Iterable[int]) -> int
)
written += 1
return written
METRIC_VERSION = "v1"
def _percentile(vals: list[int], q: float) -> int | None:
"""最近邻秩分位 ceil(q·n);空列表→None(与 0 严格区分)。"""
if not vals:
return None
s = sorted(vals)
idx = min(len(s) - 1, max(0, math.ceil(q * len(s)) - 1))
return s[idx]
async def _lifecycle_daily(
session: AsyncSession,
days: list[int],
channel: str,
amap: dict[str, str],
valid_users: set[str],
) -> dict[int, dict[str, dict[str, int]]]:
"""逐日 {day: {user: {state: count}}}——customer 当前归属 + 互动日 sweep。
数据源:msg_daily_stats friend 行全历史(互动日/条数)。16k 客户 × 窗口
天数 bisect,夜间全量约秒级;4h tick 只算当日+3 天。
"""
rows = (
await session.execute(
sa.select(
MsgDailyStats.peer_id, MsgDailyStats.day,
MsgDailyStats.send_cnt, MsgDailyStats.recv_cnt,
).where(
MsgDailyStats.channel == channel,
MsgDailyStats.peer_kind == "friend",
)
)
).all()
days_map: dict[str, set[int]] = {}
cnt: dict[str, 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()}
custs = (
await session.execute(
sa.select(Customer.owner_account_id, Customer.friend_id).where(
Customer.channel == channel,
sa.or_(Customer.delete_flag.is_(None), Customer.delete_flag == 0),
Customer.owner_account_id.is_not(None),
)
)
).all()
peer_to_user: dict[str, str] = {}
for owner, peer in custs:
uid = amap.get(owner or "")
if uid and uid in valid_users and peer:
peer_to_user[str(peer)] = uid
out: dict[int, dict[str, dict[str, int]]] = {d: {} for d in days}
for peer, days_list in interaction.items():
uid = peer_to_user.get(peer)
if uid is None:
continue
first = days_list[0]
hist = cnt.get(peer, 0)
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)
if state:
out[d].setdefault(uid, {}).setdefault(state, 0)
out[d][uid][state] += 1
return out
async def recompute_metric_days(
session: AsyncSession,
days: Iterable[int],
channels: tuple[str, ...] = ("wechat", "whatsapp"),
) -> int:
"""重算给定日的宽表:按 (day, channel) delete+insert,幂等。返回写入行数。
串类列与首响日值:扫窗口内活跃 friend 会话的**全历史**(每会话一次),
与 first_reply_fact 同一 compute_conversation——分子分母同源(v2 §1.1 #5)。
"""
days = sorted(set(days))
valid_users = await known_user_ids(session)
pairs = await pairs_active_in_days(session, days)
pairs_by_channel: dict[str, list] = {}
for p in pairs:
pairs_by_channel.setdefault(p[0], []).append((p[1], p[2]))
written = 0
for channel in channels:
amap = await account_map(session, channel)
# ---- 串类+分位:逐会话全历史一次,聚合到 (day, user) ----
thread_cols: dict[tuple[int, str], dict] = {}
frs: dict[tuple[int, str], list[int]] = {}
day_set = set(days)
for owner, peer in pairs_by_channel.get(channel, []):
uid = amap.get(owner)
if uid is None or uid not in valid_users:
continue
rows = await load_conversation(session, channel, owner, peer)
facts, agg = compute_conversation(rows)
for d, a in agg.items():
if d not in day_set:
continue
key = (d, uid)
t = thread_cols.setdefault(
key, {"threads": 0, "answered": 0, "fr": 0}
)
t["threads"] += a.threads
t["answered"] += a.answered
t["fr"] += a.fr_sum_ms
frs.setdefault(key, []).extend(
f.fr_ms for f in facts if f.day == d
)
# ---- 生命周期 ----
life = await _lifecycle_daily(session, days, channel, amap, valid_users)
# ---- 消息计数+负载:msg_daily_stats(算法无关列) ----
for day in days:
mds = (
await session.execute(
sa.select(
MsgDailyStats.owner_account_id,
MsgDailyStats.peer_kind,
MsgDailyStats.peer_id,
MsgDailyStats.send_cnt,
MsgDailyStats.recv_cnt,
).where(
MsgDailyStats.day == day, MsgDailyStats.channel == channel
)
)
).all()
await session.execute(
sa.delete(EmployeeMetricDaily).where(
EmployeeMetricDaily.day == day,
EmployeeMetricDaily.channel == channel,
EmployeeMetricDaily.metric_version == METRIC_VERSION,
)
)
agg: dict[str, dict] = {}
for owner, kind, peer, snd, rcv in mds:
uid = amap.get(owner)
if uid is None or uid not in valid_users:
continue # 幽灵过滤
a = agg.setdefault(
uid,
{"sent": 0, "recv": 0, "friends": set(), "rooms": set()},
)
# 消息计数列 = 1:1 口径(v2 §1.1 #2):room 只进 active_rooms
if kind == "friend":
a["sent"] += int(snd or 0)
a["recv"] += int(rcv or 0)
a["friends"].add(peer)
elif kind == "room":
a["rooms"].add(peer)
day_life = life.get(day, {})
# 行并集:消息源 ∪ 生命周期源——纯生命周期日(员工当日无消息
# 但持有降温/沉睡客户)也要出宽表行,否则客户经营维度漏天
all_uids = sorted(set(agg) | set(day_life))
for uid in all_uids:
a = agg.get(
uid, {"sent": 0, "recv": 0, "friends": set(), "rooms": set()}
)
key = (day, uid)
t = thread_cols.get(key, {"threads": 0, "answered": 0, "fr": 0})
lv = day_life.get(uid, {})
session.add(
EmployeeMetricDaily(
day=day,
channel=channel,
user_id=uid,
metric_version=METRIC_VERSION,
msg_sent=a["sent"],
msg_recv=a["recv"],
active_friends=len(a["friends"]),
active_rooms=len(a["rooms"]),
cust_threads=t["threads"],
threads_answered=t["answered"],
fr_sum_ms=t["fr"],
first_reply_p50_ms=_percentile(frs.get(key, []), 0.5),
first_reply_p90_ms=_percentile(frs.get(key, []), 0.9),
new_customer_count=lv.get("new", 0),
cooling_customer_count=lv.get("cooling", 0),
dormant_customer_count=lv.get("dormant", 0),
reactivated_customer_count=lv.get("reactivated", 0),
data_complete=True,
)
)
written += 1
return written
yunke-server/tests/test_employee_metrics.py
+112 -1
View File
@@ -11,14 +11,18 @@ from zoneinfo import ZoneInfo
import sqlalchemy as sa
from app.db.models import (
Customer,
EmployeeCallDaily,
EmployeeMetricDaily,
FirstReplyFact,
Message,
MsgDailyStats,
PhoneCall,
Staff,
WechatAccount,
WhatsappAccount,
)
from app.storage.employee_metrics import recompute_call_days
from app.storage.employee_metrics import recompute_call_days, recompute_metric_days
_SH = ZoneInfo("Asia/Shanghai")
@@ -93,3 +97,110 @@ async def test_call_daily_ignores_unknown_staff(make_session):
await s.commit()
assert (await s.execute(
sa.select(sa.func.count()).select_from(EmployeeCallDaily))).scalar_one() == 0
async def _seed_msg_rows(s, *, day=20260901, channel="wechat", owner="wx1",
kind="friend", peer="p1", send=3, recv=2):
s.add(MsgDailyStats(channel=channel, owner_account_id=owner, peer_kind=kind,
peer_id=peer, day=day, send_cnt=send, recv_cnt=recv))
await s.commit()
async def test_metric_daily_friend_only_caliber(make_session):
"""口径判别(v2 §1.1 #2 防线):room 行不进消息计数/串列,只进 active_rooms。"""
s = make_session
await _seed_org(s)
await _seed_msg_rows(s, send=3, recv=2, kind="friend", peer="p1")
await _seed_msg_rows(s, send=10, recv=0, kind="room", peer="r1")
# friend 会话给一条消息供串扫描(无消息则无串)
s.add(Message(msg_id="p1-0", channel="wechat", owner_account_id="wx1",
talker_id="p1", timestamp=_ts(2026, 9, 1, 10, 0),
is_send=False, type="text", text="q"))
await s.commit()
await recompute_metric_days(s, [20260901])
await s.commit()
row = (await s.execute(sa.select(EmployeeMetricDaily))).scalar_one()
assert row.msg_sent == 3 and row.msg_recv == 2 # room 的 10 不进
assert row.active_rooms == 1
assert row.cust_threads == 1 and row.threads_answered == 0 # 未回复串
async def test_metric_daily_ghost_filtered(make_session):
s = make_session
await _seed_org(s)
await _seed_msg_rows(s, owner="wx-ghost", peer="pg")
await recompute_metric_days(s, [20260901])
await s.commit()
assert (await s.execute(
sa.select(sa.func.count()).select_from(EmployeeMetricDaily))).scalar_one() == 0
async def test_metric_daily_same_source_as_fact(make_session):
"""同源判别(v2 §1.1 #5 防线):
1) threads_answered ≡ fact 行数(同归属日);
2) 多日未回串宽表 threads < 日切 cust_threads 之和(分母不再膨胀);
3) 幂等。"""
from app.storage.first_reply_facts import recompute_pairs
s = make_session
await _seed_org(s)
# 客户 D1 10:00、D1 20:00、D2 09:00 三条消息,始终未回 → 1 个跨日串
rows = [(_ts(2026, 9, 1, 10, 0), False), (_ts(2026, 9, 1, 20, 0), False),
(_ts(2026, 9, 2, 9, 0), False)]
for i, (ts, is_send) in enumerate(rows):
s.add(Message(msg_id=f"g{i}", channel="wechat", owner_account_id="wx1",
talker_id="pg", timestamp=ts, is_send=is_send,
type="text", text="x"))
# 日切口径的 msg_daily_stats:D1、D2 各计一个未回串(供对照)
await _seed_msg_rows(s, day=20260901, peer="pg", send=0, recv=2)
await _seed_msg_rows(s, day=20260902, peer="pg", send=0, recv=1)
await s.commit()
await recompute_pairs(s, [("wechat", "wx1", "pg")])
await s.commit()
await recompute_metric_days(s, [20260901, 20260902])
await s.commit()
got = {r.day: r for r in (
await s.execute(sa.select(EmployeeMetricDaily))).scalars().all()}
# 1) fact 行数=0(未回),宽表 threads_answered 也=0 且同源
assert (await s.execute(
sa.select(sa.func.count()).select_from(FirstReplyFact))).scalar_one() == 0
# 2) 跨日串只归属 D2;日切 D1+D2=2,宽表合计=1
assert got[20260901].cust_threads == 0
assert got[20260902].cust_threads == 1
# 3) 幂等
await recompute_metric_days(s, [20260901, 20260902])
await s.commit()
assert len((await s.execute(
sa.select(EmployeeMetricDaily))).scalars().all()) == 2
async def test_metric_daily_lifecycle_counts(make_session):
"""生命周期列:cooling/dormant/reactivated 判定进宽表。"""
s = make_session
await _seed_org(s)
# 客户 pc:D1 有互动后静默(历史≥5 条) → as_of=10-10 为 dormant;
# 10-11 回流 → 该日 reactivated
s.add(Customer(friend_id="pc", channel="wechat", owner_account_id="wx1",
delete_flag=0))
hist = []
for h in range(9, 17): # D1 8 条消息(≥5)
hist.append((_ts(2026, 9, 1, h, 0), h % 2 == 0))
for i, (ts, is_send) in enumerate(hist):
s.add(Message(msg_id=f"lc{i}", channel="wechat", owner_account_id="wx1",
talker_id="pc", timestamp=ts, is_send=is_send,
type="text", text="x"))
await _seed_msg_rows(s, day=20260901, peer="pc", send=4, recv=4)
await s.commit()
# D41(10-11)客户回流一条
s.add(Message(msg_id="lc-b", channel="wechat", owner_account_id="wx1",
talker_id="pc", timestamp=_ts(2026, 10, 11, 9, 0),
is_send=False, type="text", text="back"))
await _seed_msg_rows(s, day=20261011, peer="pc", send=0, recv=1)
await s.commit()
from app.storage.first_reply_facts import recompute_pairs
await recompute_pairs(s, [("wechat", "wx1", "pc")])
await recompute_metric_days(s, [20261010, 20261011])
await s.commit()
got = {r.day: r for r in (
await s.execute(sa.select(EmployeeMetricDaily))).scalars().all()}
assert got[20261010].dormant_customer_count == 1 # 39 天静默+历史 8 条
assert got[20261011].reactivated_customer_count == 1 # 当日回流且 gap>30