为什么零售情报团队需要抓取沃尔玛
沃尔玛(Walmart.com)是美国最大的电商平台之一,月活超过 4 亿,SKU 数以亿计。对 CPG 品牌、渠道管理团队和定价分析师来说,Walmart 上的 商品价格、库存状态、卖家信息和评分 是日常决策的核心数据源。但 Walmart 的反爬体系——Akamai Bot Manager + PerimeterX(现 Imperva)——让传统数据中心代理几乎无法工作。
这篇文章从实战角度出发,先讲清楚 Walmart 页面结构,再解释为什么住宅代理是刚需,然后手把手演示如何通过 __NEXT_DATA__ 提取结构化数据,最后给出限速调度方案。
Walmart 商品目录结构
Walmart 的核心页面类型有三种,每种对应不同的数据获取策略:
1. 商品详情页(Item Page)
URL 格式:
https://www.walmart.com/ip/{slug}/{itemId}
例如 https://www.walmart.com/ip/Apple-AirPods-Pro/204065916。itemId 是唯一标识,slug 变化不影响路由。一个 itemId 可以对应多个 slug(SEO 重定向),但 itemId 不变。
关键数据点:价格、库存(availabilityStatus)、卖家信息、评分、变体(size/color)、配送选项。
2. 分类页(Category / Shelf Page)
URL 格式:
https://www.walmart.com/cp/{category-slug}/{categoryId}
https://www.walmart.com/browse/{top-category}/{sub-category}/{categoryId}
分类页返回商品列表,每页约 40 个商品。适合批量发现 itemId,但分页深度通常被限制在 25 页以内。
3. 搜索结果页(Search Page)
URL 格式:
https://www.walmart.com/search?q={keyword}&sort={sort}&page={page}
搜索页同样每页约 40 条结果,支持排序(price_low, price_high, best_seller, rating)。搜索页的数据也可以从 __NEXT_DATA__ 中批量提取。
Akamai + PerimeterX 反爬体系
Walmart 使用 Akamai Bot Manager 作为第一道防线,配合 PerimeterX(现 Imperva Advanced Bot Protection)做行为验证。两者的协作方式:
- Akamai:在 TLS 握手和 HTTP/2 指纹层面识别自动化请求(JA3/JA4 指纹、HTTP/2 帧时序)。数据中心 IP 段直接被 Akamai 的
sensor_data校验拦截,返回 403 或跳转验证页。 - PerimeterX:在浏览器端注入 JavaScript 挑战(
_pxcookie),检测 WebDriver 特征、Canvas 指纹、鼠标轨迹。触发时弹出 CAPTCHA 或返回空页面。
核心结论:数据中心代理的 IP 段已被 Akamai 标记,请求成功率通常低于 5%。住宅代理(Residential Proxy)使用真实 ISP 分配的 IP,通过 Akamai 的 IP 信誉检查,成功率可达 90%+。对于搜索页和分类页等高保护页面,移动代理(Mobile Proxy)效果更佳——Walmart 移动端反爬更宽松。
代理类型对比
| 代理类型 | 商品详情页成功率 | 搜索/分类页成功率 | 平均延迟 | 适用场景 |
|---|---|---|---|---|
| 数据中心代理 | < 5% | < 1% | 低(~100ms) | 不推荐 |
| 住宅代理 | 85-95% | 70-85% | 中(~800ms) | 商品页、批量抓取 |
| 移动代理 | 95%+ | 90%+ | 较高(~1.2s) | 搜索页、高频监控 |
最简解析路径:__NEXT_DATA__
Walmart 前端使用 Next.js,服务端渲染时将完整的商品数据序列化为 JSON,嵌入 HTML 的 <script id="__NEXT_DATA__"> 标签中。这是 最容易、最稳定 的解析路径——不需要分析 DOM 结构,不依赖 CSS 选择器,数据已经是结构化 JSON。
提取方法只需两步:
- 请求 HTML 页面
- 用正则或 BeautifulSoup 找到
<script id="__NEXT_DATA__">,解析 JSON
关键 JSON 路径:
# 商品核心数据
__NEXT_DATA__.props.pageProps.initialData.data.product
# 价格
...product.priceInfo.currentPrice.price # 数值
...product.priceInfo.currentPrice.currencyUnit # 货币
# 库存状态
...product.availabilityStatus # "IN_STOCK" | "OUT_OF_STOCK" | "PREORDER"
# 评分与评论数
...product.averageRating
...product.numberOfReviews
# 卖家信息
...product.sellerId
...product.sellerName
...product.walmartMarketplaceOffer # 3P 标记
Python 实战:抓取 + 解析 Walmart 商品数据
以下示例使用 ProxyHat 住宅代理抓取 Walmart 商品页,提取 __NEXT_DATA__ 中的核心字段。
import requests
import json
import re
import time
from datetime import datetime
PROXY = "http://user-country-US:YOUR_PASSWORD@gate.proxyhat.com:8080"
PROXIES = {"http": PROXY, "https": PROXY}
HEADERS = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/125.0.0.0 Safari/537.36"
),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
"Accept-Encoding": "gzip, deflate, br",
}
def fetch_item(item_id: str) -> dict | None:
"""抓取 Walmart 商品页并提取 __NEXT_DATA__"""
url = f"https://www.walmart.com/ip/product/{item_id}"
try:
resp = requests.get(url, headers=HEADERS, proxies=PROXIES, timeout=15)
resp.raise_for_status()
except requests.RequestException as e:
print(f"[ERROR] 请求失败 item={item_id}: {e}")
return None
# 提取 __NEXT_DATA__ JSON
match = re.search(
r'<script id="__NEXT_DATA__"[^>]*>(.*?)</script>',
resp.text,
re.DOTALL
)
if not match:
print(f"[WARN] 未找到 __NEXT_DATA__, item={item_id}")
return None
next_data = json.loads(match.group(1))
product = (
next_data
.get("props", {})
.get("pageProps", {})
.get("initialData", {})
.get("data", {})
.get("product", {})
)
if not product:
print(f"[WARN] product 节点为空, item={item_id}")
return None
# 提取核心字段
price_info = product.get("priceInfo", {}).get("currentPrice", {})
return {
"item_id": item_id,
"name": product.get("name", ""),
"price": price_info.get("price"),
"currency": price_info.get("currencyUnit", "USD"),
"availability": product.get("availabilityStatus", ""),
"rating": product.get("averageRating"),
"review_count": product.get("numberOfReviews", 0),
"seller_id": product.get("sellerId", ""),
"seller_name": product.get("sellerName", ""),
"is_marketplace": bool(product.get("walmartMarketplaceOffer")),
"fetched_at": datetime.utcnow().isoformat(),
}
# 批量示例
if __name__ == "__main__":
items = ["204065916", "55506642", "368696568"]
for item_id in items:
result = fetch_item(item_id)
if result:
print(json.dumps(result, indent=2, ensure_ascii=False))
time.sleep(2) # 限速间隔
curl 单次测试
快速验证代理连通性和 __NEXT_DATA__ 存在性:
curl -x http://user-country-US:YOUR_PASSWORD@gate.proxyhat.com:8080 \
-H "User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" \
-H "Accept-Language: en-US,en;q=0.9" \
"https://www.walmart.com/ip/Apple-AirPods-Pro/204065916" \
| grep -o '<script id="__NEXT_DATA__"[^>]*>.*</script>' \
| sed 's/<script id="__NEXT_DATA__"[^>]*>//' \
| sed 's/<\/script>//' \
| python3 -m json.tool | head -50
Walmart Marketplace:3P vs 1P 数据区分
Walmart 上的商品分为两类:1P(Walmart 自营)和 3P(第三方卖家)。2024 年数据显示,Walmart Marketplace 的第三方卖家已超过 15 万,3P 商品占比持续上升。
如何区分
- sellerId:1P 商品的
sellerId通常为5BB2BD9B-4C56-4C5D-9F67-2D1559B0E649(Walmart 官方卖家 ID)或类似固定值。3P 商品为各卖家独立 ID。 - walmartMarketplaceOffer:此字段存在且为真,表示 3P 商品。
- offerCount:商品详情页可能展示多个卖家报价(同 itemId 不同 seller),
__NEXT_DATA__中的offers数组包含所有卖家。
3P 数据提取示例
def extract_offers(product: dict) -> list[dict]:
"""提取同一 itemId 下的所有卖家报价"""
offers_data = product.get("offers", [])
results = []
for offer in offers_data:
results.append({
"seller_id": offer.get("sellerId", ""),
"seller_name": offer.get("sellerName", ""),
"price": offer.get("priceInfo", {}).get("currentPrice", {}).get("price"),
"availability": offer.get("availabilityStatus", ""),
"shipping": offer.get("fulfillment", {}).get("shippingOptions", []),
})
return results
1P 与 3P 数据差异
| 维度 | 1P(Walmart 自营) | 3P(Marketplace) |
|---|---|---|
| 价格稳定性 | 相对稳定,变化频率低 | 波动大,可能每天多次变化 |
| 库存数据 | 较准确,区域差异小 | 依赖卖家更新,可能延迟 |
| 配送信息 | Walmart 配送,结构化程度高 | 因卖家而异,需逐个解析 |
| 抓取难度 | 标准流程即可 | 部分卖家页有额外反爬 |
| 数据量 | 约 40% SKU | 约 60% SKU 且持续增长 |
对零售情报团队而言,3P 价格监控是核心需求——因为 3P 卖家的定价直接影响品牌控价(MAP)合规性。建议对 3P 商品设置更高的抓取频率(每 4-6 小时),1P 商品可以降低到每 12-24 小时。
限速调度策略
Walmart 的速率限制并不公开,但根据实测经验:
- 单 IP 请求量:同一住宅 IP 每分钟超过 10-15 次请求后,触发 Akamai 挑战概率急剧上升。
- 搜索页:比商品详情页更严格,建议单 IP 每分钟不超过 5 次。
- 整站维度:Walmart 在 IP 级别做频次统计,短时间爆发式请求(如 1 秒内 10 次)会立即触发封禁。
推荐调度架构
import time
import random
from collections import deque
# 限速参数
RATE_LIMITS = {
"item_page": {"max_per_minute": 10, "min_interval": 6}, # 秒
"search_page": {"max_per_minute": 4, "min_interval": 15},
"category_page": {"max_per_minute": 5, "min_interval": 12},
}
class RateLimiter:
"""基于滑动窗口的限速器"""
def __init__(self, page_type: str):
self.config = RATE_LIMITS[page_type]
self.timestamps = deque()
def wait(self):
now = time.time()
# 清理超过 60 秒的记录
while self.timestamps and now - self.timestamps[0] > 60:
self.timestamps.popleft()
if len(self.timestamps) >= self.config["max_per_minute"]:
sleep_time = 60 - (now - self.timestamps[0]) + random.uniform(0.5, 2)
print(f"[RATE] 达到限速,等待 {sleep_time:.1f}s")
time.sleep(sleep_time)
# 最小间隔 + 随机抖动
time.sleep(self.config["min_interval"] + random.uniform(0, 3))
self.timestamps.append(time.time())
class ProxyRotator:
"""ProxyHat 住宅代理轮换器"""
def __init__(self, username: str, password: str):
self.base_user = username
self.password = password
self.session_counter = 0
def get_proxy(self, country: str = "US", sticky: bool = False):
"""获取代理配置,sticky=True 时保持同一 IP"""
if sticky:
session_id = f"sess{self.session_counter}"
user = f"{self.base_user}-country-{country}-session-{session_id}"
else:
user = f"{self.base_user}-country-{country}"
proxy_url = f"http://{user}:{self.password}@gate.proxyhat.com:8080"
return {"http": proxy_url, "https": proxy_url}
def rotate_session(self):
"""切换到新的 sticky session"""
self.session_counter += 1
# 使用示例
rotator = ProxyRotator("YOUR_USER", "YOUR_PASSWORD")
item_limiter = RateLimiter("item_page")
for item_id in ["204065916", "55506642", "368696568"]:
item_limiter.wait()
proxies = rotator.get_proxy(country="US", sticky=False)
result = fetch_item_with_proxy(item_id, proxies)
# ...
调度最佳实践
- 轮换 IP:每次请求使用不同住宅 IP(ProxyHat 默认行为),避免单 IP 积累请求历史。
- Sticky Session:需要连续请求同一站点的场景(如翻页),使用
session-{id}参数保持 IP 不变,每 10-15 分钟轮换一次。 - 随机抖动:在固定间隔基础上加 0-3 秒随机延迟,模拟人类行为。
- 失败退避:遇到 403/429 时,指数退避(2s → 4s → 8s → 16s),同时切换 IP。
- 时段分散:将大批量任务分散到不同时段执行,避免集中在整点。
搜索页批量抓取
搜索页的 __NEXT_DATA__ 包含一页约 40 个商品的摘要数据,适合快速发现 itemId 列表:
def extract_search_results(next_data: dict) -> list[dict]:
"""从搜索页 __NEXT_DATA__ 提取商品列表"""
search = (
next_data
.get("props", {})
.get("pageProps", {})
.get("initialData", {})
.get("searchResult", {})
.get("itemStacks", [{}])
)
items = []
for stack in search:
for item in stack.get("items", []):
items.append({
"item_id": item.get("id", ""),
"name": item.get("name", ""),
"price": item.get("price", {}),
"rating": item.get("averageRating"),
"review_count": item.get("numberOfReviews", 0),
})
return items
注意:搜索页的反爬比商品详情页更严格,建议使用 住宅或移动代理,并将频率控制在每分钟 4 次以内。
数据合规与伦理考量
- robots.txt:Walmart 的
robots.txt禁止抓取部分路径(如/cart,/account),商品页和搜索页虽未明确禁止,但大规模抓取仍需注意频率。 - 服务条款:Walmart ToS 禁止自动化数据采集,商业使用需评估法律风险。
- 数据最小化:只抓取业务必需的字段,不存储不必要的数据。
- GDPR/CCPA:如果数据涉及用户生成内容(评论),需遵守隐私法规。
关键要点总结
- Walmart 使用 Akamai + PerimeterX 双层反爬,数据中心代理成功率 < 5%,住宅代理是刚需。
__NEXT_DATA__是最稳定的解析路径——直接从 HTML 中提取 JSON,无需依赖 CSS 选择器。- 区分 1P 和 3P 商品看
walmartMarketplaceOffer和sellerId字段。- 商品详情页限速:每 IP 每分钟 ≤10 次;搜索页:≤4 次。加随机抖动。
- 使用 ProxyHat 住宅代理时,默认每次请求轮换 IP;需要保持会话时使用
session-{id}参数。
如果你正在搭建 Walmart 数据采集管线,可以从 ProxyHat 住宅代理套餐开始,结合本文的调度策略快速上线。需要覆盖特定城市或州?查看 全球代理位置列表,ProxyHat 支持美国 50 州的精准地理位置定位。






