fix(scraper): 微信客户抓取三连修(wechatId过滤+完整body+total-aware翻页,6563→6846全量)

- 真机抓包确认过滤字段是 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>
This commit is contained in:
浅念
2026-08-20 12:32:03 +08:00
parent 5d22d2b5f5
commit 0d04f68c07
+69 -19
View File
@@ -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