×

api 数据挖掘

OpenClaw 实战｜告别手工盯盘，搭建电商商品自动化监控与数据分析体系

admin admin 发表于2026-10-14 09:18:47 浏览8 评论0

抢沙发发表评论

适用场景：竞品盯价、库存异动跟踪、促销活动观察、商品数据归档复盘，Python 完整可运行脚本，适合开发者、电商运营做二次落地。


前言

做电商运营或者市场调研的时候，很多人都有这样的痛点：需要持续关注一批竞品商品，手动打开网页复制价格、库存、规格信息，一天反复刷好几次，耗费大量时间，还很容易漏掉降价、断货、上新规格这类关键变化。

如果商品数量多，几十上百个商品，纯人工维护基本不现实。直接写原生页面抓取，又会遇到页面加密、动态渲染、频繁访问被限制等一堆麻烦，调试成本很高。

最近在实践 OpenClaw 这套工具链，搭配成熟的数据获取能力，可以快速搭建一套稳定的商品监控系统。不需要处理复杂页面解析逻辑，只需要关注业务逻辑：定时拉取商品信息、对比历史记录、识别异动、存储数据、简单统计分析。

本文就完整分享整套落地流程，从环境准备、单商品数据获取、定时监控、异动检测，再到简单数据分析，全部附带可直接运行的 Python 代码。

整体实现思路

整套系统分为 4 个核心环节：

  2. 数据获取：输入商品编号，拿到商品标题、售价、原价、库存、店铺、多规格信息、图片链接等结构化内容。

  4. 本地持久化：把每一轮采集的数据保存到本地数据库，留存历史快照，方便后续做对比和数据分析。

  6. 异动检测：对比上一轮记录，识别价格变动、库存清零、规格增减等变化，输出告警提示。

  8. 数据分析输出：基于历史数据，输出价格波动统计、库存变化记录，导出表格做报表。

环境依赖：Python3.8 及以上，需要安装 requests、pandas、sqlite3（python 自带）

pip install requests pandas

完整代码实现

1. 基础封装：获取商品原始结构化数据

把基础请求逻辑封装，输入商品编号，返回整理后的商品字典。

# -*- coding:utf-8 -*-
import requests
import json
import time
import sqlite3
import pandas as pd

# -------------------------- 配置区域，请自行修改 --------------------------
ACCESS_KEY = "你的访问凭证key"
ACCESS_SECRET = "你的访问凭证secret"
MONITOR_GOODS_ID_LIST = ["10335871600", "10057467958584"]  # 需要监控的商品编号列表
MONITOR_INTERVAL = 60 * 30  # 监控轮询间隔，单位秒，示例30分钟跑一次
DB_FILE = "./goods_monitor.db"
# ------------------------------------------------------------------------

