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

"""
连接已打开的 AdsPower 浏览器，不重新启动/关闭
用于用户已手动开启浏览器并希望直接控制当前窗口的场景
"""
import time
import random
import argparse
from pathlib import Path
from urllib.parse import urlencode

from playwright.sync_api import sync_playwright, Page

try:
    from .ads_power_client import AdsPowerClient
except ImportError:
    import sys
    sys.path.insert(0, str(Path(__file__).parent))
    from ads_power_client import AdsPowerClient


def get_active_ws_endpoint(base_url: str, profile_id: str) -> str:
    """通过 AdsPower API 获取已打开浏览器的 ws endpoint"""
    import requests
    url = f"{base_url}/api/v1/browser/active"
    resp = requests.get(url, params={"user_id": profile_id}, timeout=10)
    resp.raise_for_status()
    data = resp.json()
    if data.get("code") != 0:
        raise RuntimeError(f"获取活动浏览器失败: {data}")
    ws = data["data"]["ws"]["puppeteer"]
    return ws


def control_opened_browser(
    profile_id: str,
    query: str = "concessionnaire voiture Casablanca",
    ads_power_url: str = "http://127.0.0.1:50325",
    screenshot_path: str = "facebook_dealer_controlled.png",
    stay_seconds: int = 60
):
    """连接已打开的浏览器并控制它访问摩洛哥汽车经销商主页"""
    ws_endpoint = get_active_ws_endpoint(ads_power_url, profile_id)
    print(f"连接到已打开的浏览器: {ws_endpoint}")

    playwright = sync_playwright().start()
    browser = playwright.chromium.connect_over_cdp(ws_endpoint)

    # 获取第一个页面
    context = browser.contexts[0] if browser.contexts else browser.new_context()
    page = context.pages[0] if context.pages else context.new_page()
    page.set_viewport_size({"width": 1280, "height": 800})

    try:
        # 构建 Facebook 搜索 URL
        encoded_query = urlencode({"q": query})
        search_url = f"https://www.facebook.com/search/pages/?{encoded_query}"
        print(f"在当前浏览器窗口访问搜索页: {search_url}")
        page.goto(search_url, wait_until="domcontentloaded", timeout=60000)
        time.sleep(random.uniform(4, 6))

        # 等待页面稳定后截图
        time.sleep(random.uniform(3, 5))
        try:
            page.screenshot(path=f"{screenshot_path}_search.png", full_page=False, timeout=60000)
            print(f"搜索结果已截图: {screenshot_path}_search.png")
        except Exception as e:
            print(f"搜索结果截图超时或失败: {e}")

        # 收集页面链接
        links = set()
        elements = page.query_selector_all('a[href*="/"]')
        for el in elements:
            href = el.get_attribute("href")
            if href and _is_valid_page_link(href):
                links.add(_normalize_url(href))

        links = list(links)
        print(f"找到 {len(links)} 个可能的主页链接")

        if not links:
            print("未找到任何主页链接，请检查搜索词或页面是否已加载")
            return

        # 访问第一个主页
        first_link = links[0]
        print(f"准备进入第一个主页: {first_link}")
        page.goto(first_link, wait_until="domcontentloaded", timeout=60000)
        time.sleep(random.uniform(3, 5))

        # 截图主页
        time.sleep(random.uniform(2, 4))
        try:
            page.screenshot(path=screenshot_path, full_page=False, timeout=60000)
            print(f"主页已截图: {screenshot_path}")
        except Exception as e:
            print(f"主页截图超时或失败: {e}")
        print(f"当前页面标题: {page.title()}")
        print(f"当前 URL: {page.url}")

        # 停留，不发送任何信息，不断开 Playwright 连接，浏览器保持打开
        print(f"浏览器将停留 {stay_seconds} 秒，不执行任何发送操作，也不断开 Playwright 连接，浏览器保持打开...")
        print("你可以在自己的电脑屏幕上看到这个窗口")
        time.sleep(stay_seconds)

    finally:
        # 不断开 Playwright 连接，浏览器保持打开，只停止 Playwright（保持浏览器窗口打开）
        print("停止 Playwright，浏览器窗口保持打开")
        playwright.stop()


def _is_valid_page_link(href: str) -> bool:
    """判断是否是有效的 Facebook 主页链接"""
    from urllib.parse import urlparse
    parsed = urlparse(href)
    if parsed.netloc not in ["www.facebook.com", "facebook.com", "m.facebook.com"]:
        return False
    excluded = ["/search/", "/watch", "/groups", "/marketplace",
                "/messages", "/notifications", "/settings",
                "/photo.php", "/videos", "/posts", "/reel"]
    for pattern in excluded:
        if pattern in parsed.path:
            return False
    path = parsed.path.strip("/")
    if not path or "/" in path:
        return False
    return True


def _normalize_url(url: str) -> str:
    """标准化 Facebook URL"""
    from urllib.parse import urlparse
    parsed = urlparse(url)
    path = parsed.path.strip("/").split("/")[0]
    return f"https://www.facebook.com/{path}/"


def main():
    parser = argparse.ArgumentParser(description="控制已打开的 AdsPower 浏览器访问 Facebook 经销商主页")
    parser.add_argument("--profile-id", required=True, help="AdsPower profile ID")
    parser.add_argument("--query", default="concessionnaire voiture Casablanca", help="搜索关键词")
    parser.add_argument("--ads-power-url", default="http://127.0.0.1:50325", help="AdsPower API URL")
    parser.add_argument("--screenshot", default="D:/WEFANBOT/facebook_dealer_controlled.png", help="截图保存路径")
    parser.add_argument("--stay", type=int, default=60, help="停留秒数")

    args = parser.parse_args()

    control_opened_browser(
        profile_id=args.profile_id,
        query=args.query,
        ads_power_url=args.ads_power_url,
        screenshot_path=args.screenshot,
        stay_seconds=args.stay
    )


if __name__ == "__main__":
    main()