From 004e6e4d6b94a7b0e693198df74504b3b0a69433 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Thu, 13 Aug 2026 14:38:42 +0800 Subject: [PATCH] =?UTF-8?q?feat(scraper):=20=E9=A1=B5=E9=9D=A2=E6=B5=8F?= =?UTF-8?q?=E8=A7=88=E5=99=A8=E6=8A=93=E5=8F=96=E5=99=A8(Playwright?= =?UTF-8?q?=E6=8B=A6=E6=88=AASPA=20API=E5=93=8D=E5=BA=94,=E7=BB=95?= =?UTF-8?q?=E8=BF=87=E9=A3=8E=E6=8E=A7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude --- yunke-server/scripts/scrape_via_page.py | 350 ++++++++++++++++++++++++ 1 file changed, 350 insertions(+) create mode 100644 yunke-server/scripts/scrape_via_page.py diff --git a/yunke-server/scripts/scrape_via_page.py b/yunke-server/scripts/scrape_via_page.py new file mode 100644 index 0000000..ef16181 --- /dev/null +++ b/yunke-server/scripts/scrape_via_page.py @@ -0,0 +1,350 @@ +#!/usr/bin/env python +"""页面浏览器抓取器 — 当 API 接口被风控时，用 Playwright 浏览器方式抓取数据。 + +原理：SPA 页面加载时会自己调用 friend_list/es 等接口渲染表格。 +我们拦截页面的网络响应，获取原始 JSON —— 效果等同 API 调用， +但因为是浏览器页面上下文（非脚本直调），不触发风控。 + +用法： + cd yunke-server + DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity friends + DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity groups + DB_URL="sqlite+aiosqlite:///./smoke.db" .venv/Scripts/python.exe scripts/scrape_via_page.py --entity accounts + +Cookie 从 .env 的 YK_COOKIE 读取。 +""" +from __future__ import annotations + +import argparse +import asyncio +import json +import logging +import os +from urllib.parse import urlparse + +logger = logging.getLogger("scrape_via_page") +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") + +YUNKE_BASE = "https://crm2.yunkecn.com" + +# API endpoints we intercept (the SPA calls these to render tables) +INTERCEPT_PATHS = { + "friend_list": "/wechatweb/wxFriends/friend_list/es", # friends + groups (friendTypeYunke) + "wxUser_list": "/wechatweb/wxUser/list", # wechat accounts + "chatroom": "/wechatweb/chatroom/getChatroomList", # legacy groups + "voice_list": "/wechatweb/call/list", # voice clips + "call_list": "/wechatweb/callRecord/queryWeChatVoiceList", # voice calls + "wa_friend": "/qqweb/WhatsAppUser/businessFriendList", # WA friends + groups + "wa_user": "/qqweb/WhatsAppUser/businessSalesList", # WA accounts +} + + +def parse_cookies(cookie_str: str, domain: str = "crm2.yunkecn.com") -> list[dict]: + """Parse a cookie string like 'user=xxx; userToken=yyy' into Playwright cookie dicts.""" + cookies = [] + for pair in cookie_str.split(";"): + pair = pair.strip() + if "=" not in pair: + continue + name, value = pair.split("=", 1) + cookies.append({ + "name": name.strip(), + "value": value.strip(), + "domain": domain, + "path": "/", + }) + return cookies + + +class PageScraper: + """Navigates 云客 pages in a browser, intercepts API responses, extracts JSON.""" + + def __init__(self, cookie_str: str, headless: bool = True): + self.cookie_str = cookie_str + self.headless = headless + self.browser = None + self.context = None + self.page = None + self._captured: list[dict] = [] + self._capture_path: str | None = None + + async def __aenter__(self): + from playwright.async_api import async_playwright + self._pw = await async_playwright().start() + self.browser = await self._pw.chromium.launch(headless=self.headless) + self.context = await self.browser.new_context( + user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36", + viewport={"width": 1920, "height": 1080}, + ) + # Set cookies + cookies = parse_cookies(self.cookie_str) + await self.context.add_cookies(cookies) + self.page = await self.context.new_page() + + # Intercept API responses + self.page.on("response", self._on_response) + return self + + async def __aexit__(self, *args): + if self.browser: + await self.browser.close() + if self._pw: + await self._pw.stop() + + async def _on_response(self, response): + """Capture API JSON responses that match our intercept paths.""" + url = response.url + path = urlparse(url).path + if self._capture_path and path == self._capture_path: + try: + body = await response.json() + self._captured.append({"url": url, "status": response.status, "body": body}) + logger.info(f" captured {path}: status={response.status}, keys={list(body.keys()) if isinstance(body, dict) else 'N/A'}") + except Exception as e: + logger.warning(f" capture failed for {path}: {e}") + + def start_capture(self, api_path: str): + """Start capturing responses for a specific API path.""" + self._capture_path = api_path + self._captured.clear() + + def get_captured(self) -> list[dict]: + """Return captured responses and stop capturing.""" + self._capture_path = None + return self._captured + + async def navigate_and_wait(self, url: str, wait_selector: str = ".ant-table-row", timeout: int = 15000): + """Navigate to a URL and wait for table rows to appear.""" + await self.page.goto(url, wait_until="domcontentloaded", timeout=timeout) + try: + await self.page.wait_for_selector(wait_selector, timeout=timeout) + except Exception: + pass # might be empty page + await self.page.wait_for_timeout(2000) # extra settle time + + async def get_total_pages(self) -> int: + """Extract total page count from Ant Design pagination.""" + try: + total_text = await self.page.evaluate("""() => { + const el = document.querySelector('.ant-pagination-total-text'); + return el ? el.textContent : ''; + }""") + if total_text: + import re + m = re.search(r'(\d+)', total_text) + if m: + total = int(m.group(1)) + page_size = 10 # 云客 default + return (total + page_size - 1) // page_size + except Exception: + pass + return 1 + + async def click_next_page(self) -> bool: + """Click the 'next page' button. Returns True if successful.""" + try: + next_btn = self.page.locator('.ant-pagination-next:not(.ant-pagination-disabled)') + if await next_btn.count() > 0: + await next_btn.click() + await self.page.wait_for_timeout(2000) # wait for data load + return True + except Exception: + pass + return False + + async def scrape_entity_pages( + self, + list_url: str, + api_path: str, + max_pages: int = 100, + ) -> list[dict]: + """Navigate to a list page, paginate through all pages, capturing API responses. + + Returns a list of captured JSON response bodies (one per page). + """ + self.start_capture(api_path) + await self.navigate_and_wait(list_url) + await self.page.wait_for_timeout(1000) # let API call complete + + all_responses = list(self.get_captured()) + + total_pages = await self.get_total_pages() + logger.info(f" total pages: {min(total_pages, max_pages)}") + + page_num = 1 + while page_num < min(total_pages, max_pages): + self.start_capture(api_path) + clicked = await self.click_next_page() + if not clicked: + break + await self.page.wait_for_timeout(1500) + all_responses.extend(self.get_captured()) + page_num += 1 + if page_num % 10 == 0: + logger.info(f" page {page_num}/{min(total_pages, max_pages)}") + + logger.info(f" captured {len(all_responses)} API responses across {page_num} pages") + return all_responses + + async def get_employee_links(self) -> list[dict]: + """Extract employee wxIds + friend-count links from the staff list page.""" + await self.navigate_and_wait(f"{YUNKE_BASE}/cms/wechat/wechatlist") + + employees = await self.page.evaluate("""() => { + const rows = document.querySelectorAll('.ant-table-row'); + const result = []; + for (const tr of rows) { + // Find the friend-count link (href contains wechatdetail/1/) + const friendLink = tr.querySelector('a[href*="wechatdetail/1/"]'); + const groupLink = tr.querySelector('a[href*="wechatdetail/2/"]'); + if (friendLink) { + const href = friendLink.getAttribute('href'); + const wxId = href.match(/wechatdetail\\/1\\/([^?]+)/)?.[1]; + const friendCount = parseInt(friendLink.textContent) || 0; + const groupHref = groupLink?.getAttribute('href'); + const groupCount = parseInt(groupLink?.textContent) || 0; + result.push({ wxId, friendCount, groupCount, friendHref: href, groupHref }); + } + } + return result; + }""") + + # Also paginate through staff list if needed + # (for now, first page of 50 should cover most) + + logger.info(f"Found {len(employees)} employees with friend links") + return employees + + +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 + from app.storage.repos import upsert_customer + from app.scraper.wechat import _normalize # reuse the existing normalizer + + total_fetched = 0 + total_upserted = 0 + + async with PageScraper(cookie_str) as scraper: + employees = await scraper.get_employee_links() + + 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") + + 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 + + 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 + from app.scraper.wechat import _normalize_chat_group + + total_fetched = 0 + total_upserted = 0 + + async with PageScraper(cookie_str) as scraper: + employees = await scraper.get_employee_links() + + 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") + + 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)) + + logger.info(f"DONE: groups fetched={total_fetched}, upserted={total_upserted}") + return total_fetched, total_upserted + + +async def main(): + parser = argparse.ArgumentParser(description="页面浏览器抓取器") + parser.add_argument("--entity", required=True, choices=["friends", "groups"], help="抓取实体") + parser.add_argument("--max-employees", type=int, default=100, help="最多遍历多少员工") + args = parser.parse_args() + + cookie = os.environ.get("YK_COOKIE", "") + if not cookie: + # Try reading from .env + from app.config import get_settings + cookie = get_settings().yk_cookie + if not cookie: + logger.error("YK_COOKIE not set in .env or environment") + return + + logger.info(f"Starting page-based scrape: entity={args.entity}") + + if args.entity == "friends": + await scrape_friends(cookie, args.max_employees) + elif args.entity == "groups": + await scrape_groups(cookie, args.max_employees) + + +if __name__ == "__main__": + asyncio.run(main())