前言
做电商运营、竞品跟踪、选品分析，很多人都遇到过这样的困境：
想要持续跟进多款商品的价格变动、库存余量、规格变化、店铺信息，尤其是大促期间，价格和优惠瞬息万变。如果依靠人工打开网页逐个复制记录，不仅每天耗费大量时间，还极易错过降价、库存清空等关键信号。
如果自己从零开发页面解析脚本，坑点会非常多：页面 JS 动态渲染、频繁访问触发平台风控封禁 IP、网页改版后原有解析逻辑直接失效；多规格款式、图片、店铺信息都需要单独编写解析逻辑，维护成本居高不下。
最近在实际项目中体验了 Open Claw 这套工具，很好地解决了以上痛点。不用处理复杂的网页解析、反访问限制等底层琐事，仅编写少量代码，就可以拿到标准化的商品全量信息，实现定时巡检、异动捕捉、历史数据存储，完成基础数据分析工作。
本文为完整实操教程，所有代码可以复制修改后直接运行，适合个人调研和中小团队竞品跟踪场景。
Open Claw 能解决什么问题
拿到完整标准化商品资料：商品标题、原价、实时售价、库存、多规格 SKU 信息、品牌、店铺名称、商品图、发货地等信息直接输出，不需要自己解析网页源码。
支持实时 / 缓存两种模式：做监控时开启实时模式，拿到最新价格库存；日常批量查询开启缓存，提升整体速度，减少资源消耗。
不受前端页面改版影响：网页样式更新不会打乱我们获取数据的逻辑，大幅降低脚本维护工作量。
方便做二次分析：输出结构化数据，可以直接存入本地数据库，做价格趋势、库存波动、竞品对标分析。
提示：使用前需要在后台获取两组身份凭证，用于身份校验。
整体实现思路
整个小系统拆分为 4 个简单模块：
数据获取模块：输入商品编号，获取商品结构化信息。
数据清洗模块：过滤无效字段，提取业务关心的核心字段。
定时监控模块：循环定时巡检商品，对比前后数据，捕捉价格、库存变动。
存储与分析模块：把每次采集快照存入本地数据库，后续做历史查询、简单统计分析。
环境准备
只需要基础 Python 环境，安装网络请求库即可，无需复杂框架。
完整实战代码
新建文件goods_monitor.py，复制下面全部代码，修改配置区凭证、需要监控的商品编号、巡检间隔，即可直接运行。
import requests
import json
import sqlite3
import time
from datetime import datetime
# ===================== 配置区域，请自行修改 =====================
CLAW_KEY = "填入你的凭证key"
CLAW_SECRET = "填入你的凭证secret"
# 需要监控的商品编号，支持多个商品批量监控
MONITOR_GOODS_ID = ["10335871600", "10057467958584"]
POLL_INTERVAL = 300 # 巡检间隔，单位秒，建议不要设置过小
PRICE_CHANGE_THRESHOLD = 0.1 # 价格变动超过10%输出告警
DB_NAME = "goods_monitor.db"
# =================================================================
def init_database():
"""初始化本地数据库，存储商品历史快照"""
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS goods_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
goods_id TEXT NOT NULL,
title TEXT,
brand TEXT,
current_price REAL,
origin_price REAL,
stock INTEGER,
shop_name TEXT,
capture_time TIMESTAMP
)
''')
conn.commit()
conn.close()
def get_goods_detail(goods_id: str, real_time: bool = False):
"""
获取商品详情
:param goods_id:商品编号
:param real_time: True获取实时最新数据，False使用缓存提升速度
:return:清洗后的商品字典，失败返回None
"""
params = {
"key": CLAW_KEY,
"secret": CLAW_SECRET,
"num_iid": goods_id,
"cache": "no" if real_time else "yes",
"result_type": "json",
"lang": "cn"
}
retry_count = 3
for attempt in range(retry_count):
try:
session = requests.Session()
resp = session.get(
"https://api-gw.onebound.cn/jd/item_get_pro",
params=params,
timeout=15
)
resp.raise_for_status()
res_json = resp.json()
item_data = res_json.get("item")
if not item_data:
print(f"商品 {goods_id} 未获取到有效数据")
return None
# 做简单字段清洗
clean_data = {
"goods_id": item_data.get("num_iid"),
"title": item_data.get("title", ""),
"brand": item_data.get("brand", ""),
"current_price": float(item_data.get("price", 0)) if item_data.get("price") else 0,
"origin_price": float(item_data.get("orginal_price", 0)) if item_data.get("orginal_price") else 0,
"stock": int(item_data.get("num", 0)) if item_data.get("num") else 0,
"shop_name": item_data.get("seller_info", {}).get("shop_name", "")
}
return clean_data
except Exception as e:
print(f"第{attempt+1}次请求异常：{str(e)}")
time.sleep(2)
return None
def save_to_db(goods_data):
"""将采集到的数据写入本地数据库"""
if not goods_data:
return
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute('''
INSERT INTO goods_history(goods_id,title,brand,current_price,origin_price,stock,shop_name,capture_time)
VALUES (?,?,?,?,?,?,?,?)
''', (
goods_data["goods_id"],
goods_data["title"],
goods_data["brand"],
goods_data["current_price"],
goods_data["origin_price"],
goods_data["stock"],
goods_data["shop_name"],
datetime.now().strftime("%Y-%m-%d %H:%M:%S")
))
conn.commit()
conn.close()
def monitor_task():
"""监控主逻辑，对比历史价格，输出变动告警"""
last_status = {}
while True:
print(f"\n===== 巡检开始 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} =====")
for gid in MONITOR_GOODS_ID:
current_info = get_goods_detail(gid, real_time=True)
if not current_info:
continue
save_to_db(current_info)
print(f"商品:{current_info['title']}")
print(f"现价:{current_info['current_price']},原价:{current_info['origin_price']},库存:{current_info['stock']}")
# 对比上一次状态，判断是否触发告警
if gid in last_status:
old_price = last_status[gid]["current_price"]
new_price = current_info["current_price"]
if old_price > 0 and abs(new_price - old_price)/old_price >= PRICE_CHANGE_THRESHOLD:
print(f"⚠️【价格异动提醒】商品{gid}，价格由{old_price}变动至{new_price}")
last_status[gid] = current_info
print(f"===== 本轮巡检结束，等待 {POLL_INTERVAL} 秒后下一轮 =====")
time.sleep(POLL_INTERVAL)
def analysis_demo():
"""简单数据分析示例：读取数据库，打印某商品的价格历史"""
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
target_goods = MONITOR_GOODS_ID[0]
cursor.execute('''
SELECT capture_time,current_price,stock FROM goods_history WHERE goods_id=? ORDER BY capture_time
''', (target_goods,))
rows = cursor.fetchall()
print(f"\n=====商品 {target_goods} 历史价格库存记录 =====")
for row in rows:
print(f"时间:{row[0]} | 价格:{row[1]} | 库存:{row[2]}")
conn.close()
if __name__ == "__main__":
init_database()
print("初始化完成，启动监控任务，Ctrl+C终止程序")
try:
monitor_task()
except KeyboardInterrupt:
print("\n程序手动终止，运行analysis_demo()可以查看历史数据分析结果")
analysis_demo()代码说明
初始化数据库：本地 SQLite 文件数据库，无需额外安装数据库服务，每次采集会写入一条快照记录。
数据获取函数：内置 3 次失败重试，网络抖动的时候提高稳定性；支持切换实时 / 缓存模式。
数据清洗：只保留业务上最核心字段，过滤大量无关内容。
监控主循环：定时循环遍历监控列表，对比前后价格，变动幅度超过阈值直接控制台输出告警。
简易分析示例：程序终止之后，可以调用analysis_demo()读取本地存储的历史记录，输出价格、库存随时间变化记录。
如何落地扩展
1、告警能力升级
现在告警只是打印控制台，实际项目可以扩展为：
2、多规格 SKU 监控
原始返回内容中包含完整 SKU 数组，每一个子规格拥有独立编号、价格、库存。可以修改解析逻辑，单独跟踪每一个规格的价格与库存，适合服装、多型号数码产品。
3、导出报表
可以结合 pandas 库，直接读取 sqlite 数据库，导出 csv、Excel 报表，用于运营周报，做竞品对标分析。
4、部署后台持续运行
Linux 服务器可以使用 nohup、systemd 把脚本放在后台长期运行，实现 7×24 小时不间断监控。Windows 可以使用任务计划程序定时启动。
注意事项
巡检间隔不要设置过小，过高频率不仅消耗配额，也没有实际业务意义。普通商品建议 3‑10 分钟巡检一次。
身份凭证注意保管，不要随代码上传公开代码仓库。
原始返回中会携带大量字段，开发时可以根据业务需求，按需取用，不用全部保存入库。
总结
很多做电商、做调研的同学，要么人工手动记录，要么花大量时间维护网页抓取脚本。Open Claw 把底层复杂的网页处理做了封装，开发者只需要聚焦业务逻辑：监控哪些商品、什么条件触发提醒、如何分析历史数据。
