diff --git a/yunke-server/app/api/main.py b/yunke-server/app/api/main.py index 01ac3e5..10055c3 100644 --- a/yunke-server/app/api/main.py +++ b/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) diff --git a/yunke-server/app/api/routers/staff_chat_stats.py b/yunke-server/app/api/routers/staff_chat_stats.py new file mode 100644 index 0000000..2dbc620 --- /dev/null +++ b/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, + ) diff --git a/yunke-server/app/api/schemas.py b/yunke-server/app/api/schemas.py index 9942d5a..bd84f53 100644 --- a/yunke-server/app/api/schemas.py +++ b/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] diff --git a/yunke-server/tests/test_api_staff_chat_stats.py b/yunke-server/tests/test_api_staff_chat_stats.py new file mode 100644 index 0000000..6ed7c3d --- /dev/null +++ b/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()