test(management): 通话followup判别补强——未接通反例/渠道收窄正反例/任一账号

Co-Authored-By: Claude Code <noreply@anthropic.com>

yunke-server/app/api/routers/customer.py

@@ -108,9 +108,14 @@ async def _list_owner_links(
             .where(*conditions)
         )
     ).scalar_one()
+    # LEFT JOIN 档案带 raw_json(2026-09-09):WA 客户列表大量列只存在于
+    # raw_json(userName/salePhone/createTime 等员工侧字段),link 快照没存;
+    # raw_json 按 friend_id 全局一份,挂谁名下内容相同。注:共享客户的员工
+    # 侧字段是「最后写入员工」的记录,可能非本员工(客户自身字段不受影响)。
     rows = (
         await session.execute(
-            sa.select(CustomerAccountLink)
+            sa.select(CustomerAccountLink, Customer.raw_json, Customer.version)
+            .outerjoin(Customer, Customer.friend_id == CustomerAccountLink.friend_id)
             .where(*conditions)
             .order_by(
                 sa.nullslast(CustomerAccountLink.last_chat_time.desc()),
@@ -119,8 +124,14 @@ async def _list_owner_links(
             .limit(page_size)
             .offset((page - 1) * page_size)
         )
-    ).scalars().all()
-    items = [CustomerOut.model_validate(r) for r in rows]
+    ).all()
+    items = []
+    for link, raw_json, version in rows:
+        out = CustomerOut.model_validate(link)
+        out.raw_json = raw_json
+        if version is not None:
+            out.version = version
+        items.append(out)
     return Page(items=items, total=total, page=page, page_size=page_size)
 
 

yunke-server/app/storage/employee_metrics.py

@@ -0,0 +1,133 @@
+"""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 消息。
+
+    含客户发来的消息——口径=窗口内存在任一方向 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

yunke-server/tests/test_api_customers.py

@@ -226,3 +226,35 @@ async def test_owner_filter_with_dept_falls_back_to_archive(make_client):
     body = r.json()
     # 档案路径:只命中 user_id 属于 d1 的档案行(link 行不带 user_id)
     assert [it["friend_id"] for it in body["items"]] == ["f_archived"]
+
+
+async def test_owner_link_path_carries_archive_raw_json(make_client):
+    """link 路径按 friend_id 带出档案 raw_json(2026-09-09 WA 列表修复)。
+
+    WA 客户列表大量列只存在于 raw_json(userName/salePhone/createTime 等),
+    link 快照没存——不带 raw_json 时整表列空白。
+    """
+    from app.db.models import Customer as C, CustomerAccountLink as L
+
+    async def seed(session):
+        session.add_all([
+            # 档案行挂在别的员工名下(共享),raw_json 里带员工侧 camelCase 字段
+            C(channel="whatsapp", friend_id="f1@s.whatsapp.net",
+              owner_account_id="wx_other", friend_nickname="Nick",
+              raw_json={"userName": "员工WA名", "salePhone": "13800000000",
+                        "uid": "15600000000@s.whatsapp.net", "createTime": 1700000000000}),
+            L(channel="whatsapp", owner_account_id="wa_me",
+              friend_id="f1@s.whatsapp.net", friend_nickname="Nick", last_chat_time=100),
+        ])
+        await session.commit()
+
+    client = await make_client(seed)
+    r = await client.get("/api/customers", params={
+        "channel": "whatsapp", "owner_account_id": "wa_me"})
+    assert r.status_code == 200
+    body = r.json()
+    assert body["total"] == 1
+    it = body["items"][0]
+    assert it["friend_nickname"] == "Nick"          # link 快照
+    assert it["raw_json"]["userName"] == "员工WA名"  # 档案 raw_json 带出
+    assert it["raw_json"]["salePhone"] == "13800000000"

yunke-server/tests/test_employee_metrics.py

@@ -0,0 +1,95 @@
+"""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,
+    WhatsappAccount,
+)
+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"))
+    # 第二个微信号:验证 followup「任一账号」语义(owner-IN 而非单账号)
+    session.add(WechatAccount(wx_id="wx2", user_id="u1"))
+    # WA 账号:验证 wa_voice_call 渠道收窄正例
+    session.add(WhatsappAccount(uid="wa1", 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))  # 未接通
+    # c3:WA 语音,接通,窗口内有 WA 消息 → followup 正例(渠道收窄匹配)
+    s.add(PhoneCall(call_id="c3", channel="wa_voice_call", user_id="u1",
+                    call_number="p3", start_call_time=_ts(2026, 9, 1, 10, 0),
+                    call_duration=60))
+    # c4:微信语音,接通,但窗口内只有 WA 消息 → followup 反例(渠道收窄)
+    s.add(PhoneCall(call_id="c4", channel="voice_call", user_id="u1",
+                    call_number="p4", start_call_time=_ts(2026, 9, 1, 10, 0),
+                    call_duration=90))
+    # c1 后 2h 员工**另一微信号 wx2** 有跟进消息(任一账号语义:owner-IN)
+    s.add(Message(msg_id="f1", channel="wechat", owner_account_id="wx2",
+                  talker_id="p1", timestamp=_ts(2026, 9, 1, 12, 0),
+                  is_send=True, type="text", text="跟进"))
+    # c2(未接通)后 1h 有消息:即使任一方向也不算 followup(仅已接通判定)
+    s.add(Message(msg_id="f2", channel="wechat", owner_account_id="wx1",
+                  talker_id="p2", timestamp=_ts(2026, 9, 1, 16, 0),
+                  is_send=False, type="text", text="客户回拨?"))
+    # c3 后 2h WA 账号有消息 → 正例
+    s.add(Message(msg_id="f3", channel="whatsapp", owner_account_id="wa1",
+                  talker_id="p3", timestamp=_ts(2026, 9, 1, 12, 0),
+                  is_send=True, type="text", text="wa 跟进"))
+    # c4(voice_call→wechat)窗口内只有 WA 消息 → 不匹配渠道,反例
+    s.add(Message(msg_id="f4", channel="whatsapp", owner_account_id="wa1",
+                  talker_id="p4", 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 == 4
+    assert row.call_seconds == 270  # 120+0+60+90;duration 单位=秒(Task0 已核实)
+    assert row.answered_call_count == 3  # c1/c3/c4
+    # c1(wx2 任一账号)+ c3(wa 渠道正例);c2 未接通守卫、c4 渠道收窄反例
+    assert row.call_followup_count == 2
+
+
+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