fix(customers): WA客户破1万上限(逐员工uid过滤,8585→11931全量)+双渠道页脚云客口径+消息同步刷新最近沟通时间

- 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 <noreply@anthropic.com>
2026-08-20 14:04:06 +08:00
parent fd3cf0003b
commit edca9b4c77
5 changed files with 187 additions and 62 deletions
@@ -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])},
    }


@@ -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)

@@ -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
@@ -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<DeclaredTotals> {
+  return getJson<DeclaredTotals>('/customers/declared-total')
 }

 export function getCustomers(query: CustomerQuery = {}): Promise<Page<Customer>> {
@@ -39,15 +39,18 @@ function freshTabState(): TabState {
  return { items: [], total: 0, page: 1, pageSize: 10, loading: false, errorMsg: null, loaded: false }
 }

-/** 云客口径微信客户合计(Σ员工好友数声明值,含云客索引重复;页脚对齐云客)。 */
-const declaredTotal = ref<number | null>(null)
+/** 云客口径客户合计(分渠道,Σ员工好友数声明值,含云客索引重复;页脚对齐云客)。 */
+const declaredTotals = ref<{ wechat: number | null; whatsapp: number | null }>({ wechat: null, whatsapp: null })

 async function loadDeclaredTotal(): Promise<void> {
  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(() => {
      <div>
        <h2 style="margin: 0 0 4px; font-size: 16px">{{ activeTab === 'wechat' ? '微信' : 'WhatsApp' }}客户列表</h2>
        <span class="yk-customers__sub">
-          共 {{ activeTab === 'wechat' && declaredTotal !== null ? declaredTotal : tabState[activeTab].total }} 条
+          共 {{ declaredTotals[activeTab] !== null ? declaredTotals[activeTab] : tabState[activeTab].total }} 条
          <template v-if="departmentName"> · 部门：{{ departmentName }}</template>
          <template v-if="ownerId"> · 员工账号：{{ ownerId }}</template>
        </span>