xinyouhe/yunke
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(staff-stats): 统计tab新客户/删除客户/红包昵称切 link——共享好友不再漏计

Browse Source 
全项目同型审计结论的收尾项:统计分析 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>
This commit is contained in:
浅念
2026-09-09 13:45:42 +08:00
parent 9dc22cd1e6
commit 99773f69b9
2 changed files with 142 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yunke-server/app/api/routers/staff_stats.py
+88 -37
View File
@@ -40,7 +40,14 @@ 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
from app.db.models import (
ChatGroup,
Customer,
CustomerAccountLink,
Message,
PhoneCall,
WechatAccount,
)
router = APIRouter(prefix="/api", tags=["staff-stats"])
@@ -96,50 +103,57 @@ async def _new_customers(
) -> tuple[int, list[dict[str, Any]]]:
"""新客户:add_time 窗内;add_time IS NULL 用首条消息时间兜底。
读 ``customer_account_link``(2026-09-09):customer 档案全局单行、
归属 last-wins,共享客户在其他员工名下会漏计;link 每账号一行,与
云客按员工统计的口径一致(数字会比档案口径大)。
无 ``detail`` 分支——计数模式有意复用明细物化(窗内新增客户通常
几十行,``len()`` 即可;COUNT 需 direct+fallback 双查询,不值)。
"""
base = [Customer.channel == "wechat", Customer.owner_account_id == wxid]
base = [
CustomerAccountLink.channel == "wechat",
CustomerAccountLink.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,
CustomerAccountLink.friend_id,
CustomerAccountLink.friend_nickname,
CustomerAccountLink.friend_remark,
CustomerAccountLink.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.owner_account_id == CustomerAccountLink.owner_account_id,
Message.talker_id == CustomerAccountLink.friend_id,
Message.timestamp.is_not(None),
),
)
.where(*base, Customer.add_time.is_(None))
# GROUP BY 显式列出全部裸选列:friend_id 是 Customer 唯一 PK,其余列
# 函数依赖于它→分组结果不变;MySQL 8 FD 推断可容忍裸列,但显式列表
# 在 ONLY_FULL_GROUP_BY 下部署安全(SQLite 同样兼容)。
.where(*base, CustomerAccountLink.add_time.is_(None))
# GROUP BY 显式列出全部裸选列:friend_id 是 link 复合 PK 的尾列,
# (owner,friend) 已由 WHERE 钉死,其余列函数依赖于它→分组结果不变;
# MySQL ONLY_FULL_GROUP_BY 部署安全(SQLite 同样兼容)。
.group_by(
Customer.friend_id,
Customer.friend_nickname,
Customer.friend_remark,
Customer.head_image_url,
CustomerAccountLink.friend_id,
CustomerAccountLink.friend_nickname,
CustomerAccountLink.friend_remark,
CustomerAccountLink.head_image_url,
)
.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)
CustomerAccountLink.friend_id,
CustomerAccountLink.friend_nickname,
CustomerAccountLink.friend_remark,
CustomerAccountLink.head_image_url,
CustomerAccountLink.add_time,
).where(*base, CustomerAccountLink.add_time >= start, CustomerAccountLink.add_time <= end)
rows: list[dict[str, Any]] = [
{
@@ -237,19 +251,43 @@ async def _red_transfer(
talker_ids = [m.talker_id for m in msgs if m.talker_id]
nick_map: dict[str, str] = {}
if talker_ids:
# 昵称优先 link per-owner 快照(共享好友档案行可能挂别的员工名下,
# 2026-09-09 同 sessions._meta),档案兜底
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),
sa.select(
CustomerAccountLink.friend_id,
CustomerAccountLink.friend_nickname,
).where(
CustomerAccountLink.channel == "wechat",
CustomerAccountLink.owner_account_id == wxid,
CustomerAccountLink.friend_id.in_(talker_ids),
)
)
).all()
if nick
}
missing = [t for t in talker_ids if t not in nick_map]
if missing:
nick_map.update(
{
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_(missing),
)
)
).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
@@ -273,15 +311,20 @@ async def _flag_customers(
flag_cond: sa.ColumnElement,
*,
detail: bool,
model: type,
) -> tuple[int, list[dict[str, Any]]]:
"""del_customers / blocked_customers 共用:现存量口径,忽略时间窗。
``model`` 由调用方决定(两者列名同形):del 走 ``CustomerAccountLink``
(delete_flag per-relation,共享好友不漏),blocked 走 ``Customer``
档案(is_block 未镜像进 link,2026-09-09)。
无时间字段→明细行 time 用 updated_at 毫秒(仅展示;update 触发即变)。
"""
conds = [Customer.channel == "wechat", Customer.owner_account_id == wxid, flag_cond]
conds = [model.channel == "wechat", model.owner_account_id == wxid, flag_cond]
count = (
await session.execute(
sa.select(sa.func.count()).select_from(Customer).where(*conds)
sa.select(sa.func.count()).select_from(model).where(*conds)
)
).scalar_one()
if not detail:
@@ -289,13 +332,13 @@ async def _flag_customers(
rows = (
await session.execute(
sa.select(
Customer.friend_id,
Customer.friend_nickname,
Customer.friend_remark,
Customer.updated_at,
model.friend_id,
model.friend_nickname,
model.friend_remark,
model.updated_at,
)
.where(*conds)
.order_by(Customer.updated_at.desc(), Customer.friend_id.asc())
.order_by(model.updated_at.desc(), model.friend_id.asc())
)
).all()
return count, [
@@ -311,18 +354,26 @@ async def _flag_customers(
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 按口径忽略)。"""
"""删除客户:delete_flag=1 现存量(start/end 按口径忽略)。
读 link(per-relation delete_flag,共享好友不漏计,2026-09-09)。
"""
return await _flag_customers(
session, wxid, Customer.delete_flag == 1, detail=detail
session, wxid, CustomerAccountLink.delete_flag == 1,
detail=detail, model=CustomerAccountLink,
)
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 按口径忽略)。"""
"""流失(拉黑)客户:is_block=1 现存量(start/end 按口径忽略)。
保持档案口径:is_block 未镜像进 link 表,切了会漏;已知偏差=共享好友
在其他员工名下的拉黑不计入(待 link 补列后切换)。
"""
return await _flag_customers(
session, wxid, Customer.is_block.is_(True), detail=detail
session, wxid, Customer.is_block.is_(True), detail=detail, model=Customer
)
yunke-server/tests/test_api_staff_stats.py
+54 -1
View File
@@ -5,7 +5,14 @@ import time
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import ChatGroup, Customer, Message, PhoneCall, WechatAccount
from app.db.models import (
ChatGroup,
Customer,
CustomerAccountLink,
Message,
PhoneCall,
WechatAccount,
)
OWNER = "wxid_o"
UID = "U32HEX" # 员工 userId(32位HEX)——phone_call.user_id 存的是它,不是 wxId!
@@ -39,6 +46,11 @@ async def _seed_counts(s: AsyncSession) -> None:
# 删除/拉黑(现存量口径,忽略窗口)
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"))
@@ -78,6 +90,9 @@ async def test_counts_default_window_last_7_days(make_client):
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,
@@ -127,6 +142,12 @@ async def test_detail_new_customers_fields(make_client):
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()
@@ -238,6 +259,9 @@ async def test_detail_del_and_blocked_fields(make_client):
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()
@@ -394,3 +418,32 @@ async def test_new_customers_fallback_first_msg_before_window(make_client):
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