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

"""
Facebook 摩洛哥汽车经销商主页访问演示
- 启动 AdsPower
- 在 Facebook 搜索摩洛哥汽车经销商
- 点击第一个结果进入主页
- 截图并停留，不发送任何信息
"""
import time
import random
import argparse
from pathlib import Path
from urllib.parse import urlencode

from playwright.sync_api import 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 search_and_visit(
    profile_id: str,
    query: str = "Morocco car dealer",
    ads_power_url: str = "http://127.0.0.1:50325",
    api_key: str = "",
    screenshot_path: str = "facebook_dealer_visited.png",
    stay_seconds: int = 60
):
    """搜索并访问第一个经销商主页"""
    client = AdsPowerClient(ads_power_url, api_key=api_key)
    page: Page = None

    try:
        print(f"启动 AdsPower profile: {profile_id}")
        client.start_browser(profile_id, headless=False)
        page = client.get_open_page() or client.new_page()
        page.set_viewport_size({"width": 1280, "height": 800})

        # 构建 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))

        # 截图搜索结果
        page.screenshot(path=f"{screenshot_path}_search.png", full_page=False)
        print(f"搜索结果已截图: {screenshot_path}_search.png")

        # 收集页面链接
        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))

        # 截图主页
        page.screenshot(path=screenshot_path, full_page=False)
        print(f"主页已截图: {screenshot_path}")
        print(f"当前页面标题: {page.title()}")
        print(f"当前 URL: {page.url}")

        # 停留，不发送任何信息
        print(f"浏览器将停留 {stay_seconds} 秒，不执行任何发送操作...")
        print("请勿关闭此窗口，关闭后浏览器会自动关闭")
        time.sleep(stay_seconds)

    finally:
        print("断开 Playwright 连接，AdsPower 浏览器保持打开")
        client.close_browser()


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",
                "/profile.php"]
    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="Facebook 摩洛哥汽车经销商主页访问演示")
    parser.add_argument("--profile-id", required=True, help="AdsPower profile ID")
    parser.add_argument("--query", default="Morocco car dealer", help="搜索关键词")
    parser.add_argument("--ads-power-url", default="http://127.0.0.1:50325", help="AdsPower API URL")
    parser.add_argument("--api-key", default="", help="AdsPower API Key（如已开启认证）")
    parser.add_argument("--screenshot", default="facebook_dealer_visited.png", help="截图保存路径")
    parser.add_argument("--stay", type=int, default=60, help="停留秒数")

    args = parser.parse_args()

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


if __name__ == "__main__":
    main()