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

如何使用代理爬取Shopify店铺：完整指南

学会如何用 JSON API 端点和住宅代理来刮刮 Shopify 存储数据. 完整的Python和Node.js代码用于提取产品,价格,和库存数据.

ProxyHat Team
如何使用代理爬取Shopify店铺：完整指南

为什么是Scrape商店?

从小型独立品牌到主要零售商,全世界超过400万家网上商店拥有商店。 这使得它成为电子商务情报最丰富的来源之一. 通过刮掉Shoppify商店,可以跟踪竞争者定价,监测产品推出,分析市场趋势,建立综合产品数据库.

好消息是Shopify有一个可预测的结构,使得刮刮比大多数电子商务平台更系统化. 每个 Shopify 商店都通过标准化的端点暴露某些数据,这意味着单个刮刮机架构可以跨数千个不同的商店工作. 关于电子商务拆散战略的更广泛的概述,见我们 电子商务数据报废指南。 。 。

理解商店的存储结构

每个 Shopify 商店都遵循相同的URL和数据模式,不管主题或定制.

公共 JSON 终点

Shopify通过JSON端点曝光不需要认证的产品数据. 这些是刮商店的最有效的方法,因为你得到的结构数据没有HTML解析.

终点数据已返回倾斜
/products.json所有具有变体、价格、图像的产品?page=N&limit=250
/products/{handle}.json单一产品细目不详
/collections.json所有收藏?page=N
/collections/{handle}/products.json收藏中的产品?page=N&limit=250
/meta.json存储元数据( 名称、 描述)不详

产品数据结构

JSON API的每个产品对象包括:

  • 基本信息 : 标题、 手柄( slug)、 body html( 描述)、 销售商、 产品 类型、 标签
  • 备选案文: 每个变体都有自己的价格,比较 价格,SKU,存货状况,以及选项值(大小,颜色等).
  • 图像 : 包含其它文本的所有产品图像的 URL
  • 日期 : 创建 at, 更新 at, 发布 at

费率限制

Shopify 应用费率限制来保护存储性能. 公共的JSON端点通常允许每IP每秒2-4个请求,然后才会节奏踢入. 这里 住宅代办 变得至关重要 —— 将请求分散到多个IP上,可以保持吞吐量,而不对任何单个IP施加速率限制.

商店化的代理配置

Shopify的速率限制基于IP,使代理旋转成为大规模刮除的主要策略.

代理服务器设置

# Rotating residential proxy (new IP per request)
http://USERNAME:PASSWORD@gate.proxyhat.com:8080
# Geo-targeted for region-specific stores
http://USERNAME-country-US:PASSWORD@gate.proxyhat.com:8080
# Sticky session for paginated scraping of one store
http://USERNAME-session-shopify001:PASSWORD@gate.proxyhat.com:8080

对于Shofify刮刮,在刮刮不同的商店时使用按要求旋转,在通过单一商店的产品目录开始时使用粘性会话. 这种模式模仿自然浏览行为.

Python 执行

这里有一个生产准备的Shopify刮刀 代理哈特的 Python SDK。 。 。

JSON API 拼写

import requests
import json
import time
import random
from dataclasses import dataclass, field
from typing import Optional
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 ShopifyProduct:
    id: int
    title: str
    handle: str
    vendor: str
    product_type: str
    tags: list[str]
    variants: list[dict]
    images: list[str]
    min_price: float
    max_price: float
    created_at: str
    updated_at: str
def get_session(store_domain: str) -> requests.Session:
    """Create a session with proxy and headers configured."""
    session = requests.Session()
    session.proxies = {"http": PROXY_URL, "https": PROXY_URL}
    session.headers.update({
        "User-Agent": random.choice(USER_AGENTS),
        "Accept": "application/json",
        "Accept-Language": "en-US,en;q=0.9",
    })
    return session
