xinyouhe/yunke
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
master
yunke/yunke-server/tests/test_api_staff_stats.py
T
浅念 99773f69b9 fix(staff-stats): 统计tab新客户/删除客户/红包昵称切 link——共享好友不再漏计 
全项目同型审计结论的收尾项:统计分析 tab 按 owner 查全局单行档案表,
共享好友(档案挂别的员工名下)不计入——

- _new_customers(新客户):主体 Customer → CustomerAccountLink(add_time
  窗+首条消息兜底 JOIN 平移到 link.owner);数字变大=对齐云客按员工口径
- _del_customers(删除客户):_flag_customers 参数化 model,del 走 link
  (delete_flag per-relation);blocked 保留档案口径(is_block 未镜像进
  link,已知偏差=他人员工名下拉黑不计,注释标明)
- _red_transfer 昵称:link 优先档案兜底(同 sessions._meta 模式)

审计同时确认无需修:群侧(chat_group 复合主键天然多对多)、搜索/长语音
(按 friend 全局直查)、analytics(全局口径)、WA(与微信共用同批修复,
12,578 行 link 已回填,1,967 共享客户在册;无昵称行=WA 联系人本身没存
名字,与云客一致显示号码)。

644 测试绿(新增共享好友新客户判别测试;存量测试种子补 link 行镜像
生产写路径)。

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-09 13:45:42 +08:00

450 lines
21 KiB
Python
Raw Permalink Blame History

"""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,
CustomerAccountLink,
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))
# link 行(生产抓取经 upsert_customer 成对写;新客户/删除统计读 link)
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c1", add_time=IN))
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c2"))
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c3", add_time=1_000))
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c9", delete_flag=1))
# 语音通话: 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(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c_now", add_time=now))
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c_old",
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(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c1", add_time=IN,
friend_nickname="牛牛", friend_remark="大客户",
head_image_url="http://a/1.png"))
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c2",
friend_nickname="兜底客"))
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c3", 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 <feedesc>￥88.00</feedesc>
s.add(Message(msg_id="m2", channel="wechat", owner_account_id=OWNER, talker_id="c1",
timestamp=IN, type="red_packet",
text="<msg><appmsg><type>2001</type>"
"<wcpayinfo><feedesc>￥88.00</feedesc></wcpayinfo>"
"</appmsg></msg>"))
# 转账: 云客纯文本「发送转账：￥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))
# 删除读 link:c_del 补 link 行;is_block 保持档案(c_blk 不补)
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER, friend_id="c_del",
delete_flag=1, friend_nickname="已删客"))
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
# ---------------------------------------------------------------------------
# 边缘口径
# ---------------------------------------------------------------------------
async def test_amount_thousands_separator(make_client):
"""金额千分位:``￥1,000.50`` → 1000.5(不能截成 1);feedesc 同口径。"""
async def seed(s: AsyncSession) -> None:
s.add(WechatAccount(wx_id=OWNER, user_id=UID))
s.add(Message(msg_id="m_plain", channel="wechat", owner_account_id=OWNER,
talker_id="c1", timestamp=IN, type="transfer",
text="发送转账：￥1,000.50"))
s.add(Message(msg_id="m_xml", channel="wechat", owner_account_id=OWNER,
talker_id="c1", timestamp=IN + 1, type="red_packet",
text="<wcpayinfo><feedesc>￥2,000.00</feedesc></wcpayinfo>"))
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
by = {row["msg_id"]: row["amount"] for row in r.json()["rows"]}
assert by["m_plain"] == 1000.5
assert by["m_xml"] == 2000.0
async def test_voice_calls_account_without_user_id(make_client):
"""账号存在但 user_id 为 NULL(未同步):voice_calls 计 0、明细空,
而非把 user_id IS NULL 的通话误归属(防跨员工串数)。"""
async def seed(s: AsyncSession) -> None:
s.add(WechatAccount(wx_id=OWNER, user_id=None))
s.add(PhoneCall(call_id="pc_null", channel="voice_call", user_id=None,
start_call_time=IN, call_number="c1"))
await s.commit()
client = await make_client(seed)
r_count = await client.get(
"/api/staff-stats",
params={"wechat_id": OWNER, "start": WIN_START, "end": WIN_END},
)
r_detail = await client.get(
"/api/staff-stats",
params={"wechat_id": OWNER, "kind": "voice_calls",
"start": WIN_START, "end": WIN_END},
)
assert r_count.status_code == r_detail.status_code == 200
assert r_count.json()["voice_calls"] == 0
assert r_detail.json()["rows"] == []
async def test_new_customers_fallback_first_msg_before_window(make_client):
"""add_time NULL 兜底排除项:首条消息在窗前 → 不计(窗后/窗内才计)。"""
async def seed(s: AsyncSession) -> None:
s.add(WechatAccount(wx_id=OWNER, user_id=UID))
s.add(Customer(friend_id="c_early", channel="wechat", owner_account_id=OWNER,
add_time=None))
s.add(Message(msg_id="m_early", channel="wechat", owner_account_id=OWNER,
talker_id="c_early", timestamp=WIN_START - 1))
await s.commit()
client = await make_client(seed)
r_count = await client.get(
"/api/staff-stats",
params={"wechat_id": OWNER, "start": WIN_START, "end": WIN_END},
)
r_detail = await client.get(
"/api/staff-stats",
params={"wechat_id": OWNER, "kind": "new_customers",
"start": WIN_START, "end": WIN_END},
)
assert r_count.status_code == r_detail.status_code == 200
assert r_count.json()["new_customers"] == 0
assert r_detail.json()["rows"] == []
async def test_new_customers_counts_shared_friend_via_link(make_client):
"""共享好友(档案挂别的员工名下)凭 link 行计入新客户(2026-09-09 口径修复)。
旧档案口径下 c_shared 对 OWNER 不可见(add_time 在窗内也不计);切 link
后按 per-owner 关系计数,与云客按员工统计一致。
"""
async def seed(s: AsyncSession) -> None:
s.add(WechatAccount(wx_id=OWNER, user_id=UID))
# 档案行(last-wins)挂在别的员工名下
s.add(Customer(friend_id="c_shared", channel="wechat",
owner_account_id="wxid_other", add_time=IN))
# OWNER 名下的 link 关系
s.add(CustomerAccountLink(channel="wechat", owner_account_id=OWNER,
friend_id="c_shared", add_time=IN))
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 [row["talker_id"] for row in body["rows"]] == ["c_shared"]
assert len(body["rows"]) == 1
Reference in New Issue View Git Blame Copy Permalink