feat(api): staff-chat-stats 聚合端点（微信员工 5 数字 + 客户明细，全本地 message）

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

yunke-server/app/api/main.py

@@ -46,6 +46,7 @@ from app.api.routers import (
     organization,
     search,
     sms,
+    staff_chat_stats,
     sync,
 )
 
@@ -65,6 +66,7 @@ app.include_router(messages.router)
 app.include_router(groups.router)
 app.include_router(calls.router)
 app.include_router(sms.router)
+app.include_router(staff_chat_stats.router)
 app.include_router(sync.router)
 app.include_router(search.router)
 app.include_router(export.router)

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

@@ -0,0 +1,149 @@
+"""GET /api/staff-chat-stats — 微信员工 5 数字 + 客户明细（本地 message 表实时聚合）。
+
+数据全本地：按 员工微信 wxid × ms-epoch 日期窗 聚合 1:1 私聊（room_id 过滤），
+不访问云客。5 汇总数从 customers 在 Python 侧派生（spec §5）。
+跨库日期提取（codebase 首例）镜像 app/storage/repos.py 的方言检测写法。
+"""
+from __future__ import annotations
+
+import time
+from typing import Annotated
+
+import sqlalchemy as sa
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.deps import get_session
+from app.api.schemas import StaffChatCustomerOut, StaffChatStatsOut, StaffChatStatsSummary
+from app.db.models import Customer, Message
+
+router = APIRouter(prefix="/api", tags=["staff-chat-stats"])
+
+SessionDep = Annotated[AsyncSession, Depends(get_session)]
+
+_MS_PER_DAY = 86_400_000
+
+
+def _private_window(msg, owner: str, start: int, end: int):
+    """1:1 私聊 + 窗口过滤条件（channel 固定 wechat），排除 timestamp IS NULL。"""
+    return [
+        msg.channel == "wechat",
+        msg.owner_account_id == owner,
+        sa.or_(msg.room_id.is_(None), msg.room_id == ""),
+        msg.timestamp.is_not(None),
+        msg.timestamp >= start,
+        msg.timestamp <= end,
+    ]
+
+
+async def _aggregate(session: AsyncSession, owner: str, start: int, end: int):
+    """返回 [(talker_id, send, recv, interaction_days, last_time)] + last_is_send map。"""
+    # 跨库日期提取：从 ms-epoch 取日期。镜像 repos.py 方言检测。
+    dialect = session.bind.name  # 镜像 app/storage/repos.py（.name 等价 .dialect.name）
+    if dialect == "sqlite":
+        day_expr = sa.func.date(Message.timestamp / 1000, "unixepoch")
+    else:  # mysql（生产）
+        day_expr = sa.func.date(sa.func.from_unixtime(Message.timestamp / 1000))
+
+    base = _private_window(Message, owner, start, end)
+    agg_stmt = (
+        sa.select(
+            Message.talker_id,
+            sa.func.sum(sa.case((Message.is_send.is_(True), 1), else_=0)).label("send_count"),
+            sa.func.sum(sa.case((Message.is_send.is_(False), 1), else_=0)).label("receive_count"),
+            sa.func.count(sa.func.distinct(day_expr)).label("interaction_days"),
+            sa.func.max(Message.timestamp).label("last_time"),
+        )
+        .where(*base)
+        .group_by(Message.talker_id)
+    )
+    agg_rows = (await session.execute(agg_stmt)).all()
+
+    # 每客户最后一条消息方向：ROW_NUMBER 窗口取 rn=1（tie-break msg_id 保证确定性）
+    inner = (
+        sa.select(
+            Message.talker_id,
+            Message.is_send,
+            sa.func.row_number()
+            .over(
+                partition_by=Message.talker_id,
+                order_by=(Message.timestamp.desc(), Message.msg_id.desc()),
+            )
+            .label("rn"),
+        )
+        .where(*base)
+    )
+    sub = inner.subquery("last_msg")
+    last_stmt = sa.select(sub.c.talker_id, sub.c.is_send).where(sub.c.rn == 1)
+    last_map = {tid: is_send for tid, is_send in (await session.execute(last_stmt)).all()}
+
+    return agg_rows, last_map
+
+
+async def _customer_names(session: AsyncSession, owner: str, talker_ids: list[str]) -> dict[str, str]:
+    if not talker_ids:
+        return {}
+    stmt = sa.select(
+        Customer.friend_id,
+        Customer.friend_remark,
+        Customer.friend_nickname,
+        Customer.real_name,
+    ).where(
+        Customer.channel == "wechat",
+        Customer.owner_account_id == owner,
+        Customer.friend_id.in_(talker_ids),
+    )
+    names: dict[str, str] = {}
+    for fid, remark, nick, real in (await session.execute(stmt)).all():
+        names[fid] = remark or nick or real or fid
+    return names
+
+
+@router.get("/staff-chat-stats")
+async def get_staff_chat_stats(
+    session: SessionDep,
+    owner_account_id: Annotated[str, Query(min_length=1)],
+    start: Annotated[int | None, Query(ge=0)] = None,
+    end: Annotated[int | None, Query(ge=0)] = None,
+) -> StaffChatStatsOut:
+    """微信员工沟通统计：5 数字 + 客户明细（全本地 message 聚合，1:1 私聊）。"""
+    if end is None:
+        end = int(time.time() * 1000)
+    if start is None:
+        start = end - 3 * _MS_PER_DAY  # 默认近 3 天
+
+    agg_rows, last_map = await _aggregate(session, owner_account_id, start, end)
+    talker_ids = [r.talker_id for r in agg_rows]
+    names = await _customer_names(session, owner_account_id, talker_ids)
+
+    customers: list[StaffChatCustomerOut] = []
+    for r in agg_rows:
+        tid = r.talker_id
+        last_is_send = last_map.get(tid)  # bool | None（None=最后一条方向未知）
+        customers.append(
+            StaffChatCustomerOut(
+                talker_id=tid,
+                friend_name=names.get(tid, tid),
+                send_count=int(r.send_count or 0),
+                receive_count=int(r.receive_count or 0),
+                interaction_days=int(r.interaction_days or 0),
+                # True→1, False→0, None→None（方向未知不计入两率；summary 用 ==1/==0 自动排除 None）
+                last_is_send=int(last_is_send) if last_is_send is not None else None,
+                last_time=r.last_time,
+            )
+        )
+
+    summary = StaffChatStatsSummary(
+        chat_count=len(customers),
+        effective_count=sum(1 for c in customers if c.send_count >= 1 and c.receive_count >= 1),
+        follow_count=sum(1 for c in customers if c.interaction_days >= 2),
+        last_reply_count=sum(1 for c in customers if c.last_is_send == 1),
+        no_reply_count=sum(1 for c in customers if c.last_is_send == 0),
+    )
+    return StaffChatStatsOut(
+        owner_account_id=owner_account_id,
+        start=start,
+        end=end,
+        summary=summary,
+        customers=customers,
+    )

