如何使用海量代理抓取亚马逊产品数据
Jason Grad
创始人
February 26, 2025
亚马逊提供了大量的产品数据（亚马逊数据），这些数据是企业、研究人员和营销人员的宝贵资源。网络抓取亚马逊需要仔细考虑其反抓取措施。本指南将解释如何使用大规模代理进行有效的 HTTP 请求，以及如何从亚马逊产品页面收集数据，同时避免 IP 封锁。
让我们潜入吧！
亚马逊产品数据的常见用例
说到网络抓取亚马逊，有各种有价值的应用程序：
- 价格监控和优化：通过自动 HTTP 请求跟踪实时定价以调整策略。
- 产品和市场研究：搜集产品详细信息和评论，以深入了解消费者偏好和市场分析。
- 竞争对手分析: 系统地收集有关竞争对手定价和产品策略的数据。
- 库存管理： 监控库存水平以优化供应链。
为什么要在网页抓取中使用代理
代理对于网络抓取亚马逊至关重要，因为它们：
- 避免 IP 禁令: 轮换 IP 地址并将其与轮换用户代理结合使用以逃避检测。
- 访问区域锁定内容: 使用代理来模仿位置并绕过地理限制。
- 绕过速率限制：在多个 IP 之间分发 HTTP 请求以避免限制。
了解亚马逊的反抓取措施
由于采取了各种保护措施，从亚马逊抓取数据并不容易。以下是你可能会遇到的一些常见障碍：
- 验证码： 亚马逊经常使用验证码来区分人类用户和机器人。这些验证码对人类来说很容易，但自动化脚本很难解决。如果管理不力，这些挑战可能会阻止你的抓取尝试。
- 速率限制： 为了防止服务器过载，亚马逊限制了您在特定时间段内可以发出的请求数量。遵守这些限制以避免被封锁非常重要。
- IP 地址阻止： 如果亚马逊检测到来自特定 IP 地址的过度抓取活动，它可能会将其完全阻止。
- 动态页面结构： 亚马逊经常更新其网站的布局和元素，这可能会破坏依赖静态目标元素的现有抓取脚本。这意味着你可能需要不断调整代码以适应这些变化。
例如，我写了一个Python Playwright脚本来从亚马逊的多个产品页面抓取数据。但是，我的刮刀最终被挡住了，如下图所示。
我再也无法抓取数据了，这确实令人沮丧且耗时。不过别担心——我们将研究解决方案：使用大规模住宅代理在雷达下抓取数据。
住宅代理与数据中心代理：哪个最适合亚马逊抓取？
在选择代理来抓取亚马逊时，重要的是要确定哪种类型最好：住宅或数据中心。
住宅代理使用互联网服务提供商 (ISP) 向真实用户提供的 IP 地址，使它们看起来像真正的用户连接，不太可能被发现。
另一方面，数据中心代理来自数据中心，通常更快、更便宜，但它们也不太可能逃避检测。
大规模住宅代理 提供大量 IP 地址池，确保高度匿名性并降低被封锁的风险。
使用海量代理进行亚马逊抓取的好处
大规模住宅代理提供了几个关键好处：
- 🌐 全球影响力： 访问 195 多个国家/地区——目标城市、邮政编码或 ASN
- 🔒 匿名： 数百万个轮换 IP + 可自定义轮换（每个请求或 1-60 分钟）
- ⚡ 速度和可靠性： 99.8% 的成功率，<0.7 秒的响应时间，99.9% 的正常运行时间
- 💰 经济实惠： 可扩展套餐的起价为 4.49 美元/GB
- ✅ 符合道德标准： 完全符合 GDPR/CCPA 的代理，100% 符合道德标准
- 🛠️ 支持： 通过 Slack、Skype、电子邮件或票证
Massive 入门
如果你是 Massive 的新手， 注册一个账号。根据您的需求选择套餐。
注意： 我们提供 2 GB 免费试用 对于公司来说。首先， 填写这张表格。如果你需要更多的带宽， 联系我们的销售团队，我们将为您提供帮助。
注册后，前往 大型仪表板 检索您的代理凭证（用户名和密码）。
配置步骤：
请访问 快速入门 用于自定义代理设置的部分：
- 选择您的首选协议（HTTP、HTTPS 或 SOCKS5）
- 在两者之间选择 旋转或粘性代理
- 设置地理定位偏好（国家、州、城市或邮政编码）
配置完成后，您将获得适用于特定用例的即用型 cURL 命令。
有关基于位置的定位和粘性会话等高级功能，请参阅 海量文档。这些文档提供了充分利用大规模住宅代理的分步说明。
通过此设置，您可以使用Massive Proxies从特定区域抓取亚马逊产品数据。
使用 Python 和 Massive 构建亚马逊抓取工具
让我们来看看使用Massive代理和Playwright构建用于从亚马逊抓取所有数据的Python抓取工具的过程。Playwright 库对于自动化浏览器交互非常有用，尤其是对于动态加载的内容。
我们将从每种亚马逊产品中提取以下数据：产品名称、评级、评论数量、价格、优惠券（如果有）和产品链接。
1。设置你的 Python 环境
首先，请确保您的计算机上安装了 Python。接下来，安装 Playwright 及其必要的浏览器二进制文件：
pip install playwright
playwright install
2。导入所需的库
在您的 Python 文件中，导入用于异步编程和浏览器自动化所需的库：
import asyncio
from playwright.async_api import async_playwright
3.创建抓取函数
定义一个名为 amazon_shopping_search 的异步函数。此函数将采用搜索查询和可选域（默认为 “com”）：
async def amazon_shopping_search(search_query, domain='com'):
4。准备搜索网址
通过将空格替换为，格式化搜索查询+要为搜索创建有效的 URL，请执行以下操作：
q = search_query.replace(" ", "+")
base_url = f"<https://www.amazon>.{domain}/s?k={q}"
5。使用代理设置启动浏览器
使用您的代理设置启动浏览器，然后添加您的 Massive 账户凭证。
async with async_playwright() as p:
browser = await p.chromium.launch(
headless=True,
proxy={
"server": "https://network.joinmassive.com:65535",
"username": "MASSIVE_USERNAME",
"password": "MASSIVE_PASSWORD",
},
)启动后，浏览器发出的所有请求都将通过指定的代理服务器，这有助于保持匿名性。
6。创建新页面
在浏览器中创建一个新页面并初始化一个变量来跟踪当前页码：
page = await browser.new_page()
page_number = 1
7。刮刮圈
实现一个循环，该循环一直持续到没有页面可以抓取为止。对于每次迭代，构造当前页面的 URL 并导航到该页面：
while True:
url = f"{base_url}&page={page_number}"
print(f"Scraping page {page_number}...")
await page.goto(url, wait_until="domcontentloaded")
8。提取产品信息
对于页面上的每个产品，使用 CSS 选择器提取各种详细信息，例如名称、价格、评级、评论、优惠券和链接。如果没有详细资料，请分配 “N/A”。
您可以通过以下方式提取每条信息：
A. 产品名称:
代码片段：
name = await product.query_selector("[data-cy=title-recipe]")
name = await name.inner_text() if name else "N/A"
B. 产品价格：
代码片段：
current_price = await product.query_selector(".a-price > .a-offscreen")
current_price = await current_price.inner_text() if current_price else "N/A"
C. 产品评级：
代码片段：
rating = await product.query_selector("[data-cy=reviews-ratings-slot]")
rating = await rating.inner_text() if rating else "N/A"
D. 产品评论：
代码片段：
reviews = await product.query_selector(
".rush-component > div > span > a > span, a.a-link-normal > span.a-size-base"
)
reviews = await reviews.inner_text() if reviews else "N/A"
E. 产品优惠券:
代码片段：
coupon = await product.query_selector(".s-coupon-unclipped")
coupon = await coupon.inner_text() if coupon else "N/A"
F. 产品链接：
代码片段：
link = await product.query_selector("a.a-link-normal")
if link:
full_link = await link.get_attribute("href")
link = (
f'<https://www.amazon>.{domain}{full_link.split("/ref=")[0]}'
if full_link
else "N/A"
)
else:
link = "N/A"
9。查看下一页
查看下一页的链接。如果它不存在，请退出循环：
next_page = await page.query_selector(".s-pagination-next")
if not next_page:
print("No more pages to scrape.")
break
10。关闭浏览器
抓取所有页面后，关闭浏览器！
await browser.close()
11。主要功能
最后，定义一个主要的异步函数，该函数使用特定的搜索查询启动抓取过程。执行此函数以启动抓取：
async def main():
await amazon_shopping_search(search_query="office chair", domain="com")
# Run the main function
asyncio.run(main())
将数据保存到 CSV
下一步是将这些数据保存到 CSV 文件中，以便进一步分析和数据处理。使用 Python 的内置 csv 模块将提取的数据保存到 CSV 文件中。
import csv
# Open the CSV file once before the loop
with open("product_data.csv", mode="w", newline="", encoding="utf-8") as file:
writer = csv.writer(file)
# Write the header
writer.writerow(["Name", "Current Price", "Rating", "Reviews", "Coupon", "Link"])
# existing code...
for product in products:
# existing code...
# Write the data if the product is valid and not sponsored
if name != "N/A" and "Sponsored" not in name:
writer.writerow([name, current_price, rating, reviews, coupon, link])
# existing code...
完整代码
看看抓取亚马逊产品数据的完整代码：
import asyncio
from playwright.async_api import async_playwright
import csv
async def amazon_shopping_search(search_query, domain="com"):
q = search_query.replace(" ", "+")
base_url = f"https://www.amazon.{domain}/s?k={q}"
async with async_playwright() as p:
# Launch browser with massive proxy to avoid IP blocking
browser = await p.chromium.launch(
headless=True,
proxy={
"server": "https://network.joinmassive.com:65535",
"username": "PROXY_USERNAME",
"password": "PROXY_PASSWORD",
},
)
page = await browser.new_page()
page_number = 1
with open("product_data.csv", mode="w", newline="", encoding="utf-8") as file:
writer = csv.writer(file)
writer.writerow(["Name", "Price", "Rating", "Reviews", "Coupon", "Link"])
while True:
url = f"{base_url}&page={page_number}"
print(f"Scraping page {page_number}...")
try:
await page.goto(url, wait_until="networkidle", timeout=60000)
await page.wait_for_selector(".s-result-item")
except Exception as e:
print(f"Navigation failed: {str(e)}")
break
# Get all product elements
products = await page.query_selector_all(".s-result-item")
for product in products:
# Skip sponsored items
is_sponsored = await product.query_selector(
".s-sponsored-label-text"
)
if is_sponsored:
continue
# Extract data with error handling
try:
name_elem = await product.query_selector("h2 a")
name = await name_elem.inner_text() if name_elem else "N/A"
price_elem = await product.query_selector(
".a-price .a-offscreen"
)
price = await price_elem.inner_text() if price_elem else "N/A"
rating_elem = await product.query_selector(
".a-icon-star-small .a-icon-alt"
)
rating = (
await rating_elem.inner_text() if rating_elem else "N/A"
)
reviews_elem = await product.query_selector(
".a-size-small .a-size-base"
)
reviews = (
await reviews_elem.inner_text() if reviews_elem else "N/A"
)
coupon_elem = await product.query_selector(
".s-coupon-unclipped"
)
coupon = (
await coupon_elem.inner_text() if coupon_elem else "N/A"
)
link = await name_elem.get_attribute("href")
link = (
f"https://www.amazon.{domain}{link.split('?')[0]}"
if link
else "N/A"
)
writer.writerow([name, price, rating, reviews, coupon, link])
except Exception as e:
print(f"Error processing product: {str(e)}")
continue
# Pagination handling
next_disabled = await page.query_selector(
".s-pagination-next.s-pagination-disabled"
)
if next_disabled:
print("Reached last page")
break
page_number += 1
if page_number > 5: # Safety limit
print("Reached page limit")
break
await browser.close()
async def main():
await amazon_shopping_search(search_query="office chair", domain="com")
if __name__ == "__main__":
asyncio.run(main())代码成功运行后，它将把所有抓取的产品数据保存到一个 CSV 文件中：
太棒了！数据易于分析和读取。
总结
本文讨论了利用Massive代理如何帮助您提取有价值的数据，同时最大限度地降低被检测和屏蔽的风险。有关代理配置或最佳实践的更多详细信息，请务必访问我们的 官方文档。
准备好开始了吗？ 立即注册大规模代理 🚀
作者简介
Jason Grad
创始人
我是Massive的联合创始人兼首席执行官。除了在创业公司工作外，我还是一名音乐家、运动员、导师、活动主持人和志愿者。
