适用场景：竞品跟踪、价格盯盘、库存状态巡检、历史价格统计分析；语言 Python3，适合开发、数据分析从业者阅读，可直接复制代码运行调试。
前言
做电商相关数据分析的时候，最头疼的就是商品页面的数据获取。直接解析网页会遇到动态渲染、反访问、页面改版、规格页面逻辑复杂等一堆麻烦，每次页面改版，之前写好的解析脚本就要全部重写，大量时间消耗在页面适配上面，真正留给业务分析的时间反而很少。
很多同学会使用浏览器自动化工具去模拟访问，但又会碰到验证码、设备指纹、访问频率限制，部署到服务器运行的时候坑非常多。
今天分享一套基于 Open‑Claw 工具的实践方案，不需要处理网页逆向、不用维护页面选择器，直接拿到标准化的商品全量信息，完成定时监控、数据落盘、异常告警、历史数据分析整套流程。
我们可以一次性拿到：商品标题、现价、原价、库存、全部规格信息、店铺信息、主图、发货地等完整内容，把精力聚焦业务逻辑，而不是和网页页面做对抗。
整体业务目标
商品全量信息快照：获取商品基础信息，包含多规格维度的数据。
持续定时巡检：周期性采集商品状态，记录每一次时间切片的数据，捕捉降价、涨价、缺货、规格变动。
本地持久化存储：将每次采集结果保存到 CSV，留存历史数据，用于后续统计分析。
异常提醒：价格低于预期阈值、库存清零，控制台输出告警。
简易数据分析：读取本地历史文件，做价格波动简单统计。
环境准备
依赖安装，执行下面命令：
完整实现代码
整体分为：封装数据获取类、字段解析、存储、告警、定时循环、数据分析几个模块。
使用前需要替换凭证信息，以及填写你需要监控的商品编号。
# coding:utf-8
import requests
import time
import csv
import os
from datetime import datetime
import pandas as pd
class GoodsMonitor:
def __init__(self, access_key, access_secret):
self.access_key = access_key
self.access_secret = access_secret
self.base_address = "https://api‑gw.onebound.cn/jd/item_get_pro/"
self.csv_file = "goods_monitor_record.csv"
self.init_csv()
def init_csv(self):
"""初始化存储文件，不存在则创建表头"""
if not os.path.exists(self.csv_file):
headers = [
"crawl_time",
"goods_id",
"goods_title",
"current_price",
"origin_price",
"stock_num",
"brand",
"shop_name",
"sku_list"
]
with open(self.csv_file, "w", encoding="utf‑8‑sig", newline="") as f:
writer = csv.writer(f)
writer.writerow(headers)
def get_raw_data(self, goods_id):
"""获取商品原始结构化数据"""
params = {
"key": self.access_key,
"secret": self.access_secret,
"num_iid": goods_id,
"cache": "no"
}
headers = {
"Accept‑Encoding": "gzip",
"Connection": "close"
}
try:
resp = requests.get(
self.base_address,
params=params,
headers=headers,
timeout=25
)
return resp.json()
except Exception as e:
print(f"网络异常: {str(e)}")
return None
def parse_raw(self, raw_data):
"""清洗解析原始返回内容，提取业务需要字段"""
if not raw_data:
return None
err_code = raw_data.get("error_code")
if err_code != "0000":
err_info = raw_data.get("error", "获取数据失败")
print(f"数据异常：{err_info}")
return None
item_info = raw_data.get("item", {})
if not item_info:
return None
sku_data = item_info.get("skus", {}).get("sku", [])
sku_text = "||".join([f'{x.get("properties_name")}:{x.get("price")}' for x in sku_data])
result = {
"crawl_time": datetime.now().strftime("%Y‑%m‑%d %H:%M:%S"),
"goods_id": item_info.get("num_iid"),
"goods_title": item_info.get("title", ""),
"current_price": float(item_info.get("price", 0)) if item_info.get("price") else 0,
"origin_price": float(item_info.get("orginal_price", 0)) if item_info.get("orginal_price") else 0,
"stock_num": int(item_info.get("num", 0)) if item_info.get("num") else 0,
"brand": item_info.get("brand", ""),
"shop_name": item_info.get("seller_info", {}).get("shop_name", ""),
"sku_list": sku_text
}
return result
def save_record(self, data):
"""写入本地csv历史记录"""
with open(self.csv_file, "a", encoding="utf‑8‑sig", newline="") as f:
writer = csv.writer(f)
row = [
data["crawl_time"],
data["goods_id"],
data["goods_title"],
data["current_price"],
data["origin_price"],
data["stock_num"],
data["brand"],
data["shop_name"],
data["sku_list"]
]
writer.writerow(row)
def check_warn(self, data, warn_price):
"""告警逻辑：价格低于阈值、库存为0触发提醒"""
if data["current_price"] <= warn_price:
print("="*70)
print(f"⚠️【价格预警】商品：{data['goods_title']}")
print(f"现价 {data['current_price']}，低于监控阈值 {warn_price}")
print("="*70)
if data["stock_num"] <= 0:
print("="*70)
print(f"🔴【库存告警】商品 {data['goods_title']} 当前缺货")
print("="*70)
def single_task(self, goods_id, warn_price):
"""单次监控任务：获取‑解析‑存储‑告警"""
raw = self.get_raw_data(goods_id)
parsed = self.parse_raw(raw)
if parsed:
self.save_record(parsed)
self.check_warn(parsed, warn_price)
print(f"✅ {parsed['crawl_time']} 完成记录：{parsed['goods_title']}")
else:
print(f"❌ 商品 {goods_id} 获取信息失败")
def run_monitor_loop(self, watch_list, interval=60*30):
"""循环持续监控
watch_list: [(商品ID,预警价格),...]
interval：循环间隔，单位秒，示例30分钟=1800
"""
print(">>>>>>启动商品循环监控<<<<<<")
while True:
for goods_id, warn_price in watch_list:
self.single_task(goods_id, warn_price)
time.sleep(3)
print(f"\n等待下一轮采集，间隔 {interval//60} 分钟\n")
time.sleep(interval)
def analysis_history(self):
"""简单历史数据分析，读取本地csv做价格统计"""
if not os.path.exists(self.csv_file):
print("暂无历史记录文件")
return
df = pd.read_csv(self.csv_file, encoding="utf‑8‑sig")
print("\n========历史数据统计分析========")
goods_group = df.groupby("goods_title")["current_price"].agg(["min","max","mean","count"])
print(goods_group)
print("\n每一件商品最低、最高、平均价格，采集总次数如上")
if __name__ == "__main__":
# =========配置区========
MY_KEY = "替换为你的凭证key"
MY_SECRET = "替换为你的凭证secret"
# 待监控列表，格式(商品编号，价格预警线)
WATCH_GOODS = [
("10335871600", 220),
("10057467958584", 600)
]
LOOP_INTERVAL = 1800 # 循环监控间隔，单位秒，这里30分钟跑一轮
# ======================
monitor = GoodsMonitor(MY_KEY, MY_SECRET)
# 1.执行循环监控，开启后会持续运行
monitor.run_monitor_loop(WATCH_GOODS, interval=LOOP_INTERVAL)
# 2.只做一次采集，测试用
# monitor.single_task("10335871600",200)
# 3.读取本地历史做数据分析
# monitor.analysis_history()代码说明
初始化文件：程序第一次运行自动生成goods_monitor_record.csv，所有采集快照全部保存在这里，Excel 可以直接打开查看。
数据获取与解析：工具返回已经结构化的商品信息，我们只需要从中筛选业务需要字段，不需要写 Xpath、CSS 选择器，页面改版不会影响业务代码。同时做异常捕获，单次获取失败不会让整个程序直接崩溃退出。
告警机制：当商品价格低于设置阈值，或者库存为 0，控制台打印提醒，可以在此基础上拓展推送能力，比如对接企业微信、邮件消息推送。
循环监控：run_monitor_loop会按照设定的间隔周期循环巡检多件商品，服务器后台运行就可以实现 7×24 小时盯盘。
数据分析模块：analysis_history读取本地历史记录，使用 pandas 统计每个商品历史最低价、最高价、平均价格，方便做价格趋势研判。
⚠️注意：不要把间隔设置过小，频繁请求会触发访问限制，业务场景建议至少 10‑30 分钟以上采集一轮。
拓展方向
1. 存储升级：替换 CSV 为数据库
如果监控商品数量很多，CSV 读写效率会下降，可以把存储逻辑修改为 SQLite / MySQL，增加批量分析、筛选查询能力。示例写入 SQL 片段：
2. 消息推送扩展
现在告警只是控制台打印，可以接入企业微信机器人、邮件、钉钉，一旦触发价格 / 库存变化，直接推送消息到手机，不用一直盯着控制台。
3. 可视化
基于 pandas + matplotlib 读取历史记录，绘制商品价格变化曲线图，直观看历史价格波动。
4. 多平台扩展
这套思路可以迁移到其他电商品类，只需要调整解析字段，监控逻辑、存储、告警代码完全复用。
踩坑总结
价格字段说明：返回的基础价格为主商品价格，如果要看各个规格真实价格，需要读取 sku_list 字段，不同规格会存在价格差异。
缓存控制：代码中设置cache=no，目的是拿到最新的实时快照；测试调试的时候可以打开缓存，减少消耗。
异常处理：商品下架、编号错误、网络抖动，程序会捕获错误并打印日志，不会直接退出循环监控任务。
服务器部署：Linux 后台运行可以使用nohup python monitor.py &方式，注意配置好 Python 环境与依赖包。
小结
借助 Open‑Claw 工具，我们绕开了复杂网页解析工作，把重心放在业务监控逻辑上。几十行代码就完成商品定时监控、历史数据留存、异常告警、数据分析整套工程，不用反复适配页面结构，极大降低维护成本。
对于个人爱好者、小团队做竞品盯盘、价格跟踪、库存巡检，这套方案足够满足大部分场景。在此基础上，还可以继续拓展报表输出、消息推送、多商品批量管理等能力。
