feat(scraper): 五实体导航拦截全改浏览器fetch(登录墙杀死拦截方式0捕获)
- friends/groups/friend_list/es 部门级全量(记录自带owner,告别逐员工遍历) - accounts/wxUser/list form模式(页码字段是page非pageNum,粘页检测保险) - wa-accounts/wa-friends 直发;wa-friends 200页1万行(疑端点1万行硬上限) - _fetch_pages 通用翻页助手:cookie失效首页code告警+粘页检测 - 实测:微信客户5916→6563/WA客户8058→8585/群851→873/账号50→55(捞回登录墙期间漏抓) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
@@ -256,108 +256,68 @@ class PageScraper:
|
||||
|
||||
|
||||
async def scrape_friends(cookie_str: str, max_employees: int = 100):
|
||||
"""Scrape wechat customers (friends) via page navigation."""
|
||||
from app.db.base import async_session_maker
|
||||
"""微信客户:浏览器直发 friend_list/es 部门级全量翻页(2026-08-20 起,
|
||||
旧导航+拦截被登录墙杀死 0 捕获)。记录自带 owner(wxId),无需逐员工遍历;
|
||||
``max_employees`` 参数保留兼容 CLI,已不使用。"""
|
||||
from app.scraper.wechat import _normalize
|
||||
from app.storage.repos import upsert_customer
|
||||
from app.scraper.wechat import _normalize # reuse the existing normalizer
|
||||
|
||||
total_fetched = 0
|
||||
total_upserted = 0
|
||||
dept = await _root_dept_id()
|
||||
base_body = {"departmentId": dept} if dept else {}
|
||||
|
||||
async with PageScraper(cookie_str) as scraper:
|
||||
employees = await scraper.get_employee_links()
|
||||
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, emp in enumerate(employees[:max_employees]):
|
||||
if emp["friendCount"] == 0:
|
||||
continue
|
||||
wx_id = emp["wxId"]
|
||||
url = f"{YUNKE_BASE}/cms/wechat/wechatlist/wechatdetail/1/{wx_id}?ymdStart=0000-00-00&ymdEnd=2099-12-31"
|
||||
logger.info(f"[{i+1}/{len(employees)}] {wx_id}: {emp['friendCount']} friends")
|
||||
records = []
|
||||
for row in rows:
|
||||
rec = _normalize(row)
|
||||
if rec.get("friend_id"):
|
||||
records.append(rec)
|
||||
|
||||
responses = await scraper.scrape_entity_pages(
|
||||
list_url=url,
|
||||
api_path=INTERCEPT_PATHS["friend_list"],
|
||||
max_pages=200,
|
||||
)
|
||||
|
||||
# Extract friend records from captured responses
|
||||
records = []
|
||||
for resp in responses:
|
||||
body = resp.get("body", {})
|
||||
data = body.get("data", {})
|
||||
rows = data.get("basicResult", {}).get("data", []) or data.get("data", [])
|
||||
for row in rows:
|
||||
if isinstance(row, dict):
|
||||
normalized = _normalize(row)
|
||||
normalized["channel"] = "wechat"
|
||||
normalized["owner_account_id"] = wx_id
|
||||
records.append(normalized)
|
||||
|
||||
total_fetched += len(records)
|
||||
|
||||
# Batch upsert
|
||||
if records:
|
||||
async with async_session_maker() as session:
|
||||
for rec in records:
|
||||
await upsert_customer(session, rec)
|
||||
await session.commit()
|
||||
total_upserted += len(records)
|
||||
logger.info(f" upserted {len(records)} customers (total: {total_upserted})")
|
||||
|
||||
# Human-like delay between employees
|
||||
await asyncio.sleep(3 + (i % 3)) # 3-5s random-ish
|
||||
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
|
||||
|
||||
|
||||
async def scrape_groups(cookie_str: str, max_employees: int = 100):
|
||||
"""Scrape wechat groups via page navigation."""
|
||||
from app.db.base import async_session_maker
|
||||
from app.storage.repos import upsert_chat_group
|
||||
"""微信群:浏览器直发 friend_list/es + friendTypeYunke=2 部门级全量翻页
|
||||
(与好友共用端点,仅 body 开关不同;2026-08-20 起改 fetch)。"""
|
||||
from app.scraper.wechat import _normalize_chat_group
|
||||
from app.storage.repos import upsert_chat_group
|
||||
|
||||
total_fetched = 0
|
||||
total_upserted = 0
|
||||
dept = await _root_dept_id()
|
||||
base_body: dict = {"friendTypeYunke": "2"}
|
||||
if dept:
|
||||
base_body["departmentId"] = dept
|
||||
|
||||
async with PageScraper(cookie_str) as scraper:
|
||||
employees = await scraper.get_employee_links()
|
||||
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, emp in enumerate(employees[:max_employees]):
|
||||
if emp.get("groupCount", 0) == 0 or not emp.get("groupHref"):
|
||||
continue
|
||||
wx_id = emp["wxId"]
|
||||
url = f"{YUNKE_BASE}{emp['groupHref']}"
|
||||
logger.info(f"[{i+1}/{len(employees)}] {wx_id}: {emp['groupCount']} groups")
|
||||
records = []
|
||||
for row in rows:
|
||||
rec = _normalize_chat_group(row)
|
||||
if rec.get("room_id"):
|
||||
records.append(rec)
|
||||
|
||||
responses = await scraper.scrape_entity_pages(
|
||||
list_url=url,
|
||||
api_path=INTERCEPT_PATHS["friend_list"],
|
||||
max_pages=100,
|
||||
)
|
||||
|
||||
records = []
|
||||
for resp in responses:
|
||||
body = resp.get("body", {})
|
||||
data = body.get("data", {})
|
||||
rows = data.get("basicResult", {}).get("data", []) or data.get("data", [])
|
||||
for row in rows:
|
||||
if isinstance(row, dict):
|
||||
normalized = _normalize_chat_group(row)
|
||||
normalized["channel"] = "wechat"
|
||||
records.append(normalized)
|
||||
|
||||
total_fetched += len(records)
|
||||
if records:
|
||||
async with async_session_maker() as session:
|
||||
for rec in records:
|
||||
await upsert_chat_group(session, rec)
|
||||
await session.commit()
|
||||
total_upserted += len(records)
|
||||
logger.info(f" upserted {len(records)} groups (total: {total_upserted})")
|
||||
|
||||
await asyncio.sleep(3 + (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
|
||||
|
||||
@@ -835,6 +795,80 @@ def _rows_flat(body: dict) -> list[dict]:
|
||||
return [r for r in rows if isinstance(r, dict)] if isinstance(rows, list) else []
|
||||
|
||||
|
||||
def _rows_basic_result(body: dict) -> list[dict]:
|
||||
"""friend_list/es 信封: ``data.basicResult.data[]``(好友与群共用)。"""
|
||||
data = body.get("data") if isinstance(body, dict) else None
|
||||
basic = data.get("basicResult") if isinstance(data, dict) else None
|
||||
rows = basic.get("data") if isinstance(basic, dict) else None
|
||||
return [r for r in rows if isinstance(r, dict)] if isinstance(rows, list) else []
|
||||
|
||||
|
||||
async def _root_dept_id() -> str:
|
||||
"""本地 department 表的公司根部门 id(多个抓取端点的必填 body 字段)。"""
|
||||
from app.db.base import async_engine
|
||||
from sqlalchemy import text as sa_text
|
||||
|
||||
async with async_engine.connect() as conn:
|
||||
row = (await conn.execute(sa_text(
|
||||
"SELECT dept_id FROM department WHERE parent_id IS NULL LIMIT 1"
|
||||
))).first()
|
||||
return row[0] if row else ""
|
||||
|
||||
|
||||
async def _fetch_pages(
|
||||
page,
|
||||
path: str,
|
||||
*,
|
||||
base_body: dict,
|
||||
parse_rows,
|
||||
page_size: int = 50,
|
||||
max_pages: int = 500,
|
||||
form: bool = False,
|
||||
page_key: str = "pageNum",
|
||||
) -> list[dict]:
|
||||
"""按页码翻页浏览器直发 POST,直到短页;收集全部 rows 返回。
|
||||
|
||||
2026-08-20 起 list 类实体一律走此路径(导航+拦截被登录墙杀死,0 捕获)。
|
||||
``form=True`` 发 form-urlencoded(Spring @RequestParam,如 wxUser/list);
|
||||
``page_key`` 兼容 WA 的 ``page``。首页响应 code 异常时醒目告警
|
||||
cookie 失效(所有 list 端点健康时 code=10000)。
|
||||
"""
|
||||
rows_all: list[dict] = []
|
||||
first_row_sig: str | None = None
|
||||
page_num = 1
|
||||
while page_num <= max_pages:
|
||||
body = {**base_body, page_key: page_num, "pageSize": page_size}
|
||||
if form:
|
||||
resp = await _browser_fetch(page, path, form_body=body)
|
||||
else:
|
||||
resp = await _browser_fetch(page, path, json_body=body)
|
||||
if page_num == 1 and isinstance(resp, dict):
|
||||
code = resp.get("code")
|
||||
if code is not None and str(code) not in ("10000",):
|
||||
logger.warning(
|
||||
f"⚠️ {path} 首页 code={code} ({resp.get('message')}) — "
|
||||
"cookie 可能已过期/失效,请从浏览器重新复制 YK_COOKIE 到 .env!"
|
||||
)
|
||||
rows = parse_rows(resp)
|
||||
# 粘页检测:端点忽略页码参数时第 2 页起首行与第 1 页相同(实测 wxUser/list
|
||||
# 的页码字段是 page 而非 pageNum,2026-08-20)——丢弃重复页并停止
|
||||
if rows:
|
||||
import json as _json
|
||||
sig = _json.dumps(rows[0], sort_keys=True, default=str)[:200]
|
||||
if page_num == 1:
|
||||
first_row_sig = sig
|
||||
elif sig == first_row_sig:
|
||||
logger.warning(f" {path} page {page_num} 与首页相同 — 端点忽略页码参数,停止翻页")
|
||||
break
|
||||
rows_all.extend(rows)
|
||||
logger.info(f" {path.rsplit('/', 1)[-1]} page {page_num}: {len(rows)} rows")
|
||||
if len(rows) < page_size:
|
||||
break
|
||||
page_num += 1
|
||||
await asyncio.sleep(1.5)
|
||||
return rows_all
|
||||
|
||||
|
||||
def _rows_overviews(body: dict) -> list[dict]:
|
||||
"""Extract rows from the WA envelope: ``data.overviews[]``
|
||||
(businessSalesList / businessFriendList)."""
|
||||
@@ -863,91 +897,109 @@ async def _upsert_all(upsert_fn, records: list[dict]) -> None:
|
||||
|
||||
|
||||
async def scrape_accounts(cookie_str: str, max_pages: int = 100):
|
||||
"""Scrape WeChat staff accounts: navigate /cms/wechat/wechatlist,
|
||||
intercept /wechatweb/wxUser/list, upsert to wechat_account (+staff)."""
|
||||
"""微信员工账号:浏览器直发 wxUser/list(form-encoded,2026-08-20 起改 fetch)。
|
||||
|
||||
body 必填 isOnJob/departmentId/startdate/enddate(缺则 -90000);时间窗只
|
||||
影响每账号统计数字不影响返回集合,取近 7 天。"""
|
||||
import datetime as _dt
|
||||
|
||||
from app.scraper.wechat import _normalize_wxuser_account
|
||||
from app.storage.repos import upsert_wechat_account
|
||||
|
||||
total_fetched = 0
|
||||
dept = await _root_dept_id()
|
||||
end_date = _dt.datetime.now().date()
|
||||
start_date = end_date - _dt.timedelta(days=7)
|
||||
|
||||
async with PageScraper(cookie_str) as scraper:
|
||||
responses = await scraper.scrape_entity_pages(
|
||||
list_url=f"{YUNKE_BASE}/cms/wechat/wechatlist",
|
||||
api_path=INTERCEPT_PATHS["wxUser_list"],
|
||||
page = await _open_fetch_page(scraper)
|
||||
rows = await _fetch_pages(
|
||||
page, INTERCEPT_PATHS["wxUser_list"],
|
||||
base_body={
|
||||
"isOnJob": "1",
|
||||
"departmentId": dept,
|
||||
"startdate": start_date.isoformat(),
|
||||
"enddate": end_date.isoformat(),
|
||||
"userId": "",
|
||||
"salesname": "",
|
||||
"weChatNickName": "",
|
||||
},
|
||||
parse_rows=_rows_flat,
|
||||
form=True,
|
||||
page_key="page", # 此端点页码字段是 page(实测 pageNum 被无视)
|
||||
max_pages=max_pages,
|
||||
)
|
||||
records = []
|
||||
for resp in responses:
|
||||
for row in _rows_flat(resp.get("body", {})):
|
||||
records.append(_normalize_wxuser_account(row))
|
||||
total_fetched = len(records)
|
||||
await _upsert_all(upsert_wechat_account, records)
|
||||
|
||||
logger.info(f"DONE: accounts fetched={total_fetched}, upserted={total_fetched}")
|
||||
return total_fetched, total_fetched
|
||||
records = [_normalize_wxuser_account(row) for row in rows]
|
||||
await _upsert_all(upsert_wechat_account, records)
|
||||
total = len(records)
|
||||
logger.info(f"DONE: accounts fetched={total}, upserted={total}")
|
||||
return total, total
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WhatsApp staff accounts (navigation + interception)
|
||||
# WhatsApp staff accounts (browser fetch)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def scrape_wa_accounts(cookie_str: str, max_pages: int = 100):
|
||||
"""Scrape WA staff accounts: navigate /cms/wechat/waBusiness/userlist,
|
||||
intercept /qqweb/WhatsAppUser/businessSalesList, upsert to whatsapp_account."""
|
||||
"""WA 员工账号:浏览器直发 businessSalesList(JSON body {pageNum,pageSize})。"""
|
||||
from app.scraper.whatsapp import _normalize_whatsapp_account
|
||||
from app.storage.repos import upsert_whatsapp_account
|
||||
|
||||
async with PageScraper(cookie_str) as scraper:
|
||||
responses = await scraper.scrape_entity_pages(
|
||||
list_url=f"{YUNKE_BASE}/cms/wechat/waBusiness/userlist",
|
||||
api_path=INTERCEPT_PATHS["wa_user"],
|
||||
page = await _open_fetch_page(scraper)
|
||||
rows = await _fetch_pages(
|
||||
page, INTERCEPT_PATHS["wa_user"],
|
||||
base_body={}, parse_rows=_rows_overviews,
|
||||
max_pages=max_pages,
|
||||
)
|
||||
records = []
|
||||
for resp in responses:
|
||||
for row in _rows_overviews(resp.get("body", {})):
|
||||
records.append(_normalize_whatsapp_account(row))
|
||||
total = len(records)
|
||||
await _upsert_all(upsert_whatsapp_account, records)
|
||||
|
||||
records = [_normalize_whatsapp_account(row) for row in rows]
|
||||
await _upsert_all(upsert_whatsapp_account, records)
|
||||
total = len(records)
|
||||
logger.info(f"DONE: wa-accounts fetched={total}, upserted={total}")
|
||||
return total, total
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WhatsApp customers (navigation + interception)
|
||||
# WhatsApp customers (browser fetch)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def scrape_wa_friends(cookie_str: str, max_pages: int = 2000):
|
||||
"""Scrape WA customers: navigate /cms/wechat/waBusiness/friendlist,
|
||||
intercept /qqweb/WhatsAppUser/businessFriendList, upsert to customer
|
||||
(channel='whatsapp'). Live total ~11k rows => high default max_pages."""
|
||||
"""WA 客户:浏览器直发 businessFriendList type=0(与 WA 群共用端点,
|
||||
2026-08-20 起改 fetch;注意此端点页码字段是 ``page`` 非 ``pageNum``)。"""
|
||||
from app.scraper.whatsapp import _normalize_whatsapp_friend
|
||||
from app.storage.repos import upsert_customer
|
||||
|
||||
total_fetched = 0
|
||||
total_upserted = 0
|
||||
async with PageScraper(cookie_str) as scraper:
|
||||
responses = await scraper.scrape_entity_pages(
|
||||
list_url=f"{YUNKE_BASE}/cms/wechat/waBusiness/friendlist",
|
||||
api_path=INTERCEPT_PATHS["wa_friend"],
|
||||
max_pages=max_pages,
|
||||
)
|
||||
# Upsert in chunks so one giant page-walk doesn't build a 11k-record list
|
||||
# before the first commit.
|
||||
chunk: list[dict] = []
|
||||
for resp in responses:
|
||||
records = [_normalize_whatsapp_friend(row) for row in _rows_overviews(resp.get("body", {}))]
|
||||
total_fetched += len(records)
|
||||
chunk.extend(records)
|
||||
if len(chunk) >= 500:
|
||||
await _upsert_all(upsert_customer, chunk)
|
||||
total_upserted += len(chunk)
|
||||
chunk = []
|
||||
await _upsert_all(upsert_customer, chunk)
|
||||
total_upserted += len(chunk)
|
||||
dept = await _root_dept_id()
|
||||
base_body: dict = {
|
||||
"searchInfo": "",
|
||||
"friendSearchInfo": "",
|
||||
"departmentId": dept,
|
||||
"type": "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,
|
||||
)
|
||||
|
||||
records = []
|
||||
for row in rows:
|
||||
rec = _normalize_whatsapp_friend(row)
|
||||
if rec.get("friend_id"):
|
||||
records.append(rec)
|
||||
|
||||
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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user