yunke-server/app/api/schemas.py

@@ -290,3 +290,38 @@ class SyncStatusOut(BaseModel):
 
     states: list[SyncStateOut]
     recent_runs: list[SyncRunOut]
+
+
+# ---------------------------------------------------------------------------
+# Staff chat stats (WeChat staff 5-number summary + customer detail)
+# ---------------------------------------------------------------------------
+
+
+class StaffChatCustomerOut(BaseModel):
+    """单个客户在日期窗内的私聊聚合行（本地 message 表派生）。"""
+
+    talker_id: str
+    friend_name: str  # remark > nickname > real_name > talker_id
+    send_count: int  # 员工发出（is_send True）
+    receive_count: int  # 客户发出（is_send False）
+    interaction_days: int  # 窗口内不同日期数
+    last_is_send: int | None  # 1=员工最后发, 0=客户最后发, None=最后一条方向未知（不计入两率）
+    last_time: int | None = None  # ms-epoch，最后一条时间
+
+
+class StaffChatStatsSummary(BaseModel):
+    """5 个汇总数字（从 customers 派生）。"""
+
+    chat_count: int  # 沟通人数：窗口内 ≥1 条消息的去重客户数
+    effective_count: int  # 有效沟通数：send≥1 且 receive≥1 的客户数
+    follow_count: int  # 跟进客户数：interaction_days≥2 的客户数（近似口径）
+    last_reply_count: int  # 员工最后回复数：last_is_send=1 的客户数
+    no_reply_count: int  # 未回复客户数：last_is_send=0 的客户数
+
+
+class StaffChatStatsOut(BaseModel):
+    owner_account_id: str
+    start: int  # ms-epoch
+    end: int  # ms-epoch
+    summary: StaffChatStatsSummary
+    customers: list[StaffChatCustomerOut]

