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