告别人工盯品：借助 OpenClaw 搭建商品自动监控与数据分析系统（完整实操）

前言

做电商运营，免不了要持续关注自家以及竞品商品状态：价格变动、库存变化、不同规格的售价波动、店铺基础信息更新。很多人最开始的做法就是打开网页，定时手动刷新页面复制数据，一旦要跟踪十几款商品，每天就要消耗大量人力。

如果自己写页面解析脚本，又会遇到不少麻烦：网页改版后解析规则直接失效，高频访问容易触发平台风控，多规格、图片、店铺信息的解析逻辑需要大量开发时间，遇到网络波动、商品下架，程序很容易直接报错终止运行。

最近在实际项目中尝试了 OpenClaw 这套工具，它可以直接输出标准化结构化商品资料，不用处理页面解析、代理池、加密校验，我们只需要聚焦业务逻辑：定时拉取信息、本地存储、异动识别、数据分析，几十行代码就可以完成一套完整的监控方案稀土掘金。

整套流程分为 4 个部分：

资料获取：输入商品编号，拿到标题、原价、现价、库存、全部规格、店铺、图片等完整内容；
定时轮询：周期性采集目标商品，对比历史记录，识别价格、库存异动；
本地持久化：把每一次采集结果保存到本地数据库，留存时序记录；
数据分析：读取历史记录，做价格波动统计、规格销量统计，输出简单报表。

准备工作

安装依赖库

pip install requests apscheduler sqlite3

2.拿到两组身份凭证，用于 OpenClaw 的身份校验；

3.准备需要跟踪的商品编号，就是商品网页链接中 item.jd.com/xxx.html 中间那串数字。

提示：工具自带缓存机制，做实时监控的时候建议关闭缓存，保证拿到最新的行情；做批量数据分析场景，开启缓存可以减少消耗，提升响应速度。工具内置异常识别，商品下架、网络故障、平台维护都会返回对应的标记，程序不会直接崩溃退出。

完整可运行代码实现

1. 基础数据获取、清洗模块

import requests
import json
import time
import sqlite3
from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime

# ----------------配置区域----------------
KEY = "你的身份凭证key"
SECRET = "你的身份凭证secret"
MONITOR_ITEM_LIST = ["10335871600","10057467958584"] # 需要监控的商品编号列表
MONITOR_INTERVAL = 3600 # 监控周期，单位秒，这里设置1小时采集一次
CACHE_SWITCH = "no" # 实时监控设置no；批量分析可以设置yes开启缓存
# ----------------------------------------

def fetch_product_data(num_iid):
    """获取商品原始资料"""
    base_url = "https://api‑gw.onebound.cn/jd/item_get_pro"
    param = {
        "key": KEY,
        "secret": SECRET,
        "num_iid": num_iid,
        "cache": CACHE_SWITCH,
        "result_type":"json",
        "lang":"cn"
    }
    headers = {
        "Accept‑Encoding":"gzip",
        "Connection":"close"
    }
    try:
        resp = requests.get(base_url,params=param,headers=headers,timeout=20)
        res_json = resp.json()
        return res_json
    except Exception as e:
        print(f"商品{num_iid}获取失败，异常信息：{str(e)}")
        return None


def parse_raw_data(raw_json):
    """清洗原始返回内容，提取业务需要字段"""
    if not raw_json:
        return None
    error_code = raw_json.get("error_code")
    if error_code != "0000":
        err_msg = raw_json.get("reason","未知异常")
        print(f"业务异常：{err_msg}")
        return None
    item = raw_json.get("item")
    if not item:
        return None

    # 提取多规格sku列表
    sku_list = []
    sku_data = item.get("skus",{}).get("sku",[])
    for sku in sku_data:
        sku_info = {
            "sku_id":sku.get("sku_id"),
            "properties_name":sku.get("properties_name"),
            "price":float(sku.get("price",0)),
            "original_price":float(sku.get("orginal_price",0)),
            "stock":int(sku.get("quantity",0))
        }
        sku_list.append(sku_info)

    clean_result = {
        "num_iid":item.get("num_iid"),
        "title":item.get("title"),
        "brand":item.get("brand",""),
        "current_price":float(item.get("price",0)),
        "original_price":float(item.get("orginal_price",0)),
        "stock_total":item.get("num",0),
        "shop_name":item.get("seller_info",{}).get("shop_name",""),
        "main_image":item.get("pic_url",""),
        "location":item.get("location",""),
        "skus":json.dumps(sku_list,ensure_ascii=False),
        "crawl_time":datetime.now().strftime("%Y‑%m‑%d %H:%M:%S")
    }
    return clean_result

2. SQLite 本地存储模块

把每次采集的数据存入本地数据库，方便后续对比和数据分析。