yunke-server/tests/test_api_staff_chat_stats.py

@@ -0,0 +1,137 @@
+"""Tests for GET /api/staff-chat-stats — 微信员工 5 数字 + 客户明细（本地 message 聚合）。"""
+from __future__ import annotations
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models import Customer, Message
+
+OWNER = "wx_owner_1"
+WIN_START = 1_700_000_000_000  # 固定 ms-epoch，确定性
+WIN_END = WIN_START + 3 * 86_400_000
+DAY = 86_400_000
+
+
+async def _seed(session: AsyncSession) -> None:
+    session.add_all(
+        [
+            # 客户 A：跨 2 天双向 → effective + follow，员工最后发
+            Message(msg_id="a1", channel="wechat", owner_account_id=OWNER, talker_id="t_a",
+                    room_id=None, is_send=True, timestamp=WIN_START, type="text", type_raw=1, text="x"),
+            Message(msg_id="a2", channel="wechat", owner_account_id=OWNER, talker_id="t_a",
+                    room_id=None, is_send=False, timestamp=WIN_START + DAY, type="text", type_raw=1, text="y"),
+            Message(msg_id="a3", channel="wechat", owner_account_id=OWNER, talker_id="t_a",
+                    room_id=None, is_send=True, timestamp=WIN_START + DAY + 1000, type="text", type_raw=1, text="z"),
+            # 客户 B：仅接收 → 非 effective，客户最后发（未回复）
+            Message(msg_id="b1", channel="wechat", owner_account_id=OWNER, talker_id="t_b",
+                    room_id=None, is_send=False, timestamp=WIN_START + 2000, type="text", type_raw=1, text="hi"),
+            # 客户 C：单条员工发 → 员工最后发，1 天
+            Message(msg_id="c1", channel="wechat", owner_account_id=OWNER, talker_id="t_c",
+                    room_id=None, is_send=True, timestamp=WIN_START + 3000, type="text", type_raw=1, text="c"),
+            # 群消息（room_id 非空）——必须被全部排除
+            Message(msg_id="g1", channel="wechat", owner_account_id=OWNER, talker_id="t_grp_member",
+                    room_id="room123@chatroom", is_send=True, timestamp=WIN_START + 4000, type="text", type_raw=1, text="g"),
+            # 窗口外私聊——必须排除（不污染 t_a 统计）
+            Message(msg_id="o1", channel="wechat", owner_account_id=OWNER, talker_id="t_a",
+                    room_id=None, is_send=True, timestamp=WIN_END + 1000, type="text", type_raw=1, text="future"),
+            # 其他 owner——必须排除
+            Message(msg_id="x1", channel="wechat", owner_account_id="wx_other", talker_id="t_a",
+                    room_id=None, is_send=True, timestamp=WIN_START + 5000, type="text", type_raw=1, text="other"),
+            Customer(friend_id="t_a", channel="wechat", owner_account_id=OWNER,
+                     friend_nickname="Alice", friend_remark="Alice备注", real_name=None),
+            Customer(friend_id="t_b", channel="wechat", owner_account_id=OWNER,
+                     friend_nickname="Bob", friend_remark=None, real_name=None),
+            # t_c 无 customer 行 → 昵称 fallback 为 talker_id "t_c"
+        ]
+    )
+    await session.commit()
+
+
+async def test_summary_five_numbers(make_client):
+    client = await make_client(_seed)
+    r = await client.get(
+        "/api/staff-chat-stats",
+        params={"owner_account_id": OWNER, "start": WIN_START, "end": WIN_END},
+    )
+    assert r.status_code == 200
+    s = r.json()["summary"]
+    assert s == {
+        "chat_count": 3,        # t_a, t_b, t_c（群/窗外/他owner 全排除）
+        "effective_count": 1,   # 仅 t_a 双向
+        "follow_count": 1,      # 仅 t_a 跨 2 天
+        "last_reply_count": 2,  # t_a, t_c 员工最后发
+        "no_reply_count": 1,    # t_b 客户最后发
+    }
+
+
+async def test_customers_fields_and_names(make_client):
+    client = await make_client(_seed)
+    r = await client.get(
+        "/api/staff-chat-stats",
+        params={"owner_account_id": OWNER, "start": WIN_START, "end": WIN_END},
+    )
+    by_id = {c["talker_id"]: c for c in r.json()["customers"]}
+    assert set(by_id) == {"t_a", "t_b", "t_c"}
+    # t_a: send=2 recv=1 days=2 last_is_send=1 last_time=a3
+    assert by_id["t_a"]["friend_name"] == "Alice备注"  # remark 优先
+    assert by_id["t_a"]["send_count"] == 2
+    assert by_id["t_a"]["receive_count"] == 1
+    assert by_id["t_a"]["interaction_days"] == 2
+    assert by_id["t_a"]["last_is_send"] == 1
+    assert by_id["t_a"]["last_time"] == WIN_START + DAY + 1000
+    # t_b
+    assert by_id["t_b"]["friend_name"] == "Bob"
+    assert by_id["t_b"]["send_count"] == 0
+    assert by_id["t_b"]["receive_count"] == 1
+    assert by_id["t_b"]["interaction_days"] == 1
+    assert by_id["t_b"]["last_is_send"] == 0
+    # t_c 无 customer 行 → fallback
+    assert by_id["t_c"]["friend_name"] == "t_c"
+    assert by_id["t_c"]["last_is_send"] == 1
+
+
+async def test_group_messages_excluded(make_client):
+    client = await make_client(_seed)
+    r = await client.get(
+        "/api/staff-chat-stats",
+        params={"owner_account_id": OWNER, "start": WIN_START, "end": WIN_END},
+    )
+    ids = {c["talker_id"] for c in r.json()["customers"]}
+    assert "t_grp_member" not in ids  # 群消息被 room_id 过滤
+
+
+async def test_date_window_and_owner_filter(make_client):
+    """窗外消息与他 owner 消息不污染统计（t_a 未被 o1/x1 抬高）。"""
+    client = await make_client(_seed)
+    r = await client.get(
+        "/api/staff-chat-stats",
+        params={"owner_account_id": OWNER, "start": WIN_START, "end": WIN_END},
+    )
+    by_id = {c["talker_id"]: c for c in r.json()["customers"]}
+    assert by_id["t_a"]["send_count"] == 2  # 非 4（a1+a3，排除 o1+x1）
+    assert by_id["t_a"]["last_time"] == WIN_START + DAY + 1000  # 非 o1
+
+
+async def test_empty_when_no_messages(make_client):
+    async def _empty(session: AsyncSession) -> None:
+        await session.commit()
+
+    client = await make_client(_empty)
+    r = await client.get(
+        "/api/staff-chat-stats",
+        params={"owner_account_id": OWNER, "start": WIN_START, "end": WIN_END},
+    )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["customers"] == []
+    assert body["summary"] == {
+        "chat_count": 0, "effective_count": 0, "follow_count": 0,
+        "last_reply_count": 0, "no_reply_count": 0,
+    }
+
+
+async def test_default_window_returns_ok(make_client):
+    """省略 start/end 走默认近 3 天；2023 固定种子在默认窗外 → 空集，仅验结构。"""
+    client = await make_client(_seed)
+    r = await client.get("/api/staff-chat-stats", params={"owner_account_id": OWNER})
+    assert r.status_code == 200
+    assert "summary" in r.json() and "customers" in r.json()