如何爬取采用GraphQL的网站产品数据 重写代码获取沃尔玛商品页JSON

核心逻辑说明 #

当前沃尔玛全站已切换为GraphQL接口架构，原有REST参数式的搜索、商品查询接口已全部下线，无论获取搜索结果还是商品详情数据，都需要按照GraphQL规范构造请求：

• 统一请求接口地址为 https://www.walmart.com/orchestra/home/graphql
• 商品详情查询的操作名为 GetProduct，仅需传入商品ID即可返回全量商品数据
• 搜索查询的操作名为 SearchPageProducts，可传入关键词、分页数量、偏移量、门店ID等参数

完整实现代码 #

import requests
import re
import time

# 通用请求头，必须携带否则会被拦截
COMMON_HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
    "Accept": "application/json",
    "Content-Type": "application/json"
}
# 商品详情GraphQL查询语句，固定结构，仅需替换productId即可
PRODUCT_QUERY = """
query GetProduct($productId: String!) {
  product(id: $productId) {
    id
    name
    price
    description
    brand
    availabilityStatus
    images {
      url
    }
    rating {
      averageRating
      reviewCount
    }
  }
}
"""
# 搜索结果GraphQL查询语句，支持传入搜索关键词、分页、门店等参数
SEARCH_QUERY = """
query SearchPageProducts($query: String!, $offset: Int!, $count: Int!, $storeId: String!) {
  search(query: $query, offset: $offset, count: $count, storeId: $storeId) {
    products {
      id
      name
      price
      productUrl
    }
    totalCount
  }
}
"""

def request_with_retry(url, method="get", **kwargs):
    # 重试逻辑，替代原嵌套try结构，更易维护
    retry_intervals = [10, 60, 360]
    for interval in retry_intervals:
        try:
            response = requests.request(method, url, headers=COMMON_HEADERS, timeout=30, **kwargs)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"请求失败，等待{interval}秒后重试: {str(e)}")
            time.sleep(interval)
    # 最后一次重试不捕获异常，失败直接抛出
    return requests.request(method, url, headers=COMMON_HEADERS, timeout=30, **kwargs).json()

# 1. 获取商品详情示例
def get_product_detail(product_url):
    # 从商品URL中提取商品ID，示例URL: https://www.walmart.com/ip/SKIPPY-Natural-Creamy-Peanut-Butter-Spread-15-oz/37447671
    product_id = re.search(r'/ip/[^/]+/(\d+)', product_url).group(1)
    # 构造GraphQL请求体
    payload = {
        "operationName": "GetProduct",
        "variables": {
            "productId": product_id
        },
        "query": PRODUCT_QUERY
    }
    return request_with_retry("https://www.walmart.com/orchestra/home/graphql", method="post", json=payload)

# 2. 适配原有搜索功能示例
def search_products(keyword, store_id, ps=24, offset=0):
    payload = {
        "operationName": "SearchPageProducts",
        "variables": {
            "query": keyword,
            "offset": offset,
            "count": ps,
            "storeId": store_id
        },
        "query": SEARCH_QUERY
    }
    return request_with_retry("https://www.walmart.com/orchestra/home/graphql", method="post", json=payload)

# 测试调用
if __name__ == "__main__":
    # 测试获取商品详情
    test_product_url = "https://www.walmart.com/ip/SKIPPY-Natural-Creamy-Peanut-Butter-Spread-15-oz/37447671"
    product_data = get_product_detail(test_product_url)
    print(product_data)
    # 测试搜索功能
    # test_store_id = "1234" # 替换为实际的门店ID
    # search_data = search_products("book", test_store_id)
    # print(search_data)

注意事项 #

• 如果你需要返回更多字段，可以直接修改查询语句里的字段列表，和前端页面请求的字段保持一致即可，不会触发反爬
• 不要短时间内发起大量高频请求，可适当调整重试间隔或者增加随机延迟降低被封概率
• 部分区域访问沃尔玛需要代理支持，可在requests请求中添加proxies参数适配

内容的提问来源于stack exchange，提问作者e.iluf