高效爬取沃尔玛海量商品信息与AI分析实战

Pythonimport scrapy
from scrapy_selenium import SeleniumRequest
from ipidea_proxy import IPIDEAProxy

class WalmartSpider(scrapy.Spider):
name = 'walmart_product'

def __init__(self):
self.proxy_manager = IPIDEAProxy(
api_key="YOUR_API_KEY",
protocol="https",
country="us",
city_level=True # 精确到城市级IP
)

def start_requests(self):
sku_list = self.load_target_skus() # 从文件加载SKU列表

for sku in sku_list:
url = f"https://www.walmart.com/ip/{sku}"
proxy = self.proxy_manager.get_proxy()

yield SeleniumRequest(
url=url,
callback=self.parse_product,
meta={
'proxy': proxy['https'],
'sku': sku,
'proxy_info': proxy
},
wait_time=3,
screenshot=True # 用于调试页面渲染
)

Pythondef parse_product(self, response):
item = {}

# 基础商品信息（静态DOM）
item['title'] = response.css('h1[itemprop="name"]::text').get()
item['brand'] = response.css('[data-testid="brand"]::text').get()
item['current_price'] = self.extract_price(
response.css('[itemprop="price"]::attr(content)').get()
)

# SKU和标识码
item['sku'] = response.meta['sku']
item['upc'] = response.xpath('//span[contains(text(),"UPC")]/following-sibling::span/text()').get()
item['gtin'] = response.css('[data-testid="gtin"]::text').get()

# 动态加载数据（需执行JavaScript）
driver = response.meta['driver']
item['rating'] = driver.execute_script(
"return window.__WML_REDUX_INITIAL_STATE__?.product?.reviewSummary?.averageRating"
)
item['review_count'] = driver.execute_script(
"return window.__WML_REDUX_INITIAL_STATE__?.product?.reviewSummary?.totalReviewCount"
)

# 规格信息提取
specs = {}
spec_elements = response.css('[data-testid="product-specification"] li')
for spec in spec_elements:
key = spec.css('span:first-child::text').get()
value = spec.css('span:last-child::text').get()
if key and value:
specs[key.strip()] = value.strip()
item['specifications'] = specs

# 图片URL提取
item['image_urls'] = response.css(
'[data-testid="media-viewer-thumbnail"] img::attr(src)'
).getall()

# 库存状态
item['in_stock'] = 'Out of stock' not in response.css(
'[data-testid="sold-out-badge"]::text'
).get(default='')

# 30+字段完整采集（部分示例）
item['seller_name'] = response.css('[data-testid="seller-name"]::text').get()
item['shipping_price'] = self.extract_shipping_info(response)
item['estimated_delivery'] = response.css(
'[data-testid="delivery-message"]::text'
).get()
item['category_path'] = self.extract_breadcrumb(response)
item['description_html'] = response.css(
'[data-testid="product-description"]'
).get()
item['bullets'] = response.css(
'[data-testid="about-item"] li::text'
).getall()

yield item

SQL-- PostgreSQL商品表结构
CREATE TABLE walmart_products (
id SERIAL PRIMARY KEY,
sku VARCHAR(50) UNIQUE NOT NULL,
title TEXT,
brand VARCHAR(200),
current_price DECIMAL(10,2),
original_price DECIMAL(10,2),
currency CHAR(3) DEFAULT 'USD',
upc VARCHAR(20),
gtin VARCHAR(20),
gtin_type VARCHAR(10),
rating DECIMAL(3,2),
review_count INTEGER,
in_stock BOOLEAN,
seller_name VARCHAR(200),
category_path JSONB,
specifications JSONB,
image_urls TEXT[],
description_text TEXT,
bullets TEXT[],
crawled_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
data_quality VARCHAR(20)
);

-- 创建查询索引
CREATE INDEX idx_category ON walmart_products USING GIN(category_path);
CREATE INDEX idx_price ON walmart_products(current_price);
CREATE INDEX idx_brand ON walmart_products(brand);

Pythonfrom prophet import Prophet
import pandas as pd

def analyze_price_trends(product_data):
# 时间序列预测
df = pd.DataFrame(product_data)
df['ds'] = pd.to_datetime(df['crawled_at'])
df['y'] = df['current_price']

model = Prophet(
changepoint_prior_scale=0.05,
seasonality_mode='multiplicative'
)
model.fit(df)

# 预测未来30天价格走势
future = model.make_future_dataframe(periods=30)
forecast = model.predict(future)

# 识别价格异常波动
df['price_change'] = df['y'].pct_change()
anomalies = df[abs(df['price_change']) > 0.15] # 超过15%的波动

return {
'forecast': forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].tail(30).to_dict('records'),
'anomalies': anomalies.to_dict('records'),
'seasonal_patterns': model.seasonalities
}