告别复杂页面解析，OpenClaw 快速搭建电商商品监控与数据分析脚本

pip install requests pandas schedule

# -*- coding:utf-8 -*-
"""
OpenClaw 电商商品监控&数据分析脚本
功能：商品信息获取、异常容错、本地保存历史、价格库存异动检测
"""
import requests
import time
import pandas as pd
import schedule
import json
from datetime import datetime

# ======================== 配置区域，请自行修改 ========================
# 你的个人凭证信息
ACCESS_KEY = "替换成你的key"
ACCESS_SECRET = "替换成你的secret"

# 需要监控的商品编号列表，可以添加多个商品ID
MONITOR_GOODS_ID = [
    "10335871600",
    "10057467958584"
]

# 监控间隔，单位秒，不要设置过小，避免调用过于频繁
MONITOR_INTERVAL = 60 * 10

# 历史数据保存文件
SAVE_FILE = "goods_monitor_history.csv"
# ====================================================================


def get_goods_detail(goods_id: str):
    """
    获取单个商品结构化数据
    goods_id:商品编号
    return: 成功返回字典，失败返回None
    """
    base_url = "https://api-gw.onebound.cn/jd/item_get_pro/"
    params = {
        "key": ACCESS_KEY,
        "secret": ACCESS_SECRET,
        "num_iid": goods_id,
        "cache": "no",   # no获取最新数据，yes读取缓存加快速度
        "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()
        error_code = resp_json.get("error_code")

        # 0000代表正常拿到数据
        if error_code == "0000":
            item_data = resp_json.get("item")
            return item_data
        else:
            err_msg = resp_json.get("error", "未知错误")
            print(f"【商品{goods_id}】获取失败，错误信息：{err_msg}")
            return None

    except Exception as e:
        print(f"【商品{goods_id}】网络或解析异常：{str(e)}")
        return None


def parse_item(raw_item: dict):
    """把原始返回数据，提取我们关心的核心字段"""
    if not raw_item:
        return None
    out = {}
    out["crawl_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    out["goods_id"] = raw_item.get("num_iid")
    out["title"] = raw_item.get("title", "")
    out["price"] = float(raw_item.get("price", 0)) if raw_item.get("price") else 0
    out["original_price"] = float(raw_item.get("orginal_price", 0)) if raw_item.get("orginal_price") else 0
    out["stock"] = raw_item.get("num", 0)
    out["shop_name"] = raw_item.get("seller_info", {}).get("shop_name", "")
    out["detail_url"] = raw_item.get("detail_url", "")
    # sku规格简单提取
    sku_list = raw_item.get("skus", {}).get("sku", [])
    sku_info = []
    for sku in sku_list:
        sku_info.append({
            "sku_id": sku.get("sku_id"),
            "spec_name": sku.get("properties_name"),
            "sku_price": sku.get("price"),
            "sku_stock": sku.get("quantity")
        })
    out["sku_info"] = json.dumps(sku_info, ensure_ascii=False)
    return out


def save_to_csv(data_row: dict):
    """单条记录追加保存到csv历史文件"""
    df_new = pd.DataFrame([data_row])
    try:
        # 文件存在就追加，不存在新建
        with open(SAVE_FILE, "r", encoding="utf-8-sig") as f:
            df_new.to_csv(SAVE_FILE, mode="a", header=False, index=False, encoding="utf-8-sig")
    except FileNotFoundError:
        df_new.to_csv(SAVE_FILE, mode="w", header=True, index=False, encoding="utf-8-sig")


last_status = {}  # 保存上一轮采集状态，用于异动对比


def monitor_task():
    """一轮完整监控任务，循环所有商品"""
    print(f"\n==========开始一轮监控 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ==========")
    global last_status
    for gid in MONITOR_GOODS_ID:
        item_raw = get_goods_detail(gid)
        parsed = parse_item(item_raw)
        if not parsed:
            continue
        save_to_csv(parsed)
        print(f"商品【{parsed['title'][0:25]}...】现价：{parsed['price']} 库存：{parsed['stock']}")

        # 和上一次数据对比，检测异动
        prev = last_status.get(gid)
        if prev:
            # 价格下跌
            if parsed["price"] < prev["price"]:
                print(f"⚠️ 价格异动提醒！商品{gid}发生降价，上次{prev['price']}，当前{parsed['price']}")
            # 库存变动
            if parsed["stock"] != prev["stock"]:
                print(f"⚠️ 库存变动提醒！商品{gid}库存由{prev['stock']}变更为{parsed['stock']}")
        # 更新记录
        last_status[gid] = parsed
        time.sleep(2)


def analysis_history():
    """简单数据分析示例，读取本地历史，打印每个商品价格区间"""
    try:
        df = pd.read_csv(SAVE_FILE, encoding="utf-8-sig")
        print("\n----------历史数据分析结果----------")
        for goods_id in df["goods_id"].unique():
            sub = df[df["goods_id"] == goods_id]
            title = sub.iloc[0]["title"]
            min_p = sub["price"].min()
            max_p = sub["price"].max()
            print(f"商品：{title[:30]}... 历史最低:{min_p} 历史最高:{max_p}，共采集{len(sub)}条记录")
    except Exception as e:
        print("暂无历史数据，等待采集完成", e)


if __name__ == "__main__":
    # 执行一次
    monitor_task()
    analysis_history()

    # 设置定时循环执行
    schedule.every(MONITOR_INTERVAL).seconds.do(monitor_task)
    print(f"\n已启动定时监控，每 {MONITOR_INTERVAL/60}分钟执行一次，Ctrl+C终止程序")
    while True:
        schedule.run_pending()
        time.sleep(1)