wuling-overseas-dealer-expansion-skill/scripts/scraper/ads_power_client.py

"""
AdsPower 指纹浏览器连接客户端
通过 AdsPower 本地 API 连接指纹浏览器，并返回 Playwright browser 实例；默认永不关闭用户浏览器
"""
import time
import requests
from typing import Optional
from playwright.sync_api import Browser, sync_playwright, Page


class AdsPowerClient:
    """AdsPower 客户端"""

    def __init__(self, base_url: str = "http://127.0.0.1:50325", api_key: str = ""):
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.profile_id: Optional[str] = None
        self.browser: Optional[Browser] = None
        self.playwright = None
        self._ws_endpoint: Optional[str] = None

    def _headers(self) -> dict:
        # AdsPower 本地 API 使用 query param 'apikey' 认证，不使用 header
        return {}

    def start_browser(self, profile_id: str, headless: bool = False) -> Browser:
        """
        启动 AdsPower 浏览器并返回 Playwright Browser 实例
        """
        self.profile_id = profile_id
        url = f"{self.base_url}/api/v1/browser/start"
        params = {
            "user_id": profile_id,
            "headless": int(headless),
        }
        if self.api_key:
            params["apikey"] = self.api_key

        try:
            resp = requests.get(url, params=params, headers=self._headers(), timeout=30)
            resp.raise_for_status()
        except requests.RequestException as e:
            raise ConnectionError(f"无法连接 AdsPower API: {e}") from e

        data = resp.json()
        if data.get("code") != 0:
            raise RuntimeError(f"AdsPower 启动浏览器失败: {data}")

        ws_data = data.get("data", {}).get("ws", {})
        # AdsPower 返回的 ws endpoint，优先 puppeteer
        ws_endpoint = ws_data.get("puppeteer") or ws_data.get("selenium")
        if not ws_endpoint:
            raise RuntimeError(f"AdsPower 未返回 ws endpoint: {data}")

        self._ws_endpoint = ws_endpoint

        # 等待浏览器启动
        time.sleep(2)

        self.playwright = sync_playwright().start()

        # 连接 CDP
        try:
            self.browser = self.playwright.chromium.connect_over_cdp(ws_endpoint)
        except Exception as e:
            raise ConnectionError(f"Playwright 连接 AdsPower 浏览器失败: {e}") from e

        return self.browser

    def new_page(self) -> Page:
        """新建一个页面"""
        if not self.browser:
            raise RuntimeError("浏览器未启动，请先调用 start_browser()")
        context = self.browser.contexts[0] if self.browser.contexts else self.browser.new_context()
        return context.new_page()

    def get_open_page(self) -> Optional[Page]:
        """获取 AdsPower 启动时已打开的页面"""
        if not self.browser:
            return None
        if self.browser.contexts and self.browser.contexts[0].pages:
            return self.browser.contexts[0].pages[0]
        return None

    def close_browser(self) -> bool:
        """
        Compatibility method: detach Playwright only, never stop or close AdsPower.

        v4.1 hard rule: scripts must not close the user's logged-in fingerprint
        browser. Keep this method name for older callers, but make it safe.
        """
        self.browser = None
        if self.playwright:
            try:
                self.playwright.stop()
            except Exception:
                pass
            self.playwright = None
        print("AdsPower browser left open; Playwright connection stopped only.")
        return True

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close_browser()


def test_connection(base_url: str = "http://127.0.0.1:50325", profile_id: str = ""):
    """测试 AdsPower 连接"""
    if not profile_id:
        print("请提供 profile_id")
        return

    client = AdsPowerClient(base_url)
    try:
        browser = client.start_browser(profile_id)
        page = client.get_open_page() or client.new_page()
        page.goto("https://www.facebook.com")
        print(f"页面标题: {page.title()}")
        time.sleep(3)
    finally:
        client.close_browser()


if __name__ == "__main__":
    import sys
    if len(sys.argv) >= 3:
        test_connection(sys.argv[1], sys.argv[2])
    else:
        print("用法: python ads_power_client.py <base_url> <profile_id>")
        print("示例: python ads_power_client.py http://127.0.0.1:50325 abc123")