告别手工盯品！Open Claw 快速搭建电商商品监控与数据分析系统（完整可运行源码）

pip install requests apscheduler

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

# ========================【配置区域，请自行修改】========================
CLAW_KEY = "替换成你的凭证key"
CLAW_SECRET = "替换成你的凭证secret"
# 需要监控的商品编号列表，可以批量添加多个商品
MONITOR_GOODS = ["10335871600", "10057467958584"]
POLL_INTERVAL = 300  # 监控轮询间隔，单位秒，建议最低不要低于300秒
PRICE_DROP_ALERT_RATIO = 0.15  # 降价幅度超过15%触发告警
# =====================================================================

# 初始化数据库，创建数据表
def init_db():
    conn = sqlite3.connect("goods_monitor.db")
    cur = conn.cursor()
    cur.execute('''
    CREATE TABLE IF NOT EXISTS goods_record (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        goods_id TEXT NOT NULL,
        title TEXT,
        price REAL,
        original_price REAL,
        stock INTEGER,
        shop_name TEXT,
        crawl_time TEXT
    )
    ''')
    conn.commit()
    conn.close()

# 获取商品详情原始结构化数据
def get_goods_data(goods_id: str):
    params = {
        "key": CLAW_KEY,
        "secret": CLAW_SECRET,
        "num_iid": goods_id,
        "result_type": "json",
        "cache": "yes"
    }
    try:
        resp = requests.get("https://api-gw.onebound.cn/jd/item_get_pro", params=params, timeout=15)
        res_json = resp.json()
        if res_json.get("error_code") != "0000":
            print(f"商品 {goods_id} 获取数据异常：{res_json.get('reason','未知错误')}")
            return None
        item = res_json.get("item")
        if not item:
            print(f"商品 {goods_id} 返回数据为空")
            return None
        data = {
            "goods_id": goods_id,
            "title": item.get("title", ""),
            "price": float(item.get("price", 0)) if item.get("price") else 0,
            "original_price": float(item.get("orginal_price",0)) if item.get("orginal_price") else 0,
            "stock": int(item.get("num", 0)) if item.get("num") else 0,
            "shop_name": item.get("seller_info", {}).get("shop_name", ""),
            "crawl_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        }
        return data
    except Exception as e:
        print(f"网络请求异常 goods_id:{goods_id}, 异常信息:{str(e)}")
        return None

# 将本次采集结果存入数据库
def save_to_db(data):
    conn = sqlite3.connect("goods_monitor.db")
    cur = conn.cursor()
    cur.execute('''
    INSERT INTO goods_record(goods_id,title,price,original_price,stock,shop_name,crawl_time)
    VALUES (?,?,?,?,?,?,?)
    ''',(
        data["goods_id"],
        data["title"],
        data["price"],
        data["original_price"],
        data["stock"],
        data["shop_name"],
        data["crawl_time"]
    ))
    conn.commit()
    conn.close()

# 获取这个商品上一次采集的记录，用于对比异动
def get_last_record(goods_id):
    conn = sqlite3.connect("goods_monitor.db")
    cur = conn.cursor()
    cur.execute('''
    SELECT price,stock FROM goods_record WHERE goods_id = ? ORDER BY id DESC LIMIT 1
    ''',(goods_id,))
    row = cur.fetchone()
    conn.close()
    if row:
        return {"last_price": row[0], "last_stock": row[1]}
    return None

# 监控单款商品，检测价格、库存变动告警
def monitor_one_goods(goods_id):
    print(f"\n===== 开始采集商品：{goods_id} =====")
    curr_data = get_goods_data(goods_id)
    if curr_data is None:
        return
    save_to_db(curr_data)
    last_info = get_last_record(goods_id)
    print(f"【{curr_data['title']}】 当前售价：{curr_data['price']}，库存：{curr_data['stock']}")
    # 第一次采集，没有历史记录，只存储不告警
    if last_info is None:
        print("首次采集，无历史对比数据")
        return
    last_price = last_info["last_price"]
    last_stock = last_info["last_stock"]
    # 价格下降告警
    if curr_data["price"] < last_price:
        drop_rate = (last_price - curr_data["price"]) / last_price
        if drop_rate >= PRICE_DROP_ALERT_RATIO:
            print(f"🔔【价格大幅下降告警】原价格{last_price} → 当前{curr_data['price']}，降幅{round(drop_rate*100,2)}%")
    # 库存清零告警
    if curr_data["stock"] == 0 and last_stock > 0:
        print(f"⚠️【库存告急告警】商品已经售罄！上一轮库存：{last_stock}")
    # 补货提醒
    if curr_data["stock"] > 0 and last_stock == 0:
        print(f"✅【补货通知】商品恢复库存，当前库存 {curr_data['stock']}")

# 历史数据分析：读取数据库，输出商品价格统计
def analysis_goods_history(goods_id):
    """输出商品历史价格统计：最高、最低、平均价格"""
    conn = sqlite3.connect("goods_monitor.db")
    cur = conn.cursor()
    cur.execute('''
    SELECT price,crawl_time FROM goods_record WHERE goods_id = ? ORDER BY crawl_time ASC
    ''', (goods_id,))
    rows = cur.fetchall()
    conn.close()
    if len(rows) < 2:
        print(f"\n商品{goods_id}历史数据不足，暂时无法分析，多跑几次采集后再来")
        return
    price_list = [r[0] for r in rows]
    max_p = max(price_list)
    min_p = min(price_list)
    avg_p = round(sum(price_list)/len(price_list),2)
    first_time = rows[0][1]
    last_time = rows[-1][1]
    print("\n====================历史数据分析报告====================")
    print(f"商品编号：{goods_id}")
    print(f"统计时间区间：{first_time} ~ {last_time}")
    print(f"历史最高售价：{max_p}")
    print(f"历史最低售价：{min_p}")
    print(f"历史平均售价：{avg_p}")
    print("========================================================")

# 定时任务执行入口
def monitor_task():
    for gid in MONITOR_GOODS:
        monitor_one_goods(gid)
    # 可选：每轮监控完成后，对第一个商品输出统计报告
    analysis_goods_history(MONITOR_GOODS[0])

if __name__ == "__main__":
    init_db()
    scheduler = BackgroundScheduler()
    scheduler.add_job(monitor_task, "interval", seconds=POLL_INTERVAL)
    print(f"监控程序已启动，监控商品列表：{MONITOR_GOODS}，轮询间隔 {POLL_INTERVAL} 秒")
    print("Ctrl+C 停止程序\n")
    scheduler.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        scheduler.shutdown()
        print("\n程序正常退出")