移动代理亚马逊数据采集 — 大规模抓取商品

pip install requests beautifulsoup4 pandas

import requests

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36',
    'Accept-Language': 'en-US,en;q=0.9',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
    'Referer': 'https://www.google.com/'
}

import requests
from bs4 import BeautifulSoup

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36',
    'Accept-Language': 'en-US,en;q=0.9',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
}

def scrape_amazon_product(url):
    response = requests.get(url, headers=headers, timeout=15)
    
    if response.status_code != 200:
        print(f"Failed to fetch page. Status code: {response.status_code}")
        return None

    soup = BeautifulSoup(response.text, 'html.parser')

    # 1. Product Title
    title_el = soup.select_one('#productTitle')
    title = title_el.text.strip() if title_el else "N/A"

    # 2. Product Price
    price_el = soup.select_one('.a-price .a-offscreen')
    price = price_el.text.strip() if price_el else "N/A"

    # 3. Rating
    rating_el = soup.select_one('#acrPopover')
    rating = rating_el.get('title') if rating_el else "N/A"

    # 4. Main Product Image
    img_el = soup.select_one('#landingImage')
    image_url = img_el.get('src') if img_el else "N/A"

    return {
        "title": title,
        "price": price,
        "rating": rating,
        "image_url": image_url
    }

# Test with a sample product URL
product = scrape_amazon_product("https://www.amazon.com/dp/B0C3HCD34R")
print(product)

import requests
from bs4 import BeautifulSoup
import pandas as pd

def scrape_search_results(keyword):
    search_url = f"https://www.amazon.com/s?k={keyword}"
    response = requests.get(search_url, headers=headers)
    soup = BeautifulSoup(response.text, 'html.parser')

    items = []
    for card in soup.select('div[data-component-type="s-search-result"]'):
        asin = card.get('data-asin')
        title_el = card.select_one('h2 a span') or card.select_one('h2')
        price_el = card.select_one('.a-price .a-offscreen')
        
        if asin and title_el:
            items.append({
                "asin": asin,
                "title": title_el.text.strip(),
                "price": price_el.text.strip() if price_el else "N/A"
            })
            
    df = pd.DataFrame(items)
    df.to_csv("amazon_search_results.csv", index=False)
    print(f"[+] Saved {len(items)} products to amazon_search_results.csv")
    return items

scrape_search_results("over ear headphones")

import requests
import time

# 1. PXM2 Mobile Proxy Credentials
PROXIES = {
    "http": "http://username:password@proxy.pxm2.io:1080",
    "https": "http://username:password@proxy.pxm2.io:1080",
}

# 2. PXM2 Instant IP Rotation API Webhook
ROTATION_URL = "https://api.pxm2.io/v1/rotate?key=YOUR_API_KEY"

def rotate_mobile_ip():
    """Cycle the cellular carrier IP when Amazon serves a CAPTCHA or 503."""
    print("[*] Bot block detected. Triggering mobile IP change...")
    requests.get(ROTATION_URL, timeout=10)
    time.sleep(8)  # Allow modem to renegotiate radio connection
    print("[+] Fresh cellular IP acquired!")

def fetch_product_safe(url):
    session = requests.Session()
    session.proxies = PROXIES
    session.headers.update(headers)

    for attempt in range(3):
        res = session.get(url, timeout=15)
        
        # Check for CAPTCHA or rate limits
        if res.status_code == 503 or "Robot Check" in res.text:
            rotate_mobile_ip()
            continue
            
        if res.status_code == 200:
            return res.text
            
    return None