diff --git a/yunke-server/app/api/routers/staff_stats.py b/yunke-server/app/api/routers/staff_stats.py index afd2dc8..6e4ca6a 100644 --- a/yunke-server/app/api/routers/staff_stats.py +++ b/yunke-server/app/api/routers/staff_stats.py @@ -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 ) diff --git a/yunke-server/tests/test_api_staff_stats.py b/yunke-server/tests/test_api_staff_stats.py index 32d4207..8ab01e1 100644 --- a/yunke-server/tests/test_api_staff_stats.py +++ b/yunke-server/tests/test_api_staff_stats.py @@ -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