竞品监控与商品数据分析实战：用 Open‑Claw 搭建自动化商品跟踪系统

pip install requests pandas

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

# ========== 配置区域，请修改为自己的凭证和目标商品 ==========
KEY = "你的身份key"
SECRET = "你的身份密钥"
# 需要监控的商品ID列表，可以填入多个商品编号
MONITOR_ITEM_IDS = ["10335871600", "10057467958584"]
# 监控间隔，单位秒，示例设置30分钟扫描一次，请勿设置过小
MONITOR_INTERVAL = 1800
# 本地存储文件
SAVE_CSV = "goods_monitor_data.csv"
# =========================================================

def fetch_goods_data(item_id):
    """获取单款商品结构化数据"""
    base_url = "https://api‑gw.onebound.cn/jd/item_get_pro"
    params = {
        "key": KEY,
        "secret": SECRET,
        "num_iid": item_id,
        "result_type": "json",
        "cache": "no"  # 设置no获取最新数据；yes读取缓存速度更快
    }
    headers = {
        "Accept‑Encoding": "gzip",
        "Connection": "close"
    }
    try:
        resp = requests.get(base_url, params=params, headers=headers, timeout=20)
        result = resp.json()
    except Exception as e:
        print(f"商品 {item_id} 请求异常：{str(e)}")
        return None

    error_code = result.get("error_code")
    if error_code != "0000":
        err_msg = result.get("error", "未知错误")
        print(f"商品 {item_id} 获取失败，错误：{err_msg}")
        return None

    item_data = result.get("item", {})
    if not item_data:
        print(f"商品 {item_id} 返回内容为空")
        return None

    # 提取核心字段
    output = {
        "crawl_time": time.strftime("%Y‑%m‑%d %H:%M:%S"),
        "num_iid": item_data.get("num_iid"),
        "title": item_data.get("title"),
        "brand": item_data.get("brand"),
        "price": item_data.get("price"),
        "orginal_price": item_data.get("orginal_price"),
        "stock": item_data.get("num"),
        "shop_name": item_data.get("seller_info", {}).get("shop_name", ""),
        "detail_url": item_data.get("detail_url"),
        "pic_url": item_data.get("pic_url")
    }
    return output


def save_to_csv(data_list):
    """把采集到的数据追加写入CSV"""
    df_new = pd.DataFrame(data_list)
    try:
        df_old = pd.read_csv(SAVE_CSV, encoding="utf‑8‑sig")
        df_total = pd.concat([df_old, df_new], ignore_index=True)
    except FileNotFoundError:
        df_total = df_new
    df_total.to_csv(SAVE_CSV, index=False, encoding="utf‑8‑sig")
    print(f"✅ {len(data_list)} 条数据已写入本地文件 {SAVE_CSV}")


def analysis_data():
    """简单本地数据分析示例，读取CSV做基础统计"""
    try:
        df = pd.read_csv(SAVE_CSV, encoding="utf‑8‑sig")
        df["price"] = pd.to_numeric(df["price"], errors="coerce")
        print("\n==== 简单数据分析报告 ====")
        print(f"总采集记录数：{len(df)}")
        print(f"监控商品数量：{df['num_iid'].nunique()}")
        print(f"商品平均售价：{df['price'].mean():.2f}")
        print("\n各商品最新价格：")
        latest = df.sort_values("crawl_time").groupby("num_iid").last()
        print(latest[["title", "price", "stock"]])
    except Exception as e:
        print(f"分析数据异常：{str(e)}，可能还没有采集到数据")


def main_loop():
    print("🚀 启动商品监控任务")
    while True:
        collect_result = []
        for goods_id in MONITOR_ITEM_IDS:
            item_info = fetch_goods_data(goods_id)
            if item_info:
                collect_result.append(item_info)
                print(f"已获取：{item_info.get('title')} | 当前价：{item_info.get('price')}")
            time.sleep(2)  # 每个请求之间间隔，避免访问过于密集
        if collect_result:
            save_to_csv(collect_result)
            analysis_data()
        print(f"\n⏳ 等待 {MONITOR_INTERVAL} 秒后进行下一轮扫描...\n")
        time.sleep(MONITOR_INTERVAL)


if __name__ == "__main__":
    main_loop()

# 解析SKU规格示例
sku_list = item_data.get("skus", {}).get("sku", [])
for sku in sku_list:
    print(f"规格名称：{sku.get('properties_name')}")
    print(f"规格价格：{sku.get('price')}，库存：{sku.get('quantity')}")