feat(scraper): 页面浏览器抓取器(Playwright拦截SPA API响应,绕过风控)

Co-Authored-By: Claude <noreply@anthropic.com>

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())