def init_db():
    """初始化本地数据表"""
    conn = sqlite3.connect("./product_monitor.db")
    cursor = conn.cursor()
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS product_history(
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        num_iid TEXT,
        title TEXT,
        brand TEXT,
        current_price REAL,
        original_price REAL,
        stock_total INTEGER,
        shop_name TEXT,
        skus TEXT,
        crawl_time TEXT
    )
    ''')
    conn.commit()
    conn.close()


def save_to_db(data):
    """将清洗完成的数据写入数据库"""
    if not data:
        return
    conn = sqlite3.connect("./product_monitor.db")
    cursor = conn.cursor()
    sql = '''
    INSERT INTO product_history
    (num_iid,title,brand,current_price,original_price,stock_total,shop_name,skus,crawl_time)
    VALUES (?,?,?,?,?,?,?,?,?)
    '''
    val = (
        data["num_iid"],data["title"],data["brand"],
        data["current_price"],data["original_price"],data["stock_total"],
        data["shop_name"],data["skus"],data["crawl_time"]
    )
    cursor.execute(sql,val)
    conn.commit()
    conn.close()
    print(f"✅成功保存记录：商品编号{data['num_iid']}，采集时间{data['crawl_time']}")

3. 异动检测 + 定时监控任务

对比上一轮记录，识别价格下调上涨、库存减少，简单打印预警，实际项目可以扩展为企业微信、钉钉消息推送。

def check_abnormal(num_iid,new_data):
    """对比历史数据，检测异动"""
    conn = sqlite3.connect("./product_monitor.db")
    cursor = conn.cursor()
    cursor.execute('''
    SELECT current_price,stock_total FROM product_history
    WHERE num_iid = ? ORDER BY id DESC LIMIT 1
    ''',(num_iid,))
    last_record = cursor.fetchone()
    conn.close()
    if not last_record:
        return
    last_price,last_stock = last_record
    new_price = new_data["current_price"]
    new_stock = new_data["stock_total"]

    # 价格异动
    if new_price != last_price:
        print(f"⚠️【价格异动提醒】商品{num_iid}，旧价{last_price} → 新价{new_price}")
    # 库存异动
    if new_stock != last_stock:
        print(f"⚠️【库存异动提醒】商品{num_iid}，旧库存{last_stock} → 新库存{new_stock}")


def monitor_task():
    """单次监控任务，循环处理全部目标商品"""
    for item_id in MONITOR_ITEM_LIST:
        raw = fetch_product_data(item_id)
        clean_data = parse_raw_data(raw)
        if clean_data:
            check_abnormal(item_id,clean_data)
            save_to_db(clean_data)
        time.sleep(1) # 请求之间短暂休眠，避免请求过于密集

4. 简单数据分析示例

读取本地历史数据，输出价格变化统计，你可以基于此扩展，做 matplotlib 绘图导出图表，或者导出 csv 报表。

def analysis_product(num_iid):
    """针对单个商品，读取历史记录做简单数据分析"""
    conn = sqlite3.connect("./product_monitor.db")
    cursor = conn.cursor()
    cursor.execute('''
    SELECT crawl_time,current_price,stock_total FROM product_history
    WHERE num_iid = ? ORDER BY id
    ''',(num_iid,))
    rows = cursor.fetchall()
    conn.close()
    if len(rows) == 0:
        print(f"商品{num_iid}暂无历史记录")
        return
    price_list = [r[1] for r in rows]
    min_p = min(price_list)
    max_p = max(price_list)
    avg_p = sum(price_list)/len(price_list)
    print(f"\n📊商品{num_iid}历史数据分析结果")
    print(f"记录总条数：{len(rows)}")
    print(f"历史最低价格：{min_p}")
    print(f"历史最高价格：{max_p}")
    print(f"平均价格：{round(avg_p,2)}")
    print("时间‑价格轨迹：")
    for line in rows:
        print(f"  {line[0]} | 价格:{line[1]} |库存:{line[2]}")


if __name__ == "__main__":
    init_db()
    print("启动商品监控程序...")
    scheduler = BackgroundScheduler()
    scheduler.add_job(monitor_task,"interval",seconds=MONITOR_INTERVAL)
    scheduler.start()
    # 先执行一次采集
    monitor_task()
    # 演示数据分析，填入需要分析的商品编号
    analysis_product(MONITOR_ITEM_LIST[0])

    try:
        while True:
            time.sleep(60)
    except KeyboardInterrupt:
        print("\n程序退出")
        scheduler.shutdown()

关键说明与踩坑记录

关于多规格处理 返回的规格列表中，每一条都独立带有规格编号、售价、库存。实际业务不要只拿商品主体价格，主体价格有可能会和真实规格售价不一致，做竞品比价建议遍历全部规格分别跟踪记录。
缓存开关怎么选

7*24 小时实时盯价格、盯库存：CACHE_SWITCH="no"，每次都获取最新行情；
一次性批量拉取大量商品做选品分析：CACHE_SWITCH="yes"，复用缓存，提升速度、减少消耗。

异常状态处理 当商品下架、编号错误的时候，返回结果会带上异常标记，代码中已经做判断，不会直接崩溃。日常开发可以把异常日志写入文件，方便排查问题。
扩展方向

异动消息推送：检测到价格 / 库存变动后，调用钉钉 / 企业微信推送通知；
可视化：读取 sqlite 数据，用 matplotlib 绘制价格趋势图，导出图片；
导出报表：把历史记录直接导出为 CSV，交给运营做复盘；
多商品批量分析，批量导出全部监控商品统计报告。

小结

对比传统手写解析脚本，OpenClaw 最大的优势就是把繁杂的页面适配工作做了封装。开发者不用再关心网页改版、反拦截、JS 渲染这些问题，我们把精力聚焦业务：定时采集、异动告警、数据统计分析。几十行代码，就实现 7×24 小时不间断商品监控，对于中小团队和个人开发者是性价比很高的落地方案。

万邦api博客

Nice to meet you, too!

api 数据挖掘