Open‑Claw 实战｜无需逆向，快速搭建电商商品监控与数据分析系统

pip install requests

import requests
import json
import time
import os
from datetime import datetime

# --------------------------配置区，请修改这里--------------------------
# 你的身份凭证
ACCESS_KEY = "your_access_key"
ACCESS_SECRET = "your_access_secret"
# 需要监控的商品ID列表，可以填多个
MONITOR_ITEM_IDS = ["10335871600"]
# 监控轮询间隔，单位秒，不要设置过小，避免访问过于频繁
MONITOR_INTERVAL = 60 * 30
# 本地快照存储目录
SNAPSHOT_DIR = "./product_snapshot"
# ----------------------------------------------------------------------

def init_dir():
    """初始化存储目录"""
    if not os.path.exists(SNAPSHOT_DIR):
        os.makedirs(SNAPSHOT_DIR)

def fetch_product_raw_data(item_id: str):
    """
    通过Open‑Claw工具获取商品原始结构化数据
    :param item_id: 商品编号
    :return: dict 商品原始字典，失败返回None
    """
    base_url = "这里填入你的访问地址"
    params = {
        "key": ACCESS_KEY,
        "secret": ACCESS_SECRET,
        "num_iid": item_id
    }
    headers = {
        "Accept‑Encoding": "gzip",
        "Connection": "close"
    }
    try:
        resp = requests.get(base_url, params=params, headers=headers, timeout=15)
        resp.raise_for_status()
        result = resp.json()
        # 判断返回状态
        if result.get("error_code") != "0000":
            print(f"【获取失败】商品 {item_id} ,原因：{result.get('reason','未知错误')}")
            return None
        item_data = result.get("item")
        if not item_data:
            print(f"【警告】商品 {item_id} 返回数据为空")
            return None
        return item_data
    except Exception as e:
        print(f"【网络异常】item_id:{item_id},异常信息:{str(e)}")
        return None

def save_snapshot(item_id: str, data: dict):
    """保存一次采集快照到本地文件"""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    item_folder = os.path.join(SNAPSHOT_DIR, item_id)
    if not os.path.exists(item_folder):
        os.makedirs(item_folder)
    save_path = os.path.join(item_folder, f"snap_{timestamp}.json")
    with open(save_path, "w", encoding="utf‑8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    print(f"✅已保存快照：{save_path}")

def get_last_snapshot(item_id: str):
    """读取该商品上一次的快照，用于对比变化"""
    item_folder = os.path.join(SNAPSHOT_DIR, item_id)
    if not os.path.exists(item_folder):
        return None
    file_list = sorted(os.listdir(item_folder))
    if len(file_list) == 0:
        return None
    last_file = file_list[-1]
    full_path = os.path.join(item_folder, last_file)
    with open(full_path, "r", encoding="utf‑8") as f:
        return json.load(f)

def detect_change(old: dict, new: dict):
    """对比新旧快照，检测价格、库存变化，输出告警信息"""
    alert_msg = []
    old_price = float(old.get("price", 0) or 0)
    new_price = float(new.get("price", 0) or 0)
    old_stock = int(old.get("num", 0) or 0)
    new_stock = int(new.get("num", 0) or 0)
    title = new.get("title", "未知商品")

    # 价格变动检测
    if new_price != old_price and new_price > 0:
        alert_msg.append(f"【价格变动】{title} 旧价:{old_price} → 新价:{new_price}")
    # 库存变动检测
    if new_stock != old_stock:
        if new_stock <= 0:
            alert_msg.append(f"⚠️【库存告警】{title} 商品已经缺货！库存:{new_stock}")
        else:
            alert_msg.append(f"【库存变动】{title} 旧库存:{old_stock} → 新库存:{new_stock}")
    for msg in alert_msg:
        print(msg)
    return alert_msg

def analysis_history(item_id: str):
    """简单数据分析：读取历史快照，输出价格最高最低、历史记录条数"""
    item_folder = os.path.join(SNAPSHOT_DIR, item_id)
    if not os.path.exists(item_folder):
        print(f"商品 {item_id} 暂无历史快照数据")
        return
    file_list = sorted(os.listdir(item_folder))
    price_list = []
    for fname in file_list:
        fp = os.path.join(item_folder, fname)
        with open(fp, "r", encoding="utf‑8") as f:
            d = json.load(f)
        p = float(d.get("price",0) or 0)
        if p>0:
            price_list.append(p)
    if len(price_list) == 0:
        print("没有有效价格记录")
        return
    print("\n====历史数据分析结果====")
    print(f"商品ID：{item_id}")
    print(f"采集快照总条数：{len(file_list)}")
    print(f"历史最高价格：{max(price_list)}")
    print(f"历史最低价格：{min(price_list)}")
    print(f"价格均值：{round(sum(price_list)/len(price_list),2)}")
    print("========================\n")

def monitor_loop():
    init_dir()
    print("🛰️商品监控程序启动...")
    while True:
        for item_id in MONITOR_ITEM_IDS:
            new_data = fetch_product_raw_data(item_id)
            if new_data is None:
                continue
            save_snapshot(item_id, new_data)
            old_data = get_last_snapshot(item_id)
            if old_data is not None:
                detect_change(old_data, new_data)
        # 每轮结束之后做一次数据分析
        for item_id in MONITOR_ITEM_IDS:
            analysis_history(item_id)
        print(f"\n本轮巡检完成，等待 {MONITOR_INTERVAL} 秒进入下一轮\n")
        time.sleep(MONITOR_INTERVAL)

if __name__ == "__main__":
    monitor_loop()