2026如何抓取亚马逊的数据（全指南)

亚马逊是全球最大的电子商务平台，蕴藏着海量的商品数据、客户反馈和市场趋势信息。无论是卖家监控竞争对手、研究人员分析市场动态，还是开发者构建价格追踪工具，亚马逊数据都具有极高的价值。

然而，亚马逊也是公认最难抓取的网站之一，其复杂的反爬机制让许多开发者望而却步。本文将为你提供一份完整的亚马逊数据抓取解决方案，从手动爬虫的实战技巧，到规模化面临的挑战，再到如何利用住宅代理构建稳定高效的数据采集Pipeline。

一、为什么要抓取亚马逊数据？

作为全球最大的电商平台，亚马逊的数据价值体现在多个方面：

应用场景	数据类型	商业价值
竞品监控	价格、评分、评论数	实时调整定价策略，保持竞争力
选品研究	畅销榜单、新品发布	发现热门品类，优化库存决策
评论分析	评论文本、评分趋势	洞察用户痛点，改进产品设计
SEO优化	标题、关键词、排名	优化Listing，提升搜索曝光

二、使用Python手动抓取亚马逊数据

在开始编写代码前，你需要先了解亚马逊的页面结构和反爬特点。

2.1 环境准备

# 创建项目目录
mkdir amazon-scraper && cd amazon-scraper

# 安装必要库
pip3 install beautifulsoup4 requests pandas playwright
playwright install

2.2 了解亚马逊页面结构

亚马逊的产品列表页和详情页有不同的数据暴露方式：

页面类型	数据内容	加载方式	抓取难度
列表页	标题、价格、评分、评论数	服务端渲染+动态加载	中等
详情页	描述、变体、QA、评论	大量动态内容	高

要分析页面结构，可以右键点击网页元素选择“检查”，在开发者工具中查看HTML标签和属性。重点关注：

• 商品卡片的容器元素（通常有data-component-type="s-search-result"属性）
• 价格元素的选择器（如.a-price-whole）
• 评分元素的标签结构

2.3 基础爬虫代码：抓取亚马逊商品列表

以下代码使用Playwright异步模式抓取亚马逊搜索结果页的商品信息：

import asyncio
from playwright.async_api import async_playwright
import pandas as pd
import random

async def scrape_amazon_search(keyword="headphones", max_pages=1):
    """
    抓取亚马逊搜索结果页的商品信息
    """
    async with async_playwright() as pw:
        # 启动浏览器
        browser = await pw.chromium.launch(headless=True)
        page = await browser.new_page()

        # 设置随机User-Agent
        await page.set_extra_http_headers({
            "User-Agent": random.choice([
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
                "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
            ])
        })

        all_products = []

        for page_num in range(1, max_pages + 1):
            # 构建搜索URL
            url = f"https://www.amazon.com/s?k={keyword}&page={page_num}"
            print(f"正在抓取第 {page_num} 页: {url}")

            try:
                # 访问页面
                await page.goto(url, timeout=60000, wait_until="domcontentloaded")

                # 等待商品卡片加载
                await page.wait_for_selector('div[data-component-type="s-search-result"]', timeout=10000)

                # 提取所有商品卡片
                products = await page.query_selector_all('div[data-component-type="s-search-result"]')

                for product in products:
                    try:
                        # 提取标题
                        title_elem = await product.query_selector('h2 a span')
                        title = await title_elem.inner_text() if title_elem else "N/A"

                        # 提取价格
                        price_whole = await product.query_selector('.a-price-whole')
                        price_fraction = await product.query_selector('.a-price-fraction')

                        if price_whole and price_fraction:
                            price = f"${await price_whole.inner_text()}.{await price_fraction.inner_text()}"
                        else:
                            price = "N/A"

                        # 提取评分
                        rating_elem = await product.query_selector('span[aria-label*="out of 5 stars"]')
                        rating = await rating_elem.get_attribute('aria-label') if rating_elem else "N/A"

                        # 提取评论数
                        reviews_elem = await product.query_selector('span[aria-label*="stars"] + span a')
                        reviews = await reviews_elem.inner_text() if reviews_elem else "0"

                        # 提取ASIN
                        asin = await product.get_attribute('data-asin')

                        all_products.append({
                            "title": title[:100] + "..." if len(title) > 100 else title,
                            "price": price,
                            "rating": rating,
                            "reviews": reviews,
                            "asin": asin,
                            "page": page_num
                        })
                    except Exception as e:
                        print(f"解析单个商品出错: {e}")
                        continue

                # 随机延迟，避免被识别为爬虫
                await asyncio.sleep(random.uniform(2, 5))

            except Exception as e:
                print(f"抓取第 {page_num} 页失败: {e}")
                break

        await browser.close()
        return all_products

# 运行爬虫
results = asyncio.run(scrape_amazon_search(keyword="wireless earbuds", max_pages=2))

# 保存到CSV
df = pd.DataFrame(results)
df.to_csv('amazon_products.csv', index=False, encoding='utf-8-sig')
print(f"抓取完成，共 {len(results)} 条商品记录")

2.4 常见问题与处理技巧

即使代码正确，第一次运行时也可能失败。以下是几个关键应对策略：

