From 0d04f68c07d907bd1ec2b2b5ca64e218f6a6b0ef Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= <email@sample.com>
Date: Thu, 20 Aug 2026 12:32:03 +0800
Subject: [PATCH] =?UTF-8?q?fix(scraper):=20=E5=BE=AE=E4=BF=A1=E5=AE=A2?=
 =?UTF-8?q?=E6=88=B7=E6=8A=93=E5=8F=96=E4=B8=89=E8=BF=9E=E4=BF=AE(wechatId?=
 =?UTF-8?q?=E8=BF=87=E6=BB=A4+=E5=AE=8C=E6=95=B4body+total-aware=E7=BF=BB?=
 =?UTF-8?q?=E9=A1=B5,6563=E2=86=926846=E5=85=A8=E9=87=8F)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 真机抓包确认过滤字段是 wechatId(原 userId 被无视,每员工拉回部门前1000行)
- 补 ymdStart/ymdEnd/friendTypeYunke=1/isDel=0 完整 body(与云客页面一致)
- total-aware 翻页:端点分页中途掉短页,翻到声明数满或连续空页(重复行幂等)
- 验证:csf 员工声明1999→distinct1846全量到手;云客total含跨页重复水分(1999声明vs1846真实)

Co-Authored-By: Claude <noreply@anthropic.com>
---
 yunke-server/scripts/scrape_via_page.py | 88 +++++++++++++++++++------
 1 file changed, 69 insertions(+), 19 deletions(-)

diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py
index 7efb22d..3a2778d 100644
--- a/yunke-server/scripts/scrape_via_page.py
+++ b/yunke-server/scripts/scrape_via_page.py
@@ -256,34 +256,84 @@ class PageScraper:
 
 
 async def scrape_friends(cookie_str: str, max_employees: int = 100):
-    """微信客户:浏览器直发 friend_list/es 部门级全量翻页(2026-08-20 起,
-    旧导航+拦截被登录墙杀死 0 捕获)。记录自带 owner(wxId),无需逐员工遍历;
-    ``max_employees`` 参数保留兼容 CLI,已不使用。"""
+    """微信客户:浏览器直发 friend_list/es,**逐员工**(userId 过滤)翻页。
+
+    部门级全量分页排序不稳定(实测:云客声明 8691 vs 到手 distinct 6563,
+    8682 行里 2119 行跨页重复挤占页容量漏抓真客户,2026-08-20);逐员工范围
+    小漂移可忽略,且与云客后台「按员工看客户列表」口径一致。员工清单来自
+    本地 wechat_account(user_id 列)。
+    """
+    from app.db.base import async_engine
+    from sqlalchemy import text as sa_text
     from app.scraper.wechat import _normalize
     from app.storage.repos import upsert_customer
 
     dept = await _root_dept_id()
-    base_body = {"departmentId": dept} if dept else {}
+    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]
 
+    import datetime as _dt
+    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):
+            # 真机抓包(2026-08-20 云客客户列表页)确认的过滤参数:wechatId(非
+            # userId!)+ymd 窗口+friendTypeYunke=1+isDel=0
+            base_body = {
+                "ymdStart": "0000-00-00",
+                "ymdEnd": today,
+                "wechatId": wx_id,
+                "salesname": "",
+                "friendTypeYunke": "1",
+                "isDel": 0,
+            }
+            # total-aware 翻页:该端点分页不稳(中途偶发短页),按声明 total
+            # 翻满或连续空页才停;跨页重复行 upsert 幂等无害。
+            rows: list[dict] = []
+            declared: int | None = None
+            empty_streak = 0
+            page_num = 1
+            while page_num <= 400:
+                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(row)
-        if rec.get("friend_id"):
-            records.append(rec)
+            records = []
+            for row in rows:
+                rec = _normalize(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(employees)}] {wx_id[:16]}..: declared={declared} "
+                f"rows={len(rows)} (total fetched {total_fetched})"
+            )
+            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: friends fetched={total_fetched}, upserted={total_upserted}")
     return total_fetched, total_upserted