def fetch_goods_detail(goods_id: str):
    """
    获取商品详情原始结构化数据
    :param goods_id: 商品编号
    :return: dict 整理后的商品信息，失败返回None
    """
    base_url = "https://api-gw.onebound.cn/jd/item_get_pro/"
    params = {
        "key": ACCESS_KEY,
        "secret": ACCESS_SECRET,
        "num_iid": goods_id,
        "result_type": "json"
    }
    headers = {
        "Accept-Encoding": "gzip",
        "Connection": "close"
    }
    try:
        resp = requests.get(base_url, params=params, headers=headers, timeout=20)
        resp_json = resp.json()
    except Exception as e:
        print(f"【网络异常】商品 {goods_id} 请求失败：{str(e)}")
        return None

    # 判断返回状态
    if resp_json.get("error_code") != "0000":
        err_msg = resp_json.get("reason", "未知错误")
        print(f"【获取失败】商品 {goods_id} ,错误信息：{err_msg}")
        return None

    item_data = resp_json.get("item", {})
    if not item_data:
        print(f"【警告】商品 {goods_id} 返回数据为空")
        return None

    # 整理核心字段
    result = {
        "num_iid": item_data.get("num_iid"),
        "title": item_data.get("title", ""),
        "price": float(item_data.get("price", 0)) if item_data.get("price") else 0,
        "orginal_price": float(item_data.get("orginal_price", 0)) if item_data.get("orginal_price") else 0,
        "stock": item_data.get("num", 0),
        "brand": item_data.get("brand", ""),
        "shop_name": item_data.get("seller_info", {}).get("shop_name", ""),
        "detail_url": item_data.get("detail_url", ""),
        "capture_ts": int(time.time()),
        "capture_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    }

    # 提取多规格SKU信息
    sku_list = []
    skus_raw = item_data.get("skus", {}).get("sku", [])
    for sku in skus_raw:
        sku_info = {
            "sku_id": sku.get("sku_id"),
            "sku_price": float(sku.get("price",0)) if sku.get("price") else 0,
            "sku_origin_price": float(sku.get("orginal_price",0)) if sku.get("orginal_price") else 0,
            "sku_prop": sku.get("properties_name", ""),
            "sku_stock": sku.get("quantity",0)
        }
        sku_list.append(sku_info)
    result["sku_list"] = sku_list
    return result

2. 数据库初始化，存储历史快照

使用轻量 sqlite 保存每一轮采集的数据，一张表存商品快照，一张表存规格快照，方便后续对比分析。

def init_database():
    """初始化数据库表"""
    conn = sqlite3.connect(DB_FILE)
    cur = conn.cursor()
    # 商品主表，每一次采集生成一条记录
    cur.execute('''
    CREATE TABLE IF NOT EXISTS goods_snapshot (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        num_iid TEXT NOT NULL,
        title TEXT,
        price REAL,
        orginal_price REAL,
        stock INTEGER,
        brand TEXT,
        shop_name TEXT,
        detail_url TEXT,
        capture_ts INTEGER,
        capture_time TEXT
    )
    ''')
    # SKU规格子表，保存每个规格的价格库存
    cur.execute('''
    CREATE TABLE IF NOT EXISTS goods_sku_snapshot (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        parent_snapshot_id INTEGER,
        sku_id TEXT,
        sku_price REAL,
        sku_origin_price REAL,
        sku_prop TEXT,
        sku_stock INTEGER
    )
    ''')
    conn.commit()
    conn.close()

def save_to_db(goods_data: dict):
    """把单次采集的商品数据存入数据库"""
    conn = sqlite3.connect(DB_FILE)
    cur = conn.cursor()
    cur.execute('''
    INSERT INTO goods_snapshot 
    (num_iid,title,price,orginal_price,stock,brand,shop_name,detail_url,capture_ts,capture_time)
    VALUES (?,?,?,?,?,?,?,?,?,?)
    ''',(
        goods_data["num_iid"],
        goods_data["title"],
        goods_data["price"],
        goods_data["orginal_price"],
        goods_data["stock"],
        goods_data["brand"],
        goods_data["shop_name"],
        goods_data["detail_url"],
        goods_data["capture_ts"],
        goods_data["capture_time"]
    ))
    snapshot_id = cur.lastrowid
    # 写入sku子表
    for sku in goods_data["sku_list"]:
        cur.execute('''
        INSERT INTO goods_sku_snapshot
        (parent_snapshot_id,sku_id,sku_price,sku_origin_price,sku_prop,sku_stock)
        VALUES (?,?,?,?,?,?)
        ''',(
            snapshot_id,
            sku["sku_id"],
            sku["sku_price"],
            sku["sku_origin_price"],
            sku["sku_prop"],
            sku["sku_stock"]
        ))
    conn.commit()
    conn.close()

3. 异动检测逻辑：对比上一轮数据，识别价格、库存变化

每次新拿到数据，查询数据库里该商品最近一条历史记录，做差值比对，输出变动提醒。

def get_last_snapshot(num_iid: str):
    """获取该商品上一次采集的快照"""
    conn = sqlite3.connect(DB_FILE)
    cur = conn.cursor()
    cur.execute('''
    SELECT price,orginal_price,stock FROM goods_snapshot 
    WHERE num_iid = ? ORDER BY capture_ts DESC LIMIT 1
    ''',(num_iid,))
    row = cur.fetchone()
    conn.close()
    if row is None:
        return None
    return {
        "price": row[0],
        "orginal_price": row[1],
        "stock": row[2]
    }

def detect_change(last_data, new_data):
    """检测商品是否发生异动，返回变动信息列表"""
    change_msg = []
    if last_data is None:
        change_msg.append("首次采集，无历史对比")
        return change_msg
    # 价格变化
    if abs(new_data["price"] - last_data["price"]) > 0.01:
        change_msg.append(f"价格变动：{last_data['price']} → {new_data['price']}")
    if abs(new_data["orginal_price"] - last_data["orginal_price"]) >0.01:
        change_msg.append(f"原价变动：{last_data['orginal_price']} → {new_data['orginal_price']}")
    # 库存变化
    if new_data["stock"] != last_data["stock"]:
        change_msg.append(f"库存变动：{last_data['stock']} → {new_data['stock']}")
    return change_msg

4. 监控主循环 + 简单数据分析导出

定时循环监控，同时提供函数可以把历史数据导出 Excel，做后续数据分析。

def run_monitor_round():
    """执行一轮完整监控任务"""
    print(f"\n========== 开始一轮监控 {time.strftime('%Y-%m-%d %H:%M:%S')} ==========")
    for goods_id in MONITOR_GOODS_ID_LIST:
        print(f"\n正在处理商品:{goods_id}")
        new_info = fetch_goods_detail(goods_id)
        if not new_info:
            continue
        last_info = get_last_snapshot(goods_id)
        changes = detect_change(last_info, new_info)
        if len(changes) > 0:
            print(f"⚠️【商品{goods_id} 检测到异动】")
            for m in changes:
                print(f" >> {m}")
        else:
            print(f"✅【商品{goods_id}无变化】当前价格:{new_info['price']},库存:{new_info['stock']}")
        save_to_db(new_info)

def export_analysis_excel(out_file="./goods_monitor_report.xlsx"):
    """导出全部历史快照到Excel，用于数据分析"""
    conn = sqlite3.connect(DB_FILE)
    df_main = pd.read_sql("SELECT * FROM goods_snapshot", conn)
    df_sku = pd.read_sql("SELECT * FROM goods_sku_snapshot", conn)
    conn.close()
    with pd.ExcelWriter(out_file) as writer:
        df_main.to_excel(writer, sheet_name="商品主快照", index=False)
        df_sku.to_excel(writer, sheet_name="规格快照", index=False)
    print(f"✅数据分析报表已导出：{out_file}")

if __name__ == "__main__":
    init_database()
    print("初始化完成，启动监控循环，按Ctrl+C停止程序")
    try:
        while True:
            run_monitor_round()
            # 可选：执行一次导出，注释掉就不会每次导出
            # export_analysis_excel()
            time.sleep(MONITOR_INTERVAL)
    except KeyboardInterrupt:
        print("\n程序手动退出，结束监控")
        # 退出前导出一份报表
        export_analysis_excel()

使用说明

  2. 将代码中ACCESS_KEY、ACCESS_SECRET替换成你自己的凭证；

  4. 修改MONITOR_GOODS_ID_LIST填入你需要监控的商品编号；

  6. MONITOR_INTERVAL代表轮询间隔，建议不要设置太短，避免频繁请求；

  8. 运行脚本，程序会循环执行监控，按Ctrl+C结束程序，结束时自动导出 Excel 报表。

运行之后，本地会生成goods_monitor.db数据库文件，保存全部历史快照，每次采集到价格、库存变动，控制台会打印告警信息。导出的 Excel 可以用来做价格趋势统计、库存波动复盘。

OpenClaw 如何赋能这套监控系统

很多人写监控脚本，难点不在于写请求代码，而在于工程化调度、异常容错、任务管理。

OpenClaw 可以把上面的业务逻辑封装成可复用的任务能力：

  2. 定时调度托管：不用自己写 while 循环 sleep，交给调度引擎，配置执行周期，程序崩溃自动重启重试；

  4. 异常容错处理：遇到网络超时、数据为空、返回异常，自动记录错误日志，不会因为单个商品失败导致整个监控任务挂掉；

  6. 告警扩展：检测到价格 / 库存异动后，可以对接消息推送，把变动消息直接发送到企业微信、钉钉，不用盯着控制台看输出；

  8. 数据分析能力增强：基于保存的历史快照，直接做趋势统计、批量报表生成，不用再手动写 pandas 处理逻辑；

  10. 横向扩展：后续增加更多平台商品监控，只需要新增数据获取逻辑，调度、存储、告警整套能力可以直接复用。

简单来说：上面的 Python 脚本是业务逻辑原型，OpenClaw 负责把原型变成稳定长期跑在服务器上的生产级监控任务。

拓展优化方向

  2. 告警推送：检测到 changes 非空的时候，增加钉钉 / 企业微信 webhook，实现消息推送；

  4. 阈值告警：设置心理价位，当商品价格低于目标价格才触发提醒；

  6. 增量过滤：历史数据库越来越大，可以增加定时清理旧快照逻辑，只保留最近 N 天记录；

  8. 多任务分组：把商品分成不同分组，不同分组设置不同监控频率，高优先级商品缩短轮询间隔；

  10. 可视化：读取 sqlite 数据，用 matplotlib 绘制价格波动曲线图，直观查看历史价格走势。

小结

对于电商从业者、市场调研人员，手动维护商品信息成本极高。借助 OpenClaw 搭配成熟的数据获取能力，我们不需要深陷页面逆向、反爬对抗，把精力聚焦在业务本身：定时采集、历史归档、异动识别、数据分析。


少长咸集

群贤毕至

访客