From b95fc5fa6fd9881d6ca46fcbba16e81b58b010ce Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Tue, 25 Aug 2026 16:22:44 +0800 Subject: [PATCH] =?UTF-8?q?feat(api):=20GET=20/api/staff-stats=207?= =?UTF-8?q?=E7=B1=BB=E8=AE=A1=E6=95=B0+=E5=88=86=E5=9E=8B=E6=98=8E?= =?UTF-8?q?=E7=BB=86(voice=E8=B5=B0user=5Fid=20join,=E6=96=B0=E5=AE=A2?= =?UTF-8?q?=E6=88=B7add=5Ftime=E5=85=9C=E5=BA=95,=E7=BA=A2=E5=8C=85?= =?UTF-8?q?=E9=87=91=E9=A2=9D=E6=8F=90=E5=8F=96)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 计数模式(无kind)返回 new_customers/new_groups/red_transfer/del_customers/ blocked_customers/voice_calls/sensitive_words(恒0,spec占位);明细模式 (?kind=)返回{kind,rows},每kind字段不同,前端按kind配列。start/end 缺省近7天。 口径(spec §4.5,全本地聚合): - new_customers: add_time窗内;IS NULL行用该对话首条消息min(timestamp)兜底 - new_groups: chat_group.create_time窗内(复合PK按owner过滤) - red_transfer: wechat type IN (red_packet,transfer)窗内;金额三路正则 (红包XML feedesc/转账纯文本/totalamount),实测真库转账399/399全提取 - del/blocked_customers: 现存量口径忽略窗口,明细time=updated_at毫秒 - voice_calls: phone_call.user_id存员工userId(32HEX)非wxId——先经 wechat_account.user_id换算,账号无uid返回空防跨员工串数 - voice明细字段语义对齐CallList.vue(connected=call_status==1, media_url本地/api/media/{hash}优先OSS兜底) 真机smoke(yunke-mysql):voice 61条/新客67/删除29/拉黑1均通过; 转账金额100%提取。真库chat_group列与message.type回填滞后(运维项,另报)。 Co-Authored-By: Claude --- yunke-server/app/api/main.py | 3 + yunke-server/app/api/routers/staff_stats.py | 407 ++++++++++++++++++++ yunke-server/app/api/schemas.py | 33 ++ yunke-server/tests/test_api_staff_stats.py | 315 +++++++++++++++ 4 files changed, 758 insertions(+) create mode 100644 yunke-server/app/api/routers/staff_stats.py create mode 100644 yunke-server/tests/test_api_staff_stats.py diff --git a/yunke-server/app/api/main.py b/yunke-server/app/api/main.py index a9db433..dfe4141 100644 --- a/yunke-server/app/api/main.py +++ b/yunke-server/app/api/main.py @@ -20,6 +20,7 @@ Includes: * ``GET /api/sync/status`` (see :mod:`app.api.routers.sync`). * ``GET /api/search`` (see :mod:`app.api.routers.search`). * ``GET /api/export/chat-html`` (see :mod:`app.api.routers.export`). +* ``GET /api/staff-stats`` (see :mod:`app.api.routers.staff_stats`). CORS is configured to read explicit allowed origins from ``Settings.cors_origins`` (comma-separated ``CORS_ORIGINS`` env var). Empty (default) falls back to the dev @@ -53,6 +54,7 @@ from app.api.routers import ( sessions, sms, staff_chat_stats, + staff_stats, sync, voice_clips, ) @@ -88,6 +90,7 @@ 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(staff_stats.router) app.include_router(sync.router) app.include_router(search.router) app.include_router(sessions.router) diff --git a/yunke-server/app/api/routers/staff_stats.py b/yunke-server/app/api/routers/staff_stats.py new file mode 100644 index 0000000..940019d --- /dev/null +++ b/yunke-server/app/api/routers/staff_stats.py @@ -0,0 +1,407 @@ +"""GET /api/staff-stats — 员工聊天窗口 7 类计数 + 分型明细(本地聚合,spec §4.5)。 + +对齐云客员工聊天窗口顶部 7 个统计数字与其点击弹出的分型明细列表: +* 计数模式(无 ``kind``)→ ``{new_customers, new_groups, red_transfer, + del_customers, blocked_customers, voice_calls, sensitive_words}`` + (``sensitive_words`` 恒 0——敏感词占位,spec 决策)。 +* 明细模式(``?kind=``)→ ``{kind, rows}``,每 kind 字段不同(统一 + ``list[dict]``,前端按 kind 配置列)。 + +统计口径(spec §4.5,全部本地聚合,逐条实测验证): +* new_customers ``customer.add_time`` 窗内;``add_time IS NULL`` 的行用 + 「该对话首条消息时间」兜底(message 同 owner+talker 的 + ``min(timestamp)`` 在窗内即算,~38% 行靠兜底)。 +* new_groups ``chat_group.create_time`` 窗内(Task 3 新列, + raw_json createTime 回填);无值行不计。 +* red_transfer wechat 消息 type IN (red_packet, transfer) 且 + timestamp 窗内(Task 4 已回填类型);金额从 text + 红包 XML ```` / 转账纯文本「发送转账：￥N」提取。 +* del_customers ``delete_flag=1`` 现存量(无时间字段→忽略窗口; + 明细行 time 用 updated_at 毫秒)。 +* blocked_customers ``is_block=1`` 现存量(同上)。 +* voice_calls ``phone_call.user_id`` 存的是员工 userId(32位HEX)而非 + wxId(实测坑!)——必须先经 ``wechat_account.user_id`` + 换算再查,直查 wxId 恒 0。 +* sensitive_words 恒 0(占位)。 + +start/end 缺省 = 近 7 天(对齐云客统计默认窗)。结构镜像 +:mod:`app.api.routers.staff_chat_stats`(SessionDep + 每类一个私有 +聚合函数 + 主端点分派)。 +""" +from __future__ import annotations + +import re +import time +from typing import Annotated, Any, Literal + +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 StaffStatsCountsOut, StaffStatsDetailOut +from app.db.models import ChatGroup, Customer, Message, PhoneCall, WechatAccount + +router = APIRouter(prefix="/api", tags=["staff-stats"]) + +SessionDep = Annotated[AsyncSession, Depends(get_session)] + +_MS_PER_DAY = 86_400_000 + +KindParam = Literal[ + "new_customers", + "new_groups", + "red_transfer", + "del_customers", + "blocked_customers", + "voice_calls", +] + +# 红包/转账金额提取(spec §4.5 red_transfer): +# 1) 红包 XML:``￥88.00``(feedesc 内货币符号非数字) +# 2) 云客转账纯文本:``发送转账：￥100.00`` +# 3) 转账 XML ``100.00`` 兜底 +_AMOUNT_FEEDESC_RE = re.compile(r"[^0-9]*([\d.]+)") +_AMOUNT_PLAIN_RE = re.compile(r"转账[：:]?\s*[￥¥]\s*([\d.]+)") +_AMOUNT_TOTAL_RE = re.compile(r"\s*([\d.]+)") + + +def _extract_amount(text: str | None) -> float | None: + """从消息 text 提取金额(元);双格式,失败 None。""" + if not text: + return None + m = ( + _AMOUNT_FEEDESC_RE.search(text) + or _AMOUNT_PLAIN_RE.search(text) + or _AMOUNT_TOTAL_RE.search(text) + ) + if not m: + return None + try: + return float(m.group(1)) + except ValueError: # pragma: no cover — 正则已限定 [\d.]+,防御畸形串 + return None + + +def _dt_ms(value: Any) -> int | None: + """DateTime 列 → ms epoch(naive,按本地时区解释;仅作展示兜底)。""" + if value is None: + return None + return int(value.timestamp() * 1000) + + +async def _new_customers( + session: AsyncSession, wxid: str, start: int, end: int, detail: bool +) -> tuple[int, list[dict[str, Any]]]: + """新客户:add_time 窗内;add_time IS NULL 用首条消息时间兜底。""" + base = [Customer.channel == "wechat", Customer.owner_account_id == wxid] + # 兜底口径:每个 add_time 为空的客户,取其与该员工首条消息 min(timestamp) + # (LEFT JOIN 走 ix_message_channel_owner_talker_time,HAVING 在 SQL 侧过滤窗)。 + fallback = ( + sa.select( + Customer.friend_id, + Customer.friend_nickname, + Customer.friend_remark, + Customer.head_image_url, + sa.func.min(Message.timestamp).label("first_ts"), + ) + .outerjoin( + Message, + sa.and_( + Message.channel == "wechat", + Message.owner_account_id == Customer.owner_account_id, + Message.talker_id == Customer.friend_id, + Message.timestamp.is_not(None), + ), + ) + .where(*base, Customer.add_time.is_(None)) + .group_by(Customer.friend_id) + .having(sa.func.min(Message.timestamp) >= start) + .having(sa.func.min(Message.timestamp) <= end) + ) + # 直接口径:add_time 本身在窗内。 + direct = sa.select( + Customer.friend_id, + Customer.friend_nickname, + Customer.friend_remark, + Customer.head_image_url, + Customer.add_time, + ).where(*base, Customer.add_time >= start, Customer.add_time <= end) + + rows: list[dict[str, Any]] = [ + { + "talker_id": fid, + "nickname": nick, + "avatar": head or None, + "remark": remark, + "time": add_time, + } + for fid, nick, remark, head, add_time in ( + await session.execute(direct) + ).all() + ] + rows += [ + { + "talker_id": fid, + "nickname": nick, + "avatar": head or None, + "remark": remark, + "time": first_ts, + } + for fid, nick, remark, head, first_ts in ( + await session.execute(fallback) + ).all() + ] + rows.sort(key=lambda r: r["time"] or 0, reverse=True) + return len(rows), rows + + +async def _new_groups( + session: AsyncSession, wxid: str, start: int, end: int, detail: bool +) -> tuple[int, list[dict[str, Any]]]: + """新群:chat_group.create_time 窗内(每员工一行,owner 过滤)。""" + conds = [ + ChatGroup.channel == "wechat", + ChatGroup.owner_account_id == wxid, + ChatGroup.create_time >= start, + ChatGroup.create_time <= end, + ] + count = ( + await session.execute( + sa.select(sa.func.count()).select_from(ChatGroup).where(*conds) + ) + ).scalar_one() + if not detail: + return count, [] + rows = ( + await session.execute( + sa.select(ChatGroup.room_id, ChatGroup.room_name, ChatGroup.create_time) + .where(*conds) + .order_by(ChatGroup.create_time.desc(), ChatGroup.room_id.asc()) + ) + ).all() + return count, [ + {"room_id": rid, "room_name": name, "time": ts} for rid, name, ts in rows + ] + + +async def _red_transfer( + session: AsyncSession, wxid: str, start: int, end: int, detail: bool +) -> tuple[int, list[dict[str, Any]]]: + """红包+转账:wechat 消息 type IN (red_packet, transfer) 且窗内。""" + conds = [ + Message.channel == "wechat", + Message.owner_account_id == wxid, + Message.type.in_(["red_packet", "transfer"]), + Message.timestamp >= start, + Message.timestamp <= end, + ] + count = ( + await session.execute( + sa.select(sa.func.count()).select_from(Message).where(*conds) + ) + ).scalar_one() + if not detail: + return count, [] + msgs = ( + await session.execute( + sa.select( + Message.msg_id, + Message.talker_id, + Message.room_id, + Message.type, + Message.text, + Message.timestamp, + Message.sender_name, + Message.raw_json, + ) + .where(*conds) + .order_by(Message.timestamp.desc(), Message.msg_id.asc()) + ) + ).all() + # nickname 优先 customer 昵称,回退 message raw_json.name(sender_name 即其 + # 结构化列);群红包 talker 是发红包的成员,同口径查 customer。 + talker_ids = [m.talker_id for m in msgs if m.talker_id] + nick_map: dict[str, str] = {} + if talker_ids: + nick_map = { + fid: nick + for fid, nick in ( + await session.execute( + sa.select(Customer.friend_id, Customer.friend_nickname).where( + Customer.channel == "wechat", + Customer.owner_account_id == wxid, + Customer.friend_id.in_(talker_ids), + ) + ) + ).all() + if nick + } + rows: list[dict[str, Any]] = [] + for m in msgs: + raw_name = (m.raw_json or {}).get("name") if m.raw_json else None + rows.append( + { + "op_type": "红包" if m.type == "red_packet" else "转账", + "talker_id": m.talker_id, + "room_id": m.room_id or None, + "nickname": nick_map.get(m.talker_id or "") or m.sender_name or raw_name, + "amount": _extract_amount(m.text), + "time": m.timestamp, + "msg_id": m.msg_id, + } + ) + return count, rows + + +async def _flag_customers( + session: AsyncSession, + wxid: str, + flag_cond: sa.ColumnElement, + *, + detail: bool, +) -> tuple[int, list[dict[str, Any]]]: + """del_customers / blocked_customers 共用:现存量口径,忽略时间窗。 + + 无时间字段→明细行 time 用 updated_at 毫秒(仅展示;update 触发即变)。 + """ + conds = [Customer.channel == "wechat", Customer.owner_account_id == wxid, flag_cond] + count = ( + await session.execute( + sa.select(sa.func.count()).select_from(Customer).where(*conds) + ) + ).scalar_one() + if not detail: + return count, [] + rows = ( + await session.execute( + sa.select( + Customer.friend_id, + Customer.friend_nickname, + Customer.friend_remark, + Customer.updated_at, + ) + .where(*conds) + .order_by(Customer.updated_at.desc(), Customer.friend_id.asc()) + ) + ).all() + return count, [ + { + "talker_id": fid, + "nickname": nick or remark, + "time": _dt_ms(updated), + } + for fid, nick, remark, updated in rows + ] + + +async def _del_customers( + session: AsyncSession, wxid: str, start: int, end: int, detail: bool +) -> tuple[int, list[dict[str, Any]]]: + """删除客户:delete_flag=1 现存量(start/end 按口径忽略)。""" + return await _flag_customers( + session, wxid, Customer.delete_flag == 1, detail=detail + ) + + +async def _blocked_customers( + session: AsyncSession, wxid: str, start: int, end: int, detail: bool +) -> tuple[int, list[dict[str, Any]]]: + """流失(拉黑)客户:is_block=1 现存量(start/end 按口径忽略)。""" + return await _flag_customers( + session, wxid, Customer.is_block.is_(True), detail=detail + ) + + +async def _voice_calls( + session: AsyncSession, wxid: str, start: int, end: int, detail: bool +) -> tuple[int, list[dict[str, Any]]]: + """语音通话:先 wechat_account.user_id 换算(防直查 wxId 恒 0)再查。 + + 明细字段语义对齐 CallList.vue / PhoneCallOut: + ``connected`` 按 voice_call 极性 call_status==1(与前端 callStatusLabel + 同判据:1=已接通,0/2=未接通);``media_url`` 本地 /api/media/{hash} + 优先、OSS record_file 兜底(与 playRecord 同顺序)。账号无 user_id + (未同步/未知员工)→ 无法归属,返回空而非全量(防跨员工串数)。 + """ + uid = ( + await session.execute( + sa.select(WechatAccount.user_id).where(WechatAccount.wx_id == wxid) + ) + ).scalar_one_or_none() + if not uid: + return 0, [] + conds = [ + PhoneCall.channel == "voice_call", + PhoneCall.user_id == uid, + PhoneCall.start_call_time >= start, + PhoneCall.start_call_time <= end, + ] + count = ( + await session.execute( + sa.select(sa.func.count()).select_from(PhoneCall).where(*conds) + ) + ).scalar_one() + if not detail: + return count, [] + calls = ( + await session.execute( + sa.select(PhoneCall) + .where(*conds) + .order_by(PhoneCall.start_call_time.desc(), PhoneCall.call_id.asc()) + ) + ).scalars().all() + rows = [ + { + "id": c.call_id, + "peer": c.call_number, + "incoming": c.incoming_call, + "connected": c.call_status == 1, + "duration": c.call_duration, + "start_time": c.start_call_time, + "media_url": ( + f"/api/media/{c.record_media_hash}" if c.record_media_hash else c.record_file + ), + } + for c in calls + ] + return count, rows + + +_AGGREGATORS = { + "new_customers": _new_customers, + "new_groups": _new_groups, + "red_transfer": _red_transfer, + "del_customers": _del_customers, + "blocked_customers": _blocked_customers, + "voice_calls": _voice_calls, +} + + +@router.get("/staff-stats") +async def get_staff_stats( + session: SessionDep, + wechat_id: Annotated[str, Query(min_length=1)], + start: Annotated[int | None, Query(ge=0)] = None, + end: Annotated[int | None, Query(ge=0)] = None, + kind: Annotated[KindParam | None, Query()] = None, +) -> StaffStatsCountsOut | StaffStatsDetailOut: + """员工 7 类计数(无 kind)或某一类的明细列表(?kind=)。 + + 窗口 [start, end] 均为 ms epoch 闭区间;缺省 = 近 7 天(云客统计默认窗)。 + del_customers / blocked_customers 是现存量口径,窗口仅约束其余 4 类。 + """ + if end is None: + end = int(time.time() * 1000) + if start is None: + start = end - 7 * _MS_PER_DAY + + if kind is None: + counts: dict[str, int] = {} + for kind_key, aggregate in _AGGREGATORS.items(): + counts[kind_key], _ = await aggregate(session, wechat_id, start, end, detail=False) + counts["sensitive_words"] = 0 # 占位,spec 决策 + return StaffStatsCountsOut(**counts) + + _, rows = await _AGGREGATORS[kind](session, wechat_id, start, end, detail=True) + return StaffStatsDetailOut(kind=kind, rows=rows) diff --git a/yunke-server/app/api/schemas.py b/yunke-server/app/api/schemas.py index 537eace..e4cc39e 100644 --- a/yunke-server/app/api/schemas.py +++ b/yunke-server/app/api/schemas.py @@ -403,6 +403,39 @@ class StaffChatStatsOut(BaseModel): customers: list[StaffChatCustomerOut] +class StaffStatsCountsOut(BaseModel): + """GET /api/staff-stats 计数模式(无 kind)——云客员工聊天窗口 7 类汇总数字。 + + 口径见 :mod:`app.api.routers.staff_stats`(spec §4.5):new_customers 含 + 「首条消息时间」兜底;del/blocked 是现存量(不受窗口约束);voice_calls + 走 wechat_account.user_id 换算。``sensitive_words`` 为敏感词占位,恒 0。 + """ + + new_customers: int + new_groups: int + red_transfer: int + del_customers: int + blocked_customers: int + voice_calls: int + sensitive_words: int = 0 + + +class StaffStatsDetailOut(BaseModel): + """GET /api/staff-stats 明细模式(``?kind=``)——rows 字段随 kind 不同。 + + 统一 ``list[dict]``,前端按 kind 配置列(字段清单见 router docstring): + new_customers {talker_id, nickname, avatar, remark, time} / + new_groups {room_id, room_name, time} / + red_transfer {op_type, talker_id, room_id, nickname, amount, time, msg_id} / + del|blocked_customers {talker_id, nickname, time} / + voice_calls {id, peer, incoming, connected, duration, start_time, media_url} + (语义对齐 CallList.vue / PhoneCallOut)。 + """ + + kind: str + rows: list[dict[str, Any]] + + class SessionOut(BaseModel): """账号会话列表的一行(好友或群)。""" diff --git a/yunke-server/tests/test_api_staff_stats.py b/yunke-server/tests/test_api_staff_stats.py new file mode 100644 index 0000000..556c249 --- /dev/null +++ b/yunke-server/tests/test_api_staff_stats.py @@ -0,0 +1,315 @@ +"""Tests for GET /api/staff-stats — 员工 7 类计数 + 分型明细(本地聚合,spec §4.5)。""" +from __future__ import annotations + +import time + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import ChatGroup, Customer, Message, PhoneCall, WechatAccount + +OWNER = "wxid_o" +UID = "U32HEX" # 员工 userId(32位HEX)——phone_call.user_id 存的是它,不是 wxId! +WIN_START = 1_700_000_000_000 +WIN_END = 1_900_000_000_000 +IN = 1_800_000_000_000 # 窗内锚点 +DAY = 86_400_000 + + +# --------------------------------------------------------------------------- +# 计数模式(无 kind) +# --------------------------------------------------------------------------- + + +async def _seed_counts(s: AsyncSession) -> None: + s.add(WechatAccount(wx_id=OWNER, user_id=UID)) + # 新客户: add_time 窗内 1 + 兜底首消息窗内 1 + 窗外 1(不计) + s.add(Customer(friend_id="c1", channel="wechat", owner_account_id=OWNER, add_time=IN)) + s.add(Customer(friend_id="c2", channel="wechat", owner_account_id=OWNER, add_time=None)) + s.add(Customer(friend_id="c3", channel="wechat", owner_account_id=OWNER, add_time=1_000)) + s.add(Message(msg_id="m1", channel="wechat", owner_account_id=OWNER, talker_id="c2", + timestamp=IN + 1)) + # 红包+转账(窗内各 1) + s.add(Message(msg_id="m2", channel="wechat", owner_account_id=OWNER, talker_id="c1", + timestamp=IN + 2, type="red_packet")) + s.add(Message(msg_id="m3", channel="wechat", owner_account_id=OWNER, talker_id="c1", + timestamp=IN + 3, type="transfer")) + # 新群: 窗内 1 + 窗外 1(不计) + s.add(ChatGroup(room_id="r1@chatroom", channel="wechat", owner_account_id=OWNER, create_time=IN)) + s.add(ChatGroup(room_id="r2@chatroom", channel="wechat", owner_account_id=OWNER, create_time=1_000)) + # 删除/拉黑(现存量口径,忽略窗口) + s.add(Customer(friend_id="c9", channel="wechat", owner_account_id=OWNER, delete_flag=1)) + s.add(Customer(friend_id="c8", channel="wechat", owner_account_id=OWNER, is_block=True)) + # 语音通话: user_id 是 32HEX 非 wxId!另 seed 一条直存 wxId 的(不计,锁死 join 口径) + s.add(PhoneCall(call_id="pc1", channel="voice_call", user_id=UID, + start_call_time=IN + 4, call_number="c1")) + s.add(PhoneCall(call_id="pc_x", channel="voice_call", user_id=OWNER, + start_call_time=IN + 5, call_number="c1")) + await s.commit() + + +async def test_counts_all_seven_kinds(make_client): + """7 类计数一次返回;sensitive_words 恒 0;voice_calls 走 user_id 换算。""" + client = await make_client(_seed_counts) + r = await client.get( + "/api/staff-stats", + params={"wechat_id": OWNER, "start": WIN_START, "end": WIN_END}, + ) + assert r.status_code == 200 + body = r.json() + assert set(body) == { + "new_customers", "new_groups", "red_transfer", + "del_customers", "blocked_customers", "voice_calls", "sensitive_words", + } + assert body["new_customers"] == 2 # c1(add_time) + c2(首消息兜底); c3 窗外 + assert body["new_groups"] == 1 + assert body["red_transfer"] == 2 + assert body["del_customers"] == 1 # 现存量 + assert body["blocked_customers"] == 1 # 现存量 + assert body["voice_calls"] == 1 # 直查 wxid 会得 0 —— join user_id + assert body["sensitive_words"] == 0 # 占位,spec 决策 + + +async def test_counts_default_window_last_7_days(make_client): + """缺省 start/end → 近 7 天:窗内计、8 天前不计。""" + + async def seed(s: AsyncSession) -> None: + now = int(time.time() * 1000) + s.add(WechatAccount(wx_id=OWNER, user_id=UID)) + s.add(Customer(friend_id="c_now", channel="wechat", owner_account_id=OWNER, add_time=now)) + s.add(Customer(friend_id="c_old", channel="wechat", owner_account_id=OWNER, + add_time=now - 8 * DAY)) + s.add(PhoneCall(call_id="pc_now", channel="voice_call", user_id=UID, + start_call_time=now, call_number="c_now")) + s.add(PhoneCall(call_id="pc_old", channel="voice_call", user_id=UID, + start_call_time=now - 8 * DAY)) + await s.commit() + + client = await make_client(seed) + r = await client.get("/api/staff-stats", params={"wechat_id": OWNER}) + assert r.status_code == 200 + body = r.json() + assert body["new_customers"] == 1 # 只计窗内 + assert body["voice_calls"] == 1 + + +async def test_counts_unknown_staff_all_zero(make_client): + """未知员工:全 0(不 404,前端弹窗兜底)。""" + + async def seed(s: AsyncSession) -> None: + await s.commit() + + client = await make_client(seed) + r = await client.get( + "/api/staff-stats", + params={"wechat_id": "wxid_ghost", "start": WIN_START, "end": WIN_END}, + ) + assert r.status_code == 200 + assert r.json() == { + "new_customers": 0, "new_groups": 0, "red_transfer": 0, + "del_customers": 0, "blocked_customers": 0, "voice_calls": 0, + "sensitive_words": 0, + } + + +# --------------------------------------------------------------------------- +# 明细模式(?kind=) — 每 kind 字段各一条 +# --------------------------------------------------------------------------- + + +async def test_detail_new_customers_fields(make_client): + """new_customers 行 {talker_id, nickname, avatar, remark, time};兜底行 time=首消息 ms。""" + + async def seed(s: AsyncSession) -> None: + s.add(WechatAccount(wx_id=OWNER, user_id=UID)) + s.add(Customer(friend_id="c1", channel="wechat", owner_account_id=OWNER, add_time=IN, + friend_nickname="牛牛", friend_remark="大客户", + head_image_url="http://a/1.png")) + s.add(Customer(friend_id="c2", channel="wechat", owner_account_id=OWNER, add_time=None, + friend_nickname="兜底客")) + s.add(Customer(friend_id="c3", channel="wechat", owner_account_id=OWNER, add_time=1_000)) + s.add(Message(msg_id="m1", channel="wechat", owner_account_id=OWNER, talker_id="c2", + timestamp=IN - 5)) + await s.commit() + + client = await make_client(seed) + r = await client.get( + "/api/staff-stats", + params={"wechat_id": OWNER, "kind": "new_customers", + "start": WIN_START, "end": WIN_END}, + ) + assert r.status_code == 200 + body = r.json() + assert body["kind"] == "new_customers" + by = {row["talker_id"]: row for row in body["rows"]} + assert set(by) == {"c1", "c2"} # c3 窗外 + assert by["c1"]["nickname"] == "牛牛" + assert by["c1"]["remark"] == "大客户" + assert by["c1"]["avatar"] == "http://a/1.png" + assert by["c1"]["time"] == IN + assert by["c2"]["time"] == IN - 5 # add_time NULL → 首条消息兜底 + # 排序:time desc(新→旧) + assert [row["talker_id"] for row in body["rows"]] == ["c1", "c2"] + + +async def test_detail_new_groups_fields(make_client): + """new_groups 行 {room_id, room_name, time};只计本 owner 窗内。""" + + async def seed(s: AsyncSession) -> None: + s.add(WechatAccount(wx_id=OWNER, user_id=UID)) + s.add(ChatGroup(room_id="r1@chatroom", channel="wechat", owner_account_id=OWNER, + room_name="老客户群", create_time=IN)) + s.add(ChatGroup(room_id="r2@chatroom", channel="wechat", owner_account_id=OWNER, + room_name="窗外群", create_time=1_000)) + s.add(ChatGroup(room_id="r3@chatroom", channel="wechat", owner_account_id="wxid_other", + room_name="别人的群", create_time=IN)) + s.add(ChatGroup(room_id="r4@chatroom", channel="wechat", owner_account_id=OWNER, + room_name="无建群时间", create_time=None)) + await s.commit() + + client = await make_client(seed) + r = await client.get( + "/api/staff-stats", + params={"wechat_id": OWNER, "kind": "new_groups", + "start": WIN_START, "end": WIN_END}, + ) + assert r.status_code == 200 + rows = r.json()["rows"] + assert rows == [{"room_id": "r1@chatroom", "room_name": "老客户群", "time": IN}] + + +async def test_detail_red_transfer_fields(make_client): + """red_transfer 行 {op_type, talker_id, room_id, nickname, amount, time, msg_id}; + 金额双格式提取(红包 XML feedesc / 转账纯文本),失败 None;nickname 客户优先。""" + + async def seed(s: AsyncSession) -> None: + s.add(WechatAccount(wx_id=OWNER, user_id=UID)) + s.add(Customer(friend_id="c1", channel="wechat", owner_account_id=OWNER, + friend_nickname="客户昵称")) + # 红包: XML ￥88.00 + s.add(Message(msg_id="m2", channel="wechat", owner_account_id=OWNER, talker_id="c1", + timestamp=IN, type="red_packet", + text="2001" + "￥88.00" + "")) + # 转账: 云客纯文本「发送转账：￥100.00」;c2 无 customer 行 → raw_json.name 兜底 + s.add(Message(msg_id="m3", channel="wechat", owner_account_id=OWNER, talker_id="c2", + timestamp=IN + 1, type="transfer", text="发送转账：￥100.00", + raw_json={"name": "裸数据客"})) + # 转账无金额可提 → amount None + s.add(Message(msg_id="m4", channel="wechat", owner_account_id=OWNER, talker_id="c2", + timestamp=IN + 2, type="transfer", text=None)) + # 普通消息不计 + s.add(Message(msg_id="m_text", channel="wechat", owner_account_id=OWNER, talker_id="c1", + timestamp=IN + 3, type="text", text="你好")) + await s.commit() + + client = await make_client(seed) + r = await client.get( + "/api/staff-stats", + params={"wechat_id": OWNER, "kind": "red_transfer", + "start": WIN_START, "end": WIN_END}, + ) + assert r.status_code == 200 + rows = r.json()["rows"] + assert [row["msg_id"] for row in rows] == ["m4", "m3", "m2"] # time desc + by = {row["msg_id"]: row for row in rows} + assert by["m2"]["op_type"] == "红包" + assert by["m2"]["amount"] == 88.0 + assert by["m2"]["talker_id"] == "c1" + assert by["m2"]["room_id"] is None + assert by["m2"]["nickname"] == "客户昵称" + assert by["m2"]["time"] == IN + assert by["m3"]["op_type"] == "转账" + assert by["m3"]["amount"] == 100.0 + assert by["m3"]["nickname"] == "裸数据客" # customer 无行 → raw_json.name + assert by["m4"]["amount"] is None + + +async def test_detail_del_and_blocked_fields(make_client): + """del/blocked 行 {talker_id, nickname, time};time=updated_at 毫秒;只计本 owner。""" + + async def seed(s: AsyncSession) -> None: + s.add(WechatAccount(wx_id=OWNER, user_id=UID)) + s.add(Customer(friend_id="c_del", channel="wechat", owner_account_id=OWNER, + delete_flag=1, friend_nickname="已删客")) + s.add(Customer(friend_id="c_blk", channel="wechat", owner_account_id=OWNER, + is_block=True, friend_nickname=None, friend_remark="被拉黑")) + # 干扰项: 正常客户 / 别人的删除客户 / delete_flag=0 + s.add(Customer(friend_id="c_ok", channel="wechat", owner_account_id=OWNER)) + s.add(Customer(friend_id="c_other", channel="wechat", owner_account_id="wxid_other", + delete_flag=1)) + s.add(Customer(friend_id="c_fine", channel="wechat", owner_account_id=OWNER, + delete_flag=0, is_block=False)) + await s.commit() + + client = await make_client(seed) + r_del = await client.get("/api/staff-stats", params={"wechat_id": OWNER, "kind": "del_customers"}) + r_blk = await client.get("/api/staff-stats", + params={"wechat_id": OWNER, "kind": "blocked_customers"}) + assert r_del.status_code == r_blk.status_code == 200 + rows_del = r_del.json()["rows"] + assert len(rows_del) == 1 + d = rows_del[0] + assert d["talker_id"] == "c_del" and d["nickname"] == "已删客" + assert isinstance(d["time"], int) and d["time"] > WIN_START # updated_at ms + rows_blk = r_blk.json()["rows"] + assert len(rows_blk) == 1 + b = rows_blk[0] + assert b["talker_id"] == "c_blk" + assert b["nickname"] == "被拉黑" # nickname 空时回退备注 + assert isinstance(b["time"], int) and b["time"] > WIN_START + + +async def test_detail_voice_calls_fields(make_client): + """voice_calls 行 {id, peer, incoming, connected, duration, start_time, media_url}; + 语义对齐 CallList.vue:connected=voice_call 口径 call_status==1, + media_url 本地 /api/media/{hash} 优先、OSS record_file 兜底。""" + + async def seed(s: AsyncSession) -> None: + s.add(WechatAccount(wx_id=OWNER, user_id=UID)) + s.add(PhoneCall(call_id="pc1", channel="voice_call", user_id=UID, + start_call_time=IN, call_number="c1", incoming_call=1, + call_status=1, call_duration=62, record_media_hash="abc123")) + s.add(PhoneCall(call_id="pc2", channel="voice_call", user_id=UID, + start_call_time=IN + 1, call_number="room@chatroom", + incoming_call=0, call_status=2, call_duration=0, + record_file="http://oss/x.mp3")) + # 干扰项: 他人语音 / 电话通话(同 uid 也不计) + s.add(PhoneCall(call_id="pc3", channel="voice_call", user_id="other_uid", + start_call_time=IN)) + s.add(PhoneCall(call_id="pc4", channel="phone_call", user_id=UID, + start_call_time=IN)) + await s.commit() + + client = await make_client(seed) + r = await client.get( + "/api/staff-stats", + params={"wechat_id": OWNER, "kind": "voice_calls", + "start": WIN_START, "end": WIN_END}, + ) + assert r.status_code == 200 + rows = r.json()["rows"] + assert [row["id"] for row in rows] == ["pc2", "pc1"] # start_time desc + by = {row["id"]: row for row in rows} + assert by["pc1"]["peer"] == "c1" + assert by["pc1"]["incoming"] == 1 + assert by["pc1"]["connected"] is True # voice_call: call_status==1 接通 + assert by["pc1"]["duration"] == 62 + assert by["pc1"]["start_time"] == IN + assert by["pc1"]["media_url"] == "/api/media/abc123" # 本地优先 + assert by["pc2"]["peer"] == "room@chatroom" + assert by["pc2"]["incoming"] == 0 + assert by["pc2"]["connected"] is False # voice_call: call_status==2 未接通 + assert by["pc2"]["media_url"] == "http://oss/x.mp3" # OSS 兜底 + + +async def test_detail_unknown_kind_422(make_client): + """非法 kind → 422(Literal 校验)。""" + + async def seed(s: AsyncSession) -> None: + await s.commit() + + client = await make_client(seed) + r = await client.get("/api/staff-stats", + params={"wechat_id": OWNER, "kind": "bogus"}) + assert r.status_code == 422