问题	表现	解决方案
请求被拒	返回503、403状态码	轮换User-Agent，使用代理IP
出现验证码	页面重定向到验证页	降低请求频率，使用住宅代理
数据加载不全	商品列表只显示部分	增加wait_for_selector等待时间
IP被封	所有请求失败	切换到新的代理IP

三、亚马逊进阶抓取技术

当你要从单次抓取升级到常态化采集时，需要掌握以下进阶技巧：

3.1 处理分页

亚马逊的结果列表通常有多个页面，通过URL参数&page={n}翻页：

base_url = "https://www.amazon.com/s?k=headphones"

for page in range(1, 6):
    # 构建分页URL
    url = f"{base_url}&page={page}"

    # 抓取逻辑...

    # 关键：随机延迟，避免被限流
    import random
    import time
    time.sleep(random.uniform(3, 7))

3.2 绕过广告和赞助商品

搜索结果中混杂着广告，需要过滤：

# 检测是否包含"Sponsored"标签
sponsored = await product.query_selector('span:has-text("Sponsored")')
if sponsored:
    continue  # 跳过广告商品

3.3 处理动态加载内容

评论、QA等区域通过AJAX动态加载，需要显式等待：

# 等待"加载更多"按钮出现
await page.wait_for_selector('li.a-last a', timeout=5000)

# 点击加载更多
await page.click('li.a-last a')

# 等待新内容加载
await page.wait_for_timeout(2000)

# 继续抓取新加载的内容

3.4 反封锁策略要点

策略	实现方式	效果
随机延迟	time.sleep(random.uniform(2, 5))	避免请求节奏规律化
轮换User-Agent	维护UA列表，每次随机选择	减少浏览器指纹识别
限制并发数	控制同时运行的爬虫数量	避免触发流量异常告警
使用代理IP	配置住宅代理轮换	最有效的反封禁手段

四、规模化抓取的挑战与解决方案

4.1 常见规模化问题

当你从单次抓取升级到常态化采集时，会遇到以下挑战：

挑战	表现	根本原因
IP被封禁	请求返回503、验证码	同一IP高频访问被识别
数据不一致	不同时间抓取结果不同	地域、登录状态影响返回内容
维护成本高	频繁调整选择器	亚马逊前端代码更新
抓取速度慢	单IP被限流	需要分布式采集

4.2 为什么需要稳定的代理IP

代理IP是规模化抓取的基石，其核心价值在于：

IP轮转分散请求

• 将请求分散到海量IP上，模拟普通用户访问
• 彻底规避频率限制和IP黑名单
• 大幅降低被识别为爬虫的概率

获取地域化数据

• 将出口IP锁定在特定城市
• 获取该地区独有的搜索结果和价格
• 实现本地化市场研究

保持会话一致性

• 粘性会话确保任务期间IP不变
• 适合需要登录或添加到购物车的场景
• 避免频繁中断导致任务失败

kookeey动态住宅代理IP

kookeey提供专业的动态住宅代理服务，有效解决规模化抓取难题：

海量纯净IP池

• 超过5500万个真实住宅IP，遍布全球
• IP来源于真实用户，难以被识别为代理
• 极大降低被平台封锁的风险

精准地理定位

• 支持城市和国家级别的精准定位
• 轻松获取目标市场的本地化数据
• 满足区域性市场研究需求

灵活的会话控制

• 支持粘性会话，最长24小时
• 满足需要保持登录状态的场景
• 避免频繁切换IP导致的中断

4.3 在代码中集成kookeey代理

进入kookeey官网，新用户200MB免费测试流量购买流量包，提取线路

将kookeey代理集成到Playwright爬虫中：

# kookeey代理配置
proxy = {
    "server": "http://gate.kookee.info:15959",
    "username": "YOUR_KOOKEEY_USERNAME",
    "password": "YOUR_KOOKEEY_PASSWORD"
}

# 启动浏览器时配置代理
browser = await pw.chromium.launch(
    headless=True,
    proxy=proxy
)

4.5 验证代理是否生效

在开始大规模抓取前，建议先验证代理配置：

import requests

# Kookeey代理配置
proxies = {
    "http": "http://YOUR_USERNAME:YOUR_PASSWORD@gate.kookee.info:15959",
    "https": "http://YOUR_USERNAME:YOUR_PASSWORD@gate.kookee.info:15959"
}

# 访问IP检测网站
try:
    response = requests.get(
        'https://lumtest.com/myip.json',
        proxies=proxies,
        timeout=10
    )
    ip_info = response.json()
    print(f"代理配置成功！当前出口IP: {ip_info['ip']}")
    print(f"地理位置: {ip_info['country']} - {ip_info['city']}")
except Exception as e:
    print(f"代理连接失败: {e}")

五、总结：

三个阶段逐步升级

阶段	目标	核心策略	工具选择
起步阶段	验证思路，小规模测试	理解页面结构，掌握基础爬虫	Python + Playwright，随机延迟
成长阶段	常态化采集，稳定运行	引入代理IP，优化反封策略	kookeey代理 + 轮换User-Agent
规模化阶段	大规模、分布式采集	IP轮转 + 地理定位 + 粘性会话	kookeey住宅代理 + 并发控制

通过以上方法，你可以将简单的手动脚本升级为能够应对常态化、大规模采集任务的商业级数据采集系统。