移动代理 eBay 数据抓取

/Applications/Google\ Chrome.app/Contents/MacOS/Google\ Chrome --remote-debugging-port=9222 --user-data-dir=/tmp/chrome-debug

import asyncio
from playwright.async_api import async_playwright

async def get_page():
    p = await async_playwright().start()
    browser = await p.chromium.connect_over_cdp("http://localhost:9222")
    context = browser.contexts[0]
    return await context.new_page()

async def search_ebay(page, query):
    url = f"https://www.ebay.com/sch/i.html?_nkw={query.replace(' ', '+')}"
    await page.goto(url, wait_until="load", timeout=45000)

    # If the splashui check appears, wait for automatic redirect
    if "splashui" in page.url or "Pardon" in await page.title():
        print("Waiting for eBay security check to pass...")
        await page.wait_for_url(lambda u: "splashui" not in u, timeout=20000)

    # Wait until product cards appear
    await page.wait_for_selector("li.s-card, li.s-item", timeout=15000)
    return await page.content()

import re
from bs4 import BeautifulSoup

def parse_search_cards(html):
    soup = BeautifulSoup(html, "html.parser")
    cards = soup.select("ul.srp-results > li.s-card, li.s-item")
    
    items = []
    for card in cards:
        # Extract numeric Item ID from attribute or link
        item_id = card.get("data-listingid")
        link = card.select_one("a.s-card__link, a[href*='/itm/']")
        if not link:
            continue
        
        url = link.get("href", "")
        if not item_id:
            m = re.search(r"/itm/(?:.*?/)?(\d{9,14})", url)
            item_id = m.group(1) if m else None

        # Filter out sponsored promo cards like 'Shop on eBay'
        if not item_id or not item_id.isdigit():
            continue

        title_el = card.select_one(".s-card__title, [role='heading']")
        raw_title = title_el.get_text(strip=True) if title_el else ""
        title = re.sub(r"Opens in a new window or tab", "", raw_title, flags=re.I).strip()

        price_el = card.select_one(".s-card__price, .s-item__price")
        price = price_el.get_text(strip=True) if price_el else ""

        img_el = card.select_one("img.s-card__image, img")
        img = img_el.get("src") if img_el else ""

        items.append({
            "id": item_id,
            "title": title,
            "price": price,
            "url": url,
            "image": img
        })
    return items

import json

def parse_product_jsonld(html):
    soup = BeautifulSoup(html, "html.parser")
    for s in soup.find_all("script", type="application/ld+json"):
        try:
            data = json.loads(s.string)
            if data.get("@type") == "Product":
                offers = data.get("offers", {})
                return {
                    "title": data.get("name"),
                    "price": offers.get("price"),
                    "currency": offers.get("priceCurrency"),
                    "availability": offers.get("availability", "").split("/")[-1],
                    "images": [img.get("url") if isinstance(img, dict) else img 
                               for img in data.get("image", [])]
                }
        except Exception:
            continue
    return {}

def parse_item_specifics(html):
    soup = BeautifulSoup(html, "html.parser")
    specs = {}
    for dl in soup.select(".ux-layout-section-evo dl, dl"):
        for dt, dd in zip(dl.find_all("dt"), dl.find_all("dd")):
            key = dt.get_text(strip=True).rstrip(":")
            val = dd.get_text(strip=True)
            if key and val:
                specs[key] = val
    return specs

def parse_auction_data(html):
    soup = BeautifulSoup(html, "html.parser")
    full_text = soup.get_text(separator=" ", strip=True)

    # 1. Bids count
    bids_match = re.search(r"(\d+)\s+bids?", full_text, re.I)
    bids = int(bids_match.group(1)) if bids_match else 0

    # 2. Time left countdown
    time_match = re.search(r"Ends in\s+([\w\s]+?)(?:[A-Z][a-z]+day|\n|$)", full_text)
    time_left = time_match.group(1).strip() if time_match else "Ended"

    # 3. Current Bid Price
    price_el = soup.select_one(".x-price-primary")
    current_bid = price_el.get_text(strip=True) if price_el else ""

    return {
        "is_auction": bids_match is not None,
        "current_bid": current_bid,
        "bids_count": bids,
        "time_left": time_left
    }

import csv

def save_to_csv(products, filename="ebay_products.csv"):
    if not products:
        return
    keys = products[0].keys()
    with open(filename, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=keys)
        writer.writeheader()
        writer.writerows(products)
    print(f"Saved {len(products)} products to {filename}")