From b2e1b699ebfcc1f2e95fd320322c68e03d0a7b50 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Wed, 9 Sep 2026 13:51:11 +0800 Subject: [PATCH] =?UTF-8?q?feat(management):=20=E9=80=9A=E8=AF=9D=E6=97=A5?= =?UTF-8?q?=E8=81=9A=E5=90=88=E5=B0=8F=E8=A1=A8=E2=80=94=E2=80=94=E6=97=A0?= =?UTF-8?q?=E6=B8=A0=E9=81=93=E7=BB=B4/=E6=8E=A5=E9=80=9A=3Dduration>0/24h?= =?UTF-8?q?=E4=BB=BB=E4=B8=80=E8=B4=A6=E5=8F=B7=E8=B7=9F=E8=BF=9B(?= =?UTF-8?q?=E6=B8=A0=E9=81=93=E6=8C=89=E9=80=9A=E8=AF=9D=E7=B1=BB=E5=9E=8B?= =?UTF-8?q?=E6=94=B6=E7=AA=84)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- yunke-server/app/storage/employee_metrics.py | 132 +++++++++++++++++++ yunke-server/tests/test_employee_metrics.py | 69 ++++++++++ 2 files changed, 201 insertions(+) create mode 100644 yunke-server/app/storage/employee_metrics.py create mode 100644 yunke-server/tests/test_employee_metrics.py diff --git a/yunke-server/app/storage/employee_metrics.py b/yunke-server/app/storage/employee_metrics.py new file mode 100644 index 0000000..dee656d --- /dev/null +++ b/yunke-server/app/storage/employee_metrics.py @@ -0,0 +1,132 @@ +"""employee_call_daily / employee_metric_daily 维护层(管理域,批次A)。 + +数据流(v2 §4): + msg_daily_stats(friend) + first_reply_facts 同源扫描 + customer 生命周期 + → employee_metric_daily (day, channel, user, metric_version) [Task 6] + phone_call(按 user_id+start_call_time 日聚合) + → employee_call_daily (day, user)——通话无渠道维(v2 §1.2 #9) [本任务] + +幽灵员工过滤(v2 §1.3 #11):owner 映射不上或不在 staff 表的不落表。 +重算语义:按 day delete+insert,幂等;夜间全量 + (批次F)4h 滚动。 +""" +from __future__ import annotations + +import logging +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.storage.staff_summary import account_map +from app.storage.summary import day_bounds_ms + +logger = logging.getLogger(__name__) + +# 通话类型 → 消息渠道(Task0 核实:channel 列可判别,voice_call=微信/ +# wa_voice_call=WA;followup 消息查询按此收窄) +_CALL_CHANNEL_TO_MSG: dict[str, str] = {"voice_call": "wechat", "wa_voice_call": "whatsapp"} + + +async def known_user_ids(session: AsyncSession) -> set[str]: + """员工主数据 user_id 集合(staff 表;Task0 已核实 41 行全在职)。""" + rows = (await session.execute(sa.select(Staff.user_id))).scalars().all() + return {u for u in rows if u} + + +async def _followup_exists( + session: AsyncSession, + msg_channel: str, + owner_accounts: list[str], + peer: str, + call_ts_ms: int, + window_ms: int = 86_400_000, +) -> bool: + """通话后 24h 内员工任一账号对该客户有 1:1 消息。 + + 按通话类型收窄消息渠道(Task0:voice_call→wechat/wa_voice_call→whatsapp, + 1:1 通话可匹配率 100%,@chatroom 群通话天然不匹配 1:1 消息)。""" + if not owner_accounts or not peer: + return False + n = ( + await session.execute( + sa.select(sa.func.count()).select_from(Message).where( + Message.channel == msg_channel, + Message.owner_account_id.in_(owner_accounts), + Message.talker_id == peer, + sa.or_(Message.room_id.is_(None), Message.room_id == ""), + Message.timestamp.is_not(None), + Message.timestamp > call_ts_ms, + Message.timestamp <= call_ts_ms + window_ms, + ) + ) + ).scalar_one() + return bool(n) + + +async def recompute_call_days(session: AsyncSession, days: Iterable[int]) -> int: + """重算给定日的通话小表:按 day delete+insert,幂等。返回写入行数。 + + 口径:接通 = call_duration > 0(Task0:与 call_status=1 差 31 条/1.0%, + 取 dur>0 与现有 roadmap 口径一致);call_seconds 单位=秒(Task0 已核实); + followup 只对已接通通话判定(未接通的回拨语义留二期)。 + """ + # 员工→该渠道账号(followup 判定用,按消息渠道分组) + acct = {ch: await account_map(session, ch) for ch in ("wechat", "whatsapp")} + rev: dict[str, dict[str, list[str]]] = {} # uid → msg_channel → owners + for msg_ch, amap in acct.items(): + for owner, uid in amap.items(): + rev.setdefault(uid, {}).setdefault(msg_ch, []).append(owner) + + written = 0 + staff_ids = await known_user_ids(session) # 幽灵过滤(v2 §1.3 #11) + for day in sorted(set(days)): + lo, hi = day_bounds_ms(day) + calls = ( + await session.execute( + sa.select( + PhoneCall.user_id, + PhoneCall.channel, + PhoneCall.call_number, + PhoneCall.start_call_time, + PhoneCall.call_duration, + ).where( + PhoneCall.start_call_time.is_not(None), + PhoneCall.start_call_time >= lo, + PhoneCall.start_call_time < hi, + ) + ) + ).all() + await session.execute( + sa.delete(EmployeeCallDaily).where(EmployeeCallDaily.day == day) + ) + # (uid, call_type, number, ts, dur)——msg_channel 按通话类型映射; + # 未知通话类型(如蜂窝 phone_call)无消息渠道可查,followup 判 + # False 但仍计数 + agg: dict[str, dict] = {} + for uid, call_ch, number, ts, dur in calls: + if not uid or uid not in staff_ids: + continue + a = agg.setdefault(uid, {"cnt": 0, "sec": 0, "answered": 0, "followup": 0}) + a["cnt"] += 1 + a["sec"] += int(dur or 0) + if (dur or 0) > 0: + a["answered"] += 1 + msg_ch = _CALL_CHANNEL_TO_MSG.get(call_ch or "") + if msg_ch and number: + owners = rev.get(uid, {}).get(msg_ch, []) + if await _followup_exists(session, msg_ch, owners, str(number), int(ts)): + a["followup"] += 1 + for uid, a in sorted(agg.items()): + session.add( + EmployeeCallDaily( + day=day, + user_id=uid, + call_count=a["cnt"], + call_seconds=a["sec"], + answered_call_count=a["answered"], + call_followup_count=a["followup"], + ) + ) + written += 1 + return written diff --git a/yunke-server/tests/test_employee_metrics.py b/yunke-server/tests/test_employee_metrics.py new file mode 100644 index 0000000..4fdc2d8 --- /dev/null +++ b/yunke-server/tests/test_employee_metrics.py @@ -0,0 +1,69 @@ +"""employee_call_daily / employee_metric_daily 生成判别测试(批次A Task5/6)。 + +判据(v2 §1.1 #2/#5、§1.2 #9、§1.3 #11): +- 通话无渠道维,独立小表;接通=duration>0;followup=24h 内任一账号有消息; +- 幽灵员工(owner 映射不上/不在 staff)过滤; +- 宽表消息计数 friend-only;串类与 fact 同源。 +""" +import datetime as dt +from zoneinfo import ZoneInfo + +import sqlalchemy as sa + +from app.db.models import ( + EmployeeCallDaily, + Message, + PhoneCall, + Staff, + WechatAccount, +) +from app.storage.employee_metrics import recompute_call_days + +_SH = ZoneInfo("Asia/Shanghai") + + +def _ts(*args) -> int: + return int(dt.datetime(*args, tzinfo=_SH).timestamp() * 1000) + + +async def _seed_org(session): + session.add(Staff(user_id="u1", name="张三", is_on_job=True)) + session.add(WechatAccount(wx_id="wx1", user_id="u1")) + await session.commit() + + +async def test_call_daily_aggregates(make_session): + s = make_session + await _seed_org(s) + s.add(PhoneCall(call_id="c1", channel="voice_call", user_id="u1", + call_number="p1", start_call_time=_ts(2026, 9, 1, 10, 0), + call_duration=120)) + s.add(PhoneCall(call_id="c2", channel="voice_call", user_id="u1", + call_number="p2", start_call_time=_ts(2026, 9, 1, 15, 0), + call_duration=0)) # 未接通 + # c1 后 2h 员工有跟进消息(多账号语义:u1 只有 wx1) + s.add(Message(msg_id="f1", channel="wechat", owner_account_id="wx1", + talker_id="p1", timestamp=_ts(2026, 9, 1, 12, 0), + is_send=True, type="text", text="跟进")) + await s.commit() + n = await recompute_call_days(s, [20260901]) + await s.commit() + assert n == 1 + row = (await s.execute(sa.select(EmployeeCallDaily))).scalar_one() + assert row.call_count == 2 + assert row.call_seconds == 120 # duration 单位=秒(Task0 已核实) + assert row.answered_call_count == 1 + assert row.call_followup_count == 1 + + +async def test_call_daily_ignores_unknown_staff(make_session): + s = make_session + await _seed_org(s) + s.add(PhoneCall(call_id="c9", channel="phone_call", user_id="ghost", + call_number="p9", start_call_time=_ts(2026, 9, 1, 10, 0), + call_duration=60)) + await s.commit() + await recompute_call_days(s, [20260901]) + await s.commit() + assert (await s.execute( + sa.select(sa.func.count()).select_from(EmployeeCallDaily))).scalar_one() == 0