From edca9b4c771b74b15b5b0ec8935eb101aebf110d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Thu, 20 Aug 2026 14:04:06 +0800 Subject: [PATCH] =?UTF-8?q?fix(customers):=20WA=E5=AE=A2=E6=88=B7=E7=A0=B4?= =?UTF-8?q?1=E4=B8=87=E4=B8=8A=E9=99=90(=E9=80=90=E5=91=98=E5=B7=A5uid?= =?UTF-8?q?=E8=BF=87=E6=BB=A4,8585=E2=86=9211931=E5=85=A8=E9=87=8F)+?= =?UTF-8?q?=E5=8F=8C=E6=B8=A0=E9=81=93=E9=A1=B5=E8=84=9A=E4=BA=91=E5=AE=A2?= =?UTF-8?q?=E5=8F=A3=E5=BE=84+=E6=B6=88=E6=81=AF=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=E5=88=B7=E6=96=B0=E6=9C=80=E8=BF=91=E6=B2=9F=E9=80=9A=E6=97=B6?= =?UTF-8?q?=E9=97=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - wa-friends 逐员工抓取:全局查询~1万行硬上限(云客11931只翻到10000), uid=员工WA号过滤(真机验证 declared 1145→1149),total-aware翻页 - declared-total 分渠道:Σ wechat/whatsapp_account.contacts_num(8590/11915) - 消息增量同步顺带刷新 customer.last_chat_time(云客列表按最近沟通实时 排序,快照每日一次导致顺序滞后;微信1:1+WA都加) Co-Authored-By: Claude --- yunke-server/app/api/routers/customer.py | 23 ++- yunke-server/scripts/scrape_via_page.py | 184 +++++++++++++++++------ yunke-server/tests/test_api_customers.py | 20 ++- yunkes/src/api/index.ts | 9 +- yunkes/src/views/CustomerList.vue | 13 +- 5 files changed, 187 insertions(+), 62 deletions(-) diff --git a/yunke-server/app/api/routers/customer.py b/yunke-server/app/api/routers/customer.py index 6e0e0dd..c3911ca 100644 --- a/yunke-server/app/api/routers/customer.py +++ b/yunke-server/app/api/routers/customer.py @@ -29,13 +29,16 @@ PageSizeParam = Annotated[int, Query(ge=1, le=100)] @router.get("/customers/declared-total") async def customers_declared_total(session: SessionDep) -> dict: - """云客口径的微信客户合计 = Σ wechat_account.contacts_num。 + """云客口径的客户合计(分渠道)= Σ 员工账号表的 contacts_num 声明值。 云客列表的页脚数字含 ES 索引重复条目(实测 csf 员工声明 1999 vs 真实 - distinct 1449,连行 id 都相同的重复 doc,2026-08-20);本地表存去重真实 - 客户,此端点把「与云客页脚一字不差」的声明口径给前端展示。 + distinct 1449,连行 id 都相同的重复 doc,2026-08-20;WA 同理 11931 vs + 8674);本地表存去重真实客户,此端点把「与云客页脚一字不差」的声明口径 + 给前端展示。 """ - row = ( + from app.db.models import WhatsappAccount + + wechat = ( await session.execute( sa.select( sa.func.coalesce(sa.func.sum(WechatAccount.contacts_num), 0), @@ -43,9 +46,17 @@ async def customers_declared_total(session: SessionDep) -> dict: ) ) ).one() + whatsapp = ( + await session.execute( + sa.select( + sa.func.coalesce(sa.func.sum(WhatsappAccount.contacts_num), 0), + sa.func.count(WhatsappAccount.contacts_num), + ) + ) + ).one() return { - "declared_total": int(row[0]), - "account_count": int(row[1]), + "wechat": {"declared_total": int(wechat[0]), "account_count": int(wechat[1])}, + "whatsapp": {"declared_total": int(whatsapp[0]), "account_count": int(whatsapp[1])}, } diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py index 3a2778d..b6b86c5 100644 --- a/yunke-server/scripts/scrape_via_page.py +++ b/yunke-server/scripts/scrape_via_page.py @@ -339,35 +339,74 @@ async def scrape_friends(cookie_str: str, max_employees: int = 100): async def scrape_groups(cookie_str: str, max_employees: int = 100): - """微信群:浏览器直发 friend_list/es + friendTypeYunke=2 部门级全量翻页 - (与好友共用端点,仅 body 开关不同;2026-08-20 起改 fetch)。""" + """微信群:浏览器直发 friend_list/es,**逐员工**(wechatId 过滤)翻页。 + + 与好友共用端点,仅 friendTypeYunke=2;部门级分页不稳(见 scrape_friends + 三连修,2026-08-20),同模式逐员工 + total-aware 翻页。""" + import datetime as _dt + + from app.db.base import async_engine + from sqlalchemy import text as sa_text from app.scraper.wechat import _normalize_chat_group from app.storage.repos import upsert_chat_group - dept = await _root_dept_id() - base_body: dict = {"friendTypeYunke": "2"} - if dept: - base_body["departmentId"] = dept + async with async_engine.connect() as conn: + result = await conn.execute(sa_text( + "SELECT wx_id FROM wechat_account WHERE wx_id IS NOT NULL AND wx_id != ''" + )) + employees = [row[0] for row in result.fetchall()][:max_employees] + today = _dt.date.today().isoformat() + total_fetched = 0 + total_upserted = 0 async with PageScraper(cookie_str) as scraper: page = await _open_fetch_page(scraper) - rows = await _fetch_pages( - page, INTERCEPT_PATHS["friend_list"], - base_body=base_body, parse_rows=_rows_basic_result, - ) + for i, wx_id in enumerate(employees): + base_body = { + "ymdStart": "0000-00-00", + "ymdEnd": today, + "wechatId": wx_id, + "salesname": "", + "friendTypeYunke": "2", + "isDel": 0, + } + rows: list[dict] = [] + declared: int | None = None + empty_streak = 0 + page_num = 1 + while page_num <= 200: + resp = await _browser_fetch( + page, INTERCEPT_PATHS["friend_list"], + json_body={**base_body, "pageNum": page_num, "pageSize": 50}, + ) + basic = ((resp.get("data") or {}).get("basicResult") or {}) + declared = basic.get("total") or declared + page_rows = basic.get("data") or [] + if not page_rows: + empty_streak += 1 + if empty_streak >= 2 or (declared and len(rows) >= declared): + break + else: + empty_streak = 0 + rows.extend(page_rows) + if declared and len(rows) >= declared: + break + page_num += 1 + await asyncio.sleep(1) - records = [] - for row in rows: - rec = _normalize_chat_group(row) - if rec.get("room_id"): - records.append(rec) + records = [] + for row in rows: + rec = _normalize_chat_group(row) + if rec.get("room_id"): + records.append(rec) + total_fetched += len(records) + if records: + await _upsert_all(upsert_chat_group, records) + total_upserted += len(records) + if (i + 1) % 10 == 0: + logger.info(f" [{i+1}/{len(employees)}] employees done, groups total: {total_fetched}") + await asyncio.sleep(2 + (i % 3)) - total_upserted = 0 - for i in range(0, len(records), 500): - await _upsert_all(upsert_chat_group, records[i:i + 500]) - total_upserted += len(records[i:i + 500]) - - total_fetched = len(records) logger.info(f"DONE: groups fetched={total_fetched}, upserted={total_upserted}") return total_fetched, total_upserted @@ -481,6 +520,8 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100): data.hasLast (paginate_chats, same cursor strategy as the WA chatRecords scraper). """ + from sqlalchemy import text as sa_text + from app.db.base import async_session_maker from app.storage.repos import upsert_messages @@ -599,6 +640,15 @@ async def scrape_messages(cookie_str: str, max_conversations: int = 100): if records: async with async_session_maker() as session: await upsert_messages(session, records) + # 时效:用本对话最新消息时间刷新 customer.last_chat_time + # (云客列表按最近沟通实时排序;friends 快照每日一次,消息增量每4h) + max_ts = max((r.get("timestamp") or 0) for r in records) + if max_ts: + await session.execute(sa_text( + "UPDATE customer SET last_chat_time = :ts " + "WHERE channel='wechat' AND owner_account_id = :o AND friend_id = :f " + "AND (last_chat_time IS NULL OR last_chat_time < :ts)" + ), {"ts": max_ts, "o": owner_id, "f": friend_id}) await session.commit() total_upserted += len(records) @@ -1017,39 +1067,73 @@ async def scrape_wa_accounts(cookie_str: str, max_pages: int = 100): async def scrape_wa_friends(cookie_str: str, max_pages: int = 2000): - """WA 客户:浏览器直发 businessFriendList type=0(与 WA 群共用端点, - 2026-08-20 起改 fetch;注意此端点页码字段是 ``page`` 非 ``pageNum``)。""" + """WA 客户:浏览器直发 businessFriendList type=0,**逐员工**(uid 过滤)翻页。 + + 全局查询有 ~1 万行硬上限(实测云客声明 11931 vs 全局翻页只到 10000, + 2026-08-20);逐员工 ``uid`` 过滤(员工 WA 号)范围小无上限,页码字段是 + ``page``,total-aware 翻页(短页继续,满声明数或连续空页停)。""" + from app.db.base import async_engine + from sqlalchemy import text as sa_text from app.scraper.whatsapp import _normalize_whatsapp_friend from app.storage.repos import upsert_customer dept = await _root_dept_id() - base_body: dict = { - "searchInfo": "", - "friendSearchInfo": "", - "departmentId": dept, - "type": "0", - } + async with async_engine.connect() as conn: + result = await conn.execute(sa_text( + "SELECT uid FROM whatsapp_account WHERE uid IS NOT NULL AND uid != ''" + )) + accounts = [row[0] for row in result.fetchall()] + total_fetched = 0 + total_upserted = 0 async with PageScraper(cookie_str) as scraper: page = await _open_fetch_page(scraper) - rows = await _fetch_pages( - page, INTERCEPT_PATHS["wa_friend"], - base_body=base_body, parse_rows=_rows_overviews, - page_key="page", max_pages=max_pages, - ) + for i, wa_uid in enumerate(accounts): + base_body = { + "searchInfo": "", + "friendSearchInfo": "", + "departmentId": dept, + "type": "0", + "uid": wa_uid, + } + rows: list[dict] = [] + declared: int | None = None + empty_streak = 0 + page_num = 1 + while page_num <= max_pages: + resp = await _browser_fetch( + page, INTERCEPT_PATHS["wa_friend"], + json_body={**base_body, "page": page_num, "pageSize": 50}, + ) + data = resp.get("data") or {} + declared = data.get("totalCount") or declared + page_rows = data.get("overviews") or [] + if not page_rows: + empty_streak += 1 + if empty_streak >= 2 or (declared and len(rows) >= declared): + break + else: + empty_streak = 0 + rows.extend(page_rows) + if declared and len(rows) >= declared: + break + page_num += 1 + await asyncio.sleep(1) - records = [] - for row in rows: - rec = _normalize_whatsapp_friend(row) - if rec.get("friend_id"): - records.append(rec) + records = [] + for row in rows: + rec = _normalize_whatsapp_friend(row) + if rec.get("friend_id"): + records.append(rec) + total_fetched += len(records) + if records: + await _upsert_all(upsert_customer, records) + total_upserted += len(records) + logger.info( + f" [{i+1}/{len(accounts)}] {wa_uid[:18]}..: declared={declared} rows={len(rows)}" + ) + await asyncio.sleep(2 + (i % 3)) - total_upserted = 0 - for i in range(0, len(records), 500): - await _upsert_all(upsert_customer, records[i:i + 500]) - total_upserted += len(records[i:i + 500]) - - total_fetched = len(records) logger.info(f"DONE: wa-friends fetched={total_fetched}, upserted={total_upserted}") return total_fetched, total_upserted @@ -1347,6 +1431,8 @@ async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100): (same cursor strategy as the WeChat msg/chats scraper). """ import time as _time + from sqlalchemy import text as sa_text + from app.db.base import async_session_maker from app.scraper.whatsapp import _is_wa_divider, _normalize_whatsapp_message from app.storage.repos import upsert_messages @@ -1410,6 +1496,14 @@ async def scrape_wa_messages(cookie_str: str, max_conversations: int = 100): # key 字符串 → TypeError, 2026-08-19 WA 阶段首条即崩的根因) async with async_session_maker() as session: await upsert_messages(session, records) + # 时效:刷新 customer.last_chat_time(同微信 1:1,云客按最近沟通实时排序) + max_ts = max((r.get("timestamp") or 0) for r in records) + if max_ts: + await session.execute(sa_text( + "UPDATE customer SET last_chat_time = :ts " + "WHERE channel='whatsapp' AND owner_account_id = :o AND friend_id = :f " + "AND (last_chat_time IS NULL OR last_chat_time < :ts)" + ), {"ts": max_ts, "o": uid, "f": friend_uid}) await session.commit() total_upserted += len(records) diff --git a/yunke-server/tests/test_api_customers.py b/yunke-server/tests/test_api_customers.py index 43e7fc0..a15ad58 100644 --- a/yunke-server/tests/test_api_customers.py +++ b/yunke-server/tests/test_api_customers.py @@ -16,10 +16,22 @@ async def _seed_accounts_for_declared(session): async def test_declared_total_sums_contacts_num(make_client): - """declared_total = Σ wechat_account.contacts_num(云客声明的含重复口径)。""" - client = await make_client(_seed_accounts_for_declared) + """分渠道 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["declared_total"] == 123 - assert body["account_count"] == 2 # 只计有值的员工 + assert body["wechat"]["declared_total"] == 123 + assert body["wechat"]["account_count"] == 2 # 只计有值的员工 + assert body["whatsapp"]["declared_total"] == 15 + assert body["whatsapp"]["account_count"] == 2 diff --git a/yunkes/src/api/index.ts b/yunkes/src/api/index.ts index fc160f1..9045666 100644 --- a/yunkes/src/api/index.ts +++ b/yunkes/src/api/index.ts @@ -48,8 +48,13 @@ export interface CustomerQuery { last_chat_since?: number } -export function getCustomerDeclaredTotal(): Promise<{ declared_total: number; account_count: number }> { - return getJson<{ declared_total: number; account_count: number }>('/customers/declared-total') +export interface DeclaredTotals { + wechat: { declared_total: number; account_count: number } + whatsapp: { declared_total: number; account_count: number } +} + +export function getCustomerDeclaredTotal(): Promise { + return getJson('/customers/declared-total') } export function getCustomers(query: CustomerQuery = {}): Promise> { diff --git a/yunkes/src/views/CustomerList.vue b/yunkes/src/views/CustomerList.vue index 32ebf25..7e77ab7 100644 --- a/yunkes/src/views/CustomerList.vue +++ b/yunkes/src/views/CustomerList.vue @@ -39,15 +39,18 @@ function freshTabState(): TabState { return { items: [], total: 0, page: 1, pageSize: 10, loading: false, errorMsg: null, loaded: false } } -/** 云客口径微信客户合计(Σ员工好友数声明值,含云客索引重复;页脚对齐云客)。 */ -const declaredTotal = ref(null) +/** 云客口径客户合计(分渠道,Σ员工好友数声明值,含云客索引重复;页脚对齐云客)。 */ +const declaredTotals = ref<{ wechat: number | null; whatsapp: number | null }>({ wechat: null, whatsapp: null }) async function loadDeclaredTotal(): Promise { try { const r = await getCustomerDeclaredTotal() - declaredTotal.value = r.declared_total + declaredTotals.value = { + wechat: r.wechat.declared_total, + whatsapp: r.whatsapp.declared_total, + } } catch { - declaredTotal.value = null // 回退真实 total + declaredTotals.value = { wechat: null, whatsapp: null } // 回退真实 total } } @@ -368,7 +371,7 @@ onMounted(() => {

{{ activeTab === 'wechat' ? '微信' : 'WhatsApp' }}客户列表

- 共 {{ activeTab === 'wechat' && declaredTotal !== null ? declaredTotal : tabState[activeTab].total }} 条 + 共 {{ declaredTotals[activeTab] !== null ? declaredTotals[activeTab] : tabState[activeTab].total }} 条