def scrape_all_products(store_domain: str) -> list[ShopifyProduct]:
    """Scrape all products from a Shopify store via JSON API."""
    products = []
    page = 1
    session = get_session(store_domain)
    while True:
        url = f"https://{store_domain}/products.json?page={page}&limit=250"
        try:
            response = session.get(url, timeout=30)
            response.raise_for_status()
        except requests.RequestException as e:
            print(f"Error on page {page}: {e}")
            break
        data = response.json()
        page_products = data.get("products", [])
        if not page_products:
            break
        for p in page_products:
            prices = [float(v["price"]) for v in p.get("variants", [])
                      if v.get("price")]
            product = ShopifyProduct(
                id=p["id"],
                title=p["title"],
                handle=p["handle"],
                vendor=p.get("vendor", ""),
                product_type=p.get("product_type", ""),
                tags=p.get("tags", "").split(", ") if p.get("tags") else [],
                variants=[{
                    "id": v["id"],
                    "title": v["title"],
                    "price": v["price"],
                    "compare_at_price": v.get("compare_at_price"),
                    "sku": v.get("sku"),
                    "available": v.get("available", False),
                } for v in p.get("variants", [])],
                images=[img["src"] for img in p.get("images", [])],
                min_price=min(prices) if prices else 0,
                max_price=max(prices) if prices else 0,
                created_at=p.get("created_at", ""),
                updated_at=p.get("updated_at", ""),
            )
            products.append(product)
        print(f"Page {page}: {len(page_products)} products (total: {len(products)})")
        page += 1
        time.sleep(random.uniform(1, 3))
    return products
def scrape_collections(store_domain: str) -> list[dict]:
    """Scrape all collections from a Shopify store."""
    collections = []
    page = 1
    session = get_session(store_domain)
    while True:
        url = f"https://{store_domain}/collections.json?page={page}"
        try:
            response = session.get(url, timeout=30)
            response.raise_for_status()
        except requests.RequestException:
            break
        data = response.json()
        page_collections = data.get("collections", [])
        if not page_collections:
            break
        collections.extend(page_collections)
        page += 1
        time.sleep(random.uniform(1, 2))
    return collections
# Example: Scrape multiple Shopify stores
if __name__ == "__main__":
    stores = [
        "example-store-1.myshopify.com",
        "example-store-2.com",
        "example-store-3.com",
    ]
    for store in stores:
        print(f"\nScraping: {store}")
        products = scrape_all_products(store)
        print(f"Found {len(products)} products")
        # Save to JSON
        with open(f"{store.replace('.', '_')}_products.json", "w") as f:
            json.dump([vars(p) for p in products], f, indent=2)
        time.sleep(random.uniform(3, 7))

监测各仓库的价格变化

def compare_prices(store_domain: str, previous_data: dict) -> list[dict]:
    """Compare current prices with previously stored data."""
    changes = []
    products = scrape_all_products(store_domain)
    for product in products:
        prev = previous_data.get(product.handle)
        if not prev:
            changes.append({
                "type": "new_product",
                "handle": product.handle,
                "title": product.title,
                "price": product.min_price,
            })
            continue
        if product.min_price != prev.get("min_price"):
            changes.append({
                "type": "price_change",
                "handle": product.handle,
                "title": product.title,
                "old_price": prev["min_price"],
                "new_price": product.min_price,
                "change_pct": ((product.min_price - prev["min_price"])
                               / prev["min_price"] * 100)
                              if prev["min_price"] else 0,
            })
    return changes

节点.js 执行

使用节点.js版本 代理哈特节点 SDK。 。 。

const axios = require("axios");
const { HttpsProxyAgent } = require("https-proxy-agent");
const fs = require("fs");
const PROXY_URL = "http://USERNAME:PASSWORD@gate.proxyhat.com:8080";
const agent = new HttpsProxyAgent(PROXY_URL);
async function scrapeShopifyProducts(storeDomain) {
  const products = [];
  let page = 1;
  while (true) {
    const url = `https://${storeDomain}/products.json?page=${page}&limit=250`;
    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: "application/json",
        },
        timeout: 30000,
      });
      const pageProducts = data.products || [];
      if (pageProducts.length === 0) break;
      for (const p of pageProducts) {
        const prices = p.variants.map((v) => parseFloat(v.price)).filter(Boolean);
        products.push({
          id: p.id,
          title: p.title,
          handle: p.handle,
          vendor: p.vendor,
          productType: p.product_type,
          tags: p.tags ? p.tags.split(", ") : [],
          minPrice: Math.min(...prices),
          maxPrice: Math.max(...prices),
          variants: p.variants.map((v) => ({
            id: v.id,
            title: v.title,
            price: v.price,
            compareAtPrice: v.compare_at_price,
            sku: v.sku,
            available: v.available,
          })),
          images: p.images.map((img) => img.src),
          updatedAt: p.updated_at,
        });
      }
      console.log(`Page ${page}: ${pageProducts.length} products (total: ${products.length})`);
      page++;
      // Random delay 1-3 seconds
      await new Promise((r) => setTimeout(r, 1000 + Math.random() * 2000));
    } catch (err) {
      console.error(`Error on page ${page}: ${err.message}`);
      break;
    }
  }
  return products;
}
async function scrapeMultipleStores(stores) {
  const results = {};
  for (const store of stores) {
    console.log(`\nScraping: ${store}`);
    const products = await scrapeShopifyProducts(store);
    results[store] = products;
    console.log(`Found ${products.length} products`);
    // Delay between stores
    await new Promise((r) => setTimeout(r, 3000 + Math.random() * 4000));
  }
  return results;
}
// Usage
scrapeMultipleStores([
  "example-store-1.myshopify.com",
  "example-store-2.com",
]).then((results) => {
  fs.writeFileSync("shopify_data.json", JSON.stringify(results, null, 2));
  console.log("Data saved to shopify_data.json");
});

切换策略

发现商店

在刮刮前,您需要识别在Shoppify上运行的哪些竞争者网站. 共同指标包括:

  • 这个 /products.json 结束点返回有效的 JSON
  • HTML 来源包含 Shopify.theme 或 cdn.shopify.com
  • 这个 x-shopify-stage 回复中包含标题

处理密码存储

一些商店需要密码才能访问. 这些通常是发射前或批发商店。 JSON端点会返回一个方向到密码页. 除非你已经授权进入这些仓库

处理自定义域

商店通常使用自定义域名而不是 .myshopify.com。JSON API在自定义域上同样发挥作用。 在你的请求中使用商店的公关域名

库存追踪

产品变体包括: available 字段表示股票状况。 通过长期跟踪这个领域,可以监测竞争对手的库存水平,并确定产品何时出库——用于定价和再分配决策的有用情报.

避免区块和利率限制

虽然Shorpify比Amazon更方便刮刮,但它仍然执行保护.

保护细节缓解
IP 费率限制JSON 端点每个IP ~ 2-4 req/sec在请求中旋转住宅代理
云浮保护有些店铺用Cloudflare有类似浏览器头的住宅IP
瓶检测监测行为模式随机延迟和用户代理
密码页已锁定的发射前/批发仓库跳过或使用授权访问

有关处理反机器人系统的更多信息,请读我们的指南 如何在不被封锁的情况下刮掉网站。 。 。

密钥取走: Shopify的JSON API是最有效的刮刮方法——它给了你结构化的数据,没有HTML解析. 在返回到 HTML 刮切之前使用它.

数据使用案例

一旦你收集了Shopify产品数据,这里就是最有价值的应用程序:

  • 竞争性定价: 跟踪各类产品竞争对手价格,实时调整定价策略.
  • 产品研究: 通过监测多个商店,查明趋势产品、新发射和市场差距。
  • 市场分析: 数百家商店的汇总数据,以了解市场趋势、定价分配和类别增长。
  • 目录浓缩: 使用竞争对手的产品描述,图像,规格来改进自己的上市.
  • 品牌监控 : 追踪未经授权的销售者 并监控整个店面的MAP合规情况

关键外卖

  • 店铺 /products.json 端点是最有效的刮刮方法——在HTML解析前使用.
  • 由于标准化的结构,一个单一的刮刮机建筑在所有Shoppify商店中工作。
  • 旋转的住宅代用品克服了 Shopify基于IP的费率限制.
  • 在通过单商店目录时使用粘度会话.
  • 跟踪可变价格和综合竞争情报的可用性。
  • 开始 代理哈特的住宅代理 以覆盖您的 Shopify 刷新可靠。

准备好开始刮店铺了吗? 探索我们 电子商务数据报废指南 全面战略,检查我们 Python 代理指南 和 节点.js 代理指南 详细执行情况。 拜访我们 定价页面 开始吧

准备开始了吗？

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

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