电商与竞争情报 9 分钟阅读

如何使用代理大规模爬取产品评论

学会从亚马逊和其他平台大规模地刮掉产品评论. Python和Node.js代码用于多平台审查收集,pagination处理,以及情绪分析准备.

ProxyHat Team
如何使用代理大规模爬取产品评论

为什么在规模上进行产品审查?

产品审查是电子商务中最有价值的数据来源之一。 它们揭示了客户情绪、产品质量问题、特征要求和竞争性定位——其他数据来源无法提供的信息。 在规模上,审查数据使:

  • 感知分析: 追踪顾客对您产品和竞争对手产品 的感受
  • 产品开发: 查明成千上万次审查中反复提出的投诉和专题请求。
  • 竞争性情报: 从顾客的言语中了解竞争对手的优缺点.
  • 市场研究: 通过分析各类审查模式,发现未得到满足的需求和新出现的趋势。
  • 质量监测: 通过监测审查情绪趋势,及早发现产品质量问题。

挑战在于审查数据分布在多个平台(亚马逊,沃尔玛,Best Buy,Trustpilot,Google),每个平台具有不同的结构和反机器人保护. 大规模审查需要针对具体平台的战略和强有力的代理基础设施。 关于基本的电子商务刮刮模式,见我们 电子商务数据报废指南。 。 。

审查跨平台的数据结构

平台审查领域倾斜反毒级
亚马逊评分、标题、文本、日期、核实、有帮助的表决基于页面(10页)高级
沃尔玛评级、标题、文本、日期、提交来源基于偏移的 API中型
最佳买家评级、标题、文本、日期、有用/无用基于页面的 API中型
信任驾驶员评分、标题、文本、日期、答复基于页面低
谷歌购物评分、文本、日期、来源基于滚动高级

审查抓取的代理配置

审查刮刮涉及斜拉式导航,这意味着在多个请求之间保持会话。 代理Hat的粘性会话是这种模式的理想.

代理服务器设置

# Per-request rotation for initial product lookups
http://USERNAME:PASSWORD@gate.proxyhat.com:8080
# Sticky session for paginating through reviews of one product
http://USERNAME-session-rev001:PASSWORD@gate.proxyhat.com:8080
# Geo-targeted for region-specific review pages
http://USERNAME-country-US:PASSWORD@gate.proxyhat.com:8080

在审查刮刮时,在对单一产品进行所有审查时使用粘性会议,在不同产品之间移动时使用按要求轮换。 这模仿了自然浏览行为,一个用户在移动到下一个产品之前会读取一个产品的多个评论页.

Python 执行

这里有一个多平台的审查刮刮机 代理哈特的 Python SDK。 。 。

亚马逊评论

import requests
from bs4 import BeautifulSoup
import json
import time
import random
from dataclasses import dataclass
from datetime import datetime
PROXY_URL = "http://USERNAME:PASSWORD@gate.proxyhat.com:8080"
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/124.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 Chrome/124.0.0.0 Safari/537.36",
]
@dataclass
class Review:
    platform: str
    product_id: str
    rating: float
    title: str
    text: str
    date: str
    author: str
    verified: bool
    helpful_votes: int
def scrape_amazon_reviews(asin, max_pages=10):
    """Scrape all reviews for an Amazon product."""
    reviews = []
    session_id = f"rev-{asin}-{random.randint(1000, 9999)}"
    proxy = f"http://USERNAME-session-{session_id}:PASSWORD@gate.proxyhat.com:8080"
    session = requests.Session()
    session.proxies = {"http": proxy, "https": proxy}
    session.headers.update({
        "User-Agent": random.choice(USER_AGENTS),
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "Accept-Language": "en-US,en;q=0.9",
    })
    for page in range(1, max_pages + 1):
        url = (f"https://www.amazon.com/product-reviews/{asin}"
               f"?pageNumber={page}&sortBy=recent")
        try:
            response = session.get(url, timeout=30)
            if response.status_code != 200:
                break
            if "captcha" in response.text.lower():
                print(f"CAPTCHA on page {page}, switching session")
                break
            soup = BeautifulSoup(response.text, "html.parser")
            review_divs = soup.find_all("div", {"data-hook": "review"})
            if not review_divs:
                break
            for div in review_divs:
                review = parse_amazon_review(div, asin)
                if review:
                    reviews.append(review)
            print(f"Page {page}: {len(review_divs)} reviews (total: {len(reviews)})")
            time.sleep(random.uniform(2, 5))
        except requests.RequestException as e:
            print(f"Error on page {page}: {e}")
            break
    return reviews
def parse_amazon_review(div, asin):
    """Parse a single Amazon review element."""
    try:
        rating_el = div.find("i", {"data-hook": "review-star-rating"})
        rating = float(rating_el.get_text().split(" ")[0]) if rating_el else None
        title_el = div.find("a", {"data-hook": "review-title"})
        title = title_el.get_text(strip=True) if title_el else ""
        body_el = div.find("span", {"data-hook": "review-body"})
        text = body_el.get_text(strip=True) if body_el else ""
        date_el = div.find("span", {"data-hook": "review-date"})
        date_str = date_el.get_text(strip=True) if date_el else ""
        author_el = div.find("span", {"class": "a-profile-name"})
        author = author_el.get_text(strip=True) if author_el else ""
        verified = bool(div.find("span", {"data-hook": "avp-badge"}))
        helpful_el = div.find("span", {"data-hook": "helpful-vote-statement"})
        helpful = 0
        if helpful_el:
            text_h = helpful_el.get_text()
            if "one" in text_h.lower():
                helpful = 1
            else:
                nums = [int(s) for s in text_h.split() if s.isdigit()]
                helpful = nums[0] if nums else 0
        return Review(
            platform="amazon",
            product_id=asin,
            rating=rating,
            title=title,
            text=text,
            date=date_str,
            author=author,
            verified=verified,
            helpful_votes=helpful,
        )
    except Exception:
        return None

多平台审查收集器

class ReviewCollector:
    """Collect reviews from multiple platforms for a product."""
    def __init__(self):
        self.scrapers = {
            "amazon": scrape_amazon_reviews,
        }
    def collect_all(self, product_ids: dict) -> list[Review]:
        """
        Collect reviews from all platforms.
        product_ids: {"amazon": "B0CHX3QBCH", "walmart": "12345"}
        """
        all_reviews = []
        for platform, product_id in product_ids.items():
            if platform in self.scrapers:
                print(f"\nScraping {platform} reviews for {product_id}")
                reviews = self.scrapers[platform](product_id)
                all_reviews.extend(reviews)
                print(f"Collected {len(reviews)} reviews from {platform}")
                time.sleep(random.uniform(5, 10))
        return all_reviews
    def to_dataframe(self, reviews: list[Review]):
        """Convert reviews to a pandas DataFrame for analysis."""
        import pandas as pd
        return pd.DataFrame([vars(r) for r in reviews])
# Usage
collector = ReviewCollector()
reviews = collector.collect_all({
    "amazon": "B0CHX3QBCH",
})
print(f"\nTotal reviews collected: {len(reviews)}")

节点.js 执行

节点J审查刮刮机 代理哈特节点 SDK。 。 。

const axios = require("axios");
const cheerio = require("cheerio");
const { HttpsProxyAgent } = require("https-proxy-agent");
function getProxy(sessionId = null) {
  if (sessionId) {
    return `http://USERNAME-session-${sessionId}:PASSWORD@gate.proxyhat.com:8080`;
  }
  return "http://USERNAME:PASSWORD@gate.proxyhat.com:8080";
}
async function scrapeAmazonReviews(asin, maxPages = 10) {
  const reviews = [];
  const sessionId = `rev-${asin}-${Math.floor(Math.random() * 9000 + 1000)}`;
  const agent = new HttpsProxyAgent(getProxy(sessionId));
  for (let page = 1; page <= maxPages; page++) {
    const url = `https://www.amazon.com/product-reviews/${asin}?pageNumber=${page}&sortBy=recent`;
    try {
      const { data } = await axios.get(url, {
        httpsAgent: agent,
        headers: {
          "User-Agent":
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/124.0.0.0 Safari/537.36",
          "Accept-Language": "en-US,en;q=0.9",
        },
        timeout: 30000,
      });
      if (data.toLowerCase().includes("captcha")) {
        console.log(`CAPTCHA on page ${page}`);
        break;
      }
      const $ = cheerio.load(data);
      const reviewDivs = $('[data-hook="review"]');
      if (reviewDivs.length === 0) break;
      reviewDivs.each((_, el) => {
        const $el = $(el);
        const ratingText = $el.find('[data-hook="review-star-rating"]').text();
        const rating = parseFloat(ratingText.split(" ")[0]) || null;
        reviews.push({
          platform: "amazon",
          productId: asin,
          rating,
          title: $el.find('[data-hook="review-title"]').text().trim(),
          text: $el.find('[data-hook="review-body"]').text().trim(),
          date: $el.find('[data-hook="review-date"]').text().trim(),
          author: $el.find(".a-profile-name").text().trim(),
          verified: $el.find('[data-hook="avp-badge"]').length > 0,
        });
      });
      console.log(`Page ${page}: ${reviewDivs.length} reviews (total: ${reviews.length})`);
      await new Promise((r) => setTimeout(r, 2000 + Math.random() * 3000));
    } catch (err) {
      console.error(`Error page ${page}: ${err.message}`);
      break;
    }
  }
  return reviews;
}
// Usage
scrapeAmazonReviews("B0CHX3QBCH", 5).then((reviews) => {
  console.log(`Collected ${reviews.length} reviews`);
  console.log(JSON.stringify(reviews.slice(0, 2), null, 2));
});

缩放处理

检讨降级是大规模检讨刮片中最大的挑战之一.

亚马逊定居战略

亚马逊将评论页数限制在每篇10篇评论,一般显示多达500页(5000篇评论). 对于有更多评论的产品,使用过滤参数进行片段:

# Filter by star rating to get more reviews
star_filters = [
    "one_star", "two_star", "three_star",
    "four_star", "five_star"
]
for star in star_filters:
    url = (f"https://www.amazon.com/product-reviews/{asin}"
           f"?filterByStar={star}&pageNumber={page}")
    # This lets you access more reviews per product

插位会话管理

每种产品的审查标注应使用自己的粘度段. 当完成一个产品并移动到下一个产品时,用不同的IP创建一个新的会话.

阶段代理战略原因
寻找产品按要求轮换独立检查,不需要会议
暂停审查每件产品粘性会议相同的IP跨页看起来很自然
产品之间新会议/IP每种产品的新特性

为感知分析准备数据

原始审查文本需要在情绪分析前进行预处理.

import re
from collections import Counter
def clean_review_text(text):
    """Clean review text for analysis."""
    # Remove HTML entities
    text = re.sub(r'&\w+;', ' ', text)
    # Remove excessive whitespace
    text = re.sub(r'\s+', ' ', text).strip()
    # Remove very short reviews (likely not useful)
    if len(text) < 20:
        return None
    return text
def extract_key_phrases(reviews, min_frequency=3):
    """Extract frequently mentioned phrases from reviews."""
    from collections import Counter
    import re
    words = []
    for review in reviews:
        if review.text:
            # Simple bigram extraction
            tokens = re.findall(r'\b\w+\b', review.text.lower())
            for i in range(len(tokens) - 1):
                bigram = f"{tokens[i]} {tokens[i+1]}"
                words.append(bigram)
    return Counter(words).most_common(50)
def aggregate_sentiment(reviews):
    """Calculate aggregate sentiment metrics."""
    if not reviews:
        return {}
    ratings = [r.rating for r in reviews if r.rating]
    return {
        "total_reviews": len(reviews),
        "avg_rating": sum(ratings) / len(ratings) if ratings else 0,
        "rating_distribution": {
            str(i): len([r for r in reviews if r.rating == i])
            for i in range(1, 6)
        },
        "verified_pct": (
            len([r for r in reviews if r.verified]) / len(reviews) * 100
            if reviews else 0
        ),
    }

扩大至百万次审查

当您的目标列表增长到成千上万的产品 跨越多个平台, 建筑很重要。

基于队列的架构

  • 使用消息队列(Redis, RabbitMQ)来管理产品列表,并在工人之间分配工作.
  • 每个工人一次处理一种产品:通过所有审查粘贴,存储结果,转移到下一个产品.
  • 每个平台分别排队,以遵守不同的速率限制.

储存战略

  • 当解析器改变时,将原始 HTML 存储在对象存储(S3)中进行后处理.
  • 在 PostgreSQL 中存储解析评论,并进行全文搜索以进行分析.
  • 使用基于审查ID或散列的调值来避免重复存储在重排上.

递增划线

对于持续的监控,你不需要每次重复所有的评论. 按最近的情况排序, 并在您已经收集到的评论中停止 。 这极大地减少了代理用户,加速了收藏.

密钥外卖:先按最新评论排序, 并当您击中以前收集的内容时停止刮刮 。 这使得一个完全的重击变成了递增更新.

最佳做法

  • 使用粘贴会话进行页标 : 为单一产品保持相同的IP跨审查页面,以避免触发反机器人检测.
  • 尊重率限制 : 2-5秒的两页间延迟,产品间延迟的时间更长. 不同的平台具有不同的容忍度.
  • 处理空页 : 空审查页表示您已到达尾端 。 不要继续尝试更多的页面。
  • 验证数据质量: 检查 CAPTCHA 的页面, 空的内容, 并重复您的管道中的审查 。
  • 使用 住宅代办编号 : 对亚马逊和其他高度保护的平台至关重要。
  • 渐进存储 : 处理和存储评论 当您刮它们, 而不是一个批次在结尾。

关键外卖

  • 审查数据提供了其他数据来源无法提供的独特竞争情报。
  • 不同的平台需要不同的刮刮策略——每个平台建造模块式刮刮器.
  • 使用粘贴会话进行审查,在产品之间进行按要求旋转。
  • 先按最新数据排序,然后在以前收集的审查中止步,以便有效地逐步报废。
  • 情绪分析的预处理审查文本:干净,去重复,并提取关键短语.
  • 使用 代理哈特的住宅代理 并设定地理目标,以便可靠地访问所有平台的审查网页。

准备好开始收集审查数据了吗? 看我们的 亚马逊刮纸指南 具体平台和我们 电子商务数据报废指南 为整个战略。 检查 使用 Python 中的代理 和 使用节点中的代理 执行模式。

准备开始了吗？

通过AI过滤访问148多个国家的5000多万个住宅IP。

查看价格住宅代理
← 返回博客