# --------------------------------------------------------------------------- # GET /api/customers/declared-total — 云客口径合计(页脚对齐云客显示,2026-08-20) # --------------------------------------------------------------------------- async def _seed_accounts_for_declared(session): from app.db.models import WechatAccount session.add_all([ WechatAccount(wx_id="w1", contacts_num=100), WechatAccount(wx_id="w2", contacts_num=23), WechatAccount(wx_id="w3", contacts_num=None), ]) await session.commit() async def test_declared_total_sums_contacts_num(make_client): """分渠道 declared_total = Σ 员工账号表 contacts_num(云客含重复口径)。""" from app.db.models import WhatsappAccount async def seed_both(session): await _seed_accounts_for_declared(session) session.add_all([ WhatsappAccount(uid="a1@s.whatsapp.net", contacts_num=10), WhatsappAccount(uid="a2@s.whatsapp.net", contacts_num=5), ]) await session.commit() client = await make_client(seed_both) r = await client.get("/api/customers/declared-total") assert r.status_code == 200 body = r.json() assert body["wechat"]["declared_total"] == 123 assert body["wechat"]["account_count"] == 2 # 只计有值的员工 assert body["whatsapp"]["declared_total"] == 15 assert body["whatsapp"]["account_count"] == 2 # --------------------------------------------------------------------------- # send_today / receive_today — 今日发送/接收条数(对齐云客口径,2026-08-22) # --------------------------------------------------------------------------- async def _seed_today_counts(session): import datetime as dt from app.db.models import Customer as C, Message as M now_ms = int(dt.datetime.now(dt.UTC).timestamp() * 1000) yesterday_ms = now_ms - 86_400_000 session.add_all([ C(channel="wechat", owner_account_id="o1", friend_id="f1"), C(channel="wechat", owner_account_id="o1", friend_id="f2"), ]) # f1: 今日 3发2收 + 昨日 5发(不应计入) for i in range(3): session.add(M(msg_id=f"t1s{i}", channel="wechat", owner_account_id="o1", talker_id="f1", is_send=True, timestamp=now_ms - i * 1000)) for i in range(2): session.add(M(msg_id=f"t1r{i}", channel="wechat", owner_account_id="o1", talker_id="f1", is_send=False, timestamp=now_ms - i * 1000)) for i in range(5): session.add(M(msg_id=f"t0s{i}", channel="wechat", owner_account_id="o1", talker_id="f1", is_send=True, timestamp=yesterday_ms)) # f2: 无消息 await session.commit() async def test_customers_today_send_receive(make_client): """发送/接收今日列 = 本地消息表按 Asia/Shanghai 今日聚合。""" client = await make_client(_seed_today_counts) r = await client.get("/api/customers", params={"channel": "wechat"}) assert r.status_code == 200 by_fid = {it["friend_id"]: it for it in r.json()["items"]} assert by_fid["f1"]["send_today"] == 3 assert by_fid["f1"]["receive_today"] == 2 assert by_fid["f2"]["send_today"] == 0 assert by_fid["f2"]["receive_today"] == 0 # --------------------------------------------------------------------------- # GET /api/customers?q= — 文本搜索(昵称/备注/friend_id/手机号,LIKE 通配转义) # 对齐云客通讯录搜索框「支持微信昵称、微信号、备注名、手机号搜索」。 # --------------------------------------------------------------------------- async def _seed_q(session): from app.db.models import Customer as C session.add_all([ # 昵称命中「阿里」 C(channel="wechat", owner_account_id="o1", friend_id="wx_nick", friend_nickname="阿里巴巴顾问"), # 备注命中「阿里」 C(channel="wechat", owner_account_id="o1", friend_id="wx_remark", friend_nickname="腾讯小哥", friend_remark="阿里系客户"), # 不含词;手机号唯一可命中 C(channel="wechat", owner_account_id="o1", friend_id="wx_plain", friend_nickname="网易产品", friend_remark="同行", phone="13900002222"), # friend_id 唯一可命中;顺带另一手机号 C(channel="wechat", owner_account_id="o1", friend_id="wxid_ali888", friend_nickname="字节跳动", phone="13866668888"), ]) await session.commit() async def test_q_matches_nickname_and_remark(make_client): """q=阿里 -> 昵称命中 + 备注命中(2 行),其余行不回。""" client = await make_client(_seed_q) r = await client.get("/api/customers", params={"q": "阿里"}) assert r.status_code == 200 body = r.json() assert body["total"] == 2 assert {it["friend_id"] for it in body["items"]} == {"wx_nick", "wx_remark"} async def test_q_matches_friend_id_and_phone(make_client): """q 命中 friend_id / phone 两列。""" client = await make_client(_seed_q) r = await client.get("/api/customers", params={"q": "ali888"}) assert r.status_code == 200 assert {it["friend_id"] for it in r.json()["items"]} == {"wxid_ali888"} r2 = await client.get("/api/customers", params={"q": "1390000"}) assert r2.status_code == 200 assert {it["friend_id"] for it in r2.json()["items"]} == {"wx_plain"} async def test_q_empty_no_filter(make_client): """q 空串 -> 不过滤,全量返回。""" client = await make_client(_seed_q) r = await client.get("/api/customers", params={"q": ""}) assert r.status_code == 200 assert r.json()["total"] == 4 async def test_q_escapes_like_wildcards(make_client): """q 里的 % 按字面匹配(镜像 messages 的 q 转义行为)。""" async def seed(session): from app.db.models import Customer as C session.add_all([ C(channel="wechat", friend_id="wx_pct", friend_nickname="折扣100%off"), C(channel="wechat", friend_id="wx_digit", friend_nickname="折扣1001off"), ]) await session.commit() client = await make_client(seed) r = await client.get("/api/customers", params={"q": "100%"}) assert r.status_code == 200 body = r.json() # 只命中字面含 "100%" 的行;不因 % 通配漏扩到 "1001" assert body["total"] == 1 assert [it["friend_id"] for it in body["items"]] == ["wx_pct"] # --------------------------------------------------------------------------- # owner 维度读 customer_account_link(共享好友,2026-09-08) # --------------------------------------------------------------------------- async def _seed_links(session): from app.db.models import CustomerAccountLink as L # 档案路径的旧行:共享好友 f_shared 档案 last-wins 挂在 wx_b 名下 session.add_all([ L(channel="wechat", owner_account_id="wx_a", friend_id="f_shared", friend_remark="A的备注", friend_nickname="共享客户", last_chat_time=2000, send_num=3, receive_num=1, from_type=1), L(channel="wechat", owner_account_id="wx_b", friend_id="f_shared", friend_remark="B的备注", friend_nickname="共享客户", last_chat_time=1000), L(channel="wechat", owner_account_id="wx_a", friend_id="f_only_a", friend_nickname="A独有", last_chat_time=3000), ]) await session.commit() async def test_owner_filter_reads_links(make_client): """owner 过滤走 link:共享好友在两个员工名下各出现,备注各是各的。""" client = await make_client(_seed_links) r_a = await client.get("/api/customers", params={ "channel": "wechat", "owner_account_id": "wx_a"}) assert r_a.status_code == 200 body_a = r_a.json() assert body_a["total"] == 2 by_id = {it["friend_id"]: it for it in body_a["items"]} # 排序:last_chat_time desc assert [it["friend_id"] for it in body_a["items"]] == ["f_only_a", "f_shared"] assert by_id["f_shared"]["friend_remark"] == "A的备注" assert by_id["f_shared"]["send_num"] == 3 assert by_id["f_shared"]["from_type"] == 1 r_b = await client.get("/api/customers", params={ "channel": "wechat", "owner_account_id": "wx_b"}) body_b = r_b.json() assert body_b["total"] == 1 assert body_b["items"][0]["friend_remark"] == "B的备注" async def test_owner_filter_q_and_friend_id_on_links(make_client): """link 路径的 q 四列搜索 + friend_id 精确查找(聊天窗名字解析用)。""" client = await make_client(_seed_links) r = await client.get("/api/customers", params={ "channel": "wechat", "owner_account_id": "wx_a", "q": "A的备注"}) assert r.status_code == 200 body = r.json() assert [it["friend_id"] for it in body["items"]] == ["f_shared"] r2 = await client.get("/api/customers", params={ "channel": "wechat", "owner_account_id": "wx_a", "friend_id": "f_shared"}) body2 = r2.json() assert body2["total"] == 1 assert body2["items"][0]["friend_nickname"] == "共享客户" async def test_owner_filter_with_dept_falls_back_to_archive(make_client): """owner+department 组合回退档案路径(旧行为;link 不解析部门)。""" from app.db.models import Customer as C, Staff as S async def seed(session): await _seed_links(session) session.add(S(user_id="u1", name="员工", dept_id="d1")) session.add(C(channel="wechat", owner_account_id="wx_a", friend_id="f_archived", user_id="u1")) await session.commit() client = await make_client(seed) r = await client.get("/api/customers", params={ "channel": "wechat", "owner_account_id": "wx_a", "department_id": "d1"}) assert r.status_code == 200 body = r.json() # 档案路径:只命中 user_id 属于 d1 的档案行(link 行不带 user_id) assert [it["friend_id"] for it in body["items"]] == ["f_archived"]