如何爬取亚马逊：2026年伦理数据提取与CAPTCHA解决指南

import requests
import time
import random
from bs4 import BeautifulSoup

user_agents = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36',
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36',
    'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.1 Safari/605.1.15',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 13_1) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.1 Safari/605.1.15',
]

def fetch_amazon_page(url):
    headers = {'User-Agent': random.choice(user_agents)}
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()  # Raise an exception for HTTP errors
        time.sleep(random.uniform(5, 15)) # Random delay
        return response.text
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}")
        return None

# Example usage:
# product_page_url = "https://www.amazon.com/dp/B08XYZ123"
# html_content = fetch_amazon_page(product_page_url)
# if html_content:
#     soup = BeautifulSoup(html_content, 'lxml')
#     # Process soup object

import requests
import time

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY" # 替换为您的实际CapSolver API密钥

def create_aws_waf_task(website_url, aws_key, aws_iv, aws_context, aws_challenge_js, proxy=None):
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": "AntiAwsWafTask", # 如果不使用自己的代理，请使用AntiAwsWafTaskProxyless
            "websiteURL": website_url,
            "awsKey": aws_key,
            "awsIv": aws_iv,
            "awsContext": aws_context,
            "awsChallengeJS": aws_challenge_js
        }
    }
    if proxy:
        payload["task"]["proxy"] = proxy # 如果提供代理，请添加

    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    response.raise_for_status()
    return response.json().get("taskId")

def get_task_result(task_id):
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "taskId": task_id
    }
    while True:
        response = requests.post("https://api.capsolver.com/getTaskResult", json=payload)
        response.raise_for_status()
        result = response.json()
        if result.get("status") == "ready":
            return result.get("solution")
        elif result.get("status") == "failed":
            raise Exception(f"CapSolver任务失败: {result.get('errorDescription')}")
        time.sleep(3) # 每3秒轮询一次

# 示例用法（替换为从亚马逊挑战页面获取的实际值）:
# website_url = "https://efw47fpad9.execute-api.us-east-1.amazonaws.com/latest"
# aws_key = "从亚马逊页面获取的key值"
# aws_iv = "从亚马逊页面获取的iv值"
# aws_context = "从亚马逊页面获取的context值"
# aws_challenge_js = "js挑战脚本的URL"
# proxy_string = "http://user:pass@proxy:port" # 如果使用AntiAwsWafTask，可选

# try:
#     task_id = create_aws_waf_task(website_url, aws_key, aws_iv, aws_context, aws_challenge_js, proxy_string)
#     print(f"CapSolver任务ID: {task_id}")
#     solution = get_task_result(task_id)
#     aws_waf_token = solution.get("cookie")
#     print(f"AWS WAF令牌: {aws_waf_token}")
#     # 在后续请求中使用此令牌作为cookie:
#     # cookies = {'aws-waf-token': aws_waf_token}
#     # response = requests.get(target_url, headers=headers, cookies=cookies)
# except Exception as e:
#     print(f"解决CAPTCHA时出错: {e}")

import requests
import base64
import time

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY" # 替换为您的实际CapSolver API密钥

def solve_aws_waf_classification(image_path, question):
    with open(image_path, "rb") as f:
        image_base64 = base64.b64encode(f.read()).decode("utf-8")

    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": "AwsWafClassification",
            "image": image_base64,
            "question": question
        }
    }
    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    response.raise_for_status()
    task_id = response.json().get("taskId")

    get_payload = {"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
    while True:
        res = requests.post("https://api.capsolver.com/getTaskResult", json=get_payload)
        res.raise_for_status()
        data = res.json()
        if data.get("status") == "ready":
            return data.get("solution")
        elif data.get("status") == "failed":
            raise Exception(f"CapSolver分类任务失败: {data.get('errorDescription')}")
        time.sleep(2)

# 示例用法:
# 假设'captcha_image.png'是下载的CAPTCHA图像文件
# question_text = "选择所有包含自行车的图片" # 与图像相关的问题
# try:
#     result = solve_aws_waf_classification("captcha_image.png", question_text)
#     print(f"Selected indices: {result}")
#     # The result will be a list of indices corresponding to the selected images.
#     # You would then use these indices to interact with the Amazon page.
# except Exception as e:
#     print(f"Error solving image CAPTCHA: {e}")

# ... (之前用于获取HTML内容的代码)

def parse_amazon_product_page(html_content):
    soup = BeautifulSoup(html_content, 'lxml')
    product_data = {}

    # 示例：提取产品标题
    title_element = soup.select_one('#productTitle')
    if title_element:
        product_data['title'] = title_element.get_text(strip=True)

    # 示例：提取产品价格
    price_element = soup.select_one('.a-price .a-offscreen')
    if price_element:
        product_data['price'] = price_element.get_text(strip=True)

    # 示例：提取产品评分
    rating_element = soup.select_one('#acrCustomerReviewText')
    if rating_element:
        product_data['reviews_count'] = rating_element.get_text(strip=True)

    # 根据需要添加更多提取逻辑以获取其他数据点
    return product_data

# 示例用法:
# html_content = fetch_amazon_page("https://www.amazon.com/dp/B08XYZ123")
# if html_content:
#     data = parse_amazon_product_page(html_content)
#     print(data)

import json
import csv

def save_to_json(data, filename):
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=4)
    print(f"数据已保存到 {filename}")

def save_to_csv(data, filename, fieldnames):
    with open(filename, 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(data)
    print(f"数据已保存到 {filename}")

# 示例用法:
# all_product_data = [
#     {'title': '产品A', 'price': '$10.99', 'reviews_count': '1,234 评分'},
#     {'title': '产品B', 'price': '$25.00', 'reviews_count': '567 评分'},
# ]
# save_to_json(all_product_data, 'amazon_products.json')
# save_to_csv(all_product_data, 'amazon_products.csv', ['title', 'price', 'reviews_count'])