告别人工盯品！基于 Open Claw 搭建京东商品全自动监控与数据分析系统（完整可运行代码）

pip install requests pandas sqlite3 schedule

import requests
import json
import sqlite3
import time
import pandas as pd
import schedule
from datetime import datetime

# ===================== 配置区，自行修改 =====================
# Open Claw身份凭证
CLAW_KEY = "替换为你的专属KEY"
CLAW_SECRET = "替换为你的专属SECRET"
# 需要监控的商品ID列表
MONITOR_GOODS_IDS = ["10335871600", "10057467958584"]
# 监控轮询间隔，单位秒（普通商品5分钟，爆款建议1分钟）
POLL_SECONDS = 300
# 降价预警阈值，0.2=降价超20%触发提醒
PRICE_ALERT_THRESHOLD = 0.2
# 本地数据库文件名
DB_NAME = "goods_monitor.db"
# ==========================================================

def init_sqlite_db():
    """初始化本地数据库，创建商品数据表"""
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    # 建表：存储每一次采集的全量商品数据
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS goods_data (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        goods_id TEXT NOT NULL,
        collect_time TEXT NOT NULL,
        title TEXT,
        brand TEXT,
        current_price REAL,
        original_price REAL,
        stock_num INTEGER,
        shop_name TEXT,
        sales_volume INTEGER,
        sku_info TEXT
    )
    ''')
    conn.commit()
    conn.close()
    print("数据库初始化完成")

def fetch_single_goods(goods_id):
    """通过Open Claw通道拉取单款商品完整结构化数据"""
    request_url = "https://api-gw.onebound.cn/jd/item_get_pro"
    params = {
        "key": CLAW_KEY,
        "secret": CLAW_SECRET,
        "num_iid": goods_id,
        "result_type": "json",
        "cache": "yes",
        "lang": "cn"
    }
    headers = {
        "Accept-Encoding": "gzip",
        "Connection": "close"
    }
    try:
        resp = requests.get(url=request_url, params=params, headers=headers, timeout=15)
        resp_json = resp.json()
        # 判断返回状态，0000代表数据正常返回
        if resp_json.get("error_code") != "0000":
            print(f"【采集失败】商品{goods_id}，错误信息：{resp_json.get('error')}")
            return None
        item_data = resp_json.get("item", {})
        return item_data
    except Exception as e:
        print(f"【网络异常】商品{goods_id}采集出错：{str(e)}")
        return None

def save_data_to_db(goods_raw):
    """将采集到的商品数据存入本地数据库"""
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    now_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    goods_id = goods_raw.get("num_iid")
    title = goods_raw.get("title", "")
    brand = goods_raw.get("brand", "")
    curr_price = float(goods_raw.get("price", 0))
    ori_price = float(goods_raw.get("orginal_price", 0))
    stock = int(goods_raw.get("num", 0))
    shop = goods_raw.get("seller_info", {}).get("shop_name", "")
    sales = int(goods_raw.get("sales", 0))
    sku_str = json.dumps(goods_raw.get("sku", {}), ensure_ascii=False)

    insert_sql = '''
    INSERT INTO goods_data 
    (goods_id, collect_time, title, brand, current_price, original_price, stock_num, shop_name, sales_volume, sku_info)
    VALUES (?,?,?,?,?,?,?,?,?,?)
    '''
    cursor.execute(insert_sql, (goods_id, now_time, title, brand, curr_price, ori_price, stock, shop, sales, sku_str))
    conn.commit()
    conn.close()
    print(f"【数据入库成功】{goods_id} | {title[:20]}... 现价：{curr_price}")

def check_price_alert(goods_raw):
    """对比历史数据，判断是否触发降价预警"""
    goods_id = goods_raw.get("num_iid")
    curr_price = float(goods_raw.get("price", 0))
    conn = sqlite3.connect(DB_NAME)
    df_history = pd.read_sql(
        f"SELECT current_price FROM goods_data WHERE goods_id = '{goods_id}' ORDER BY collect_time DESC LIMIT 1",
        conn
    )
    conn.close()
    # 首次采集无历史数据，跳过对比
    if len(df_history) == 0:
        return
    last_price = df_history.iloc[0]["current_price"]
    # 计算降价幅度
    if last_price == 0:
        return
    drop_rate = (last_price - curr_price) / last_price
    if drop_rate >= PRICE_ALERT_THRESHOLD:
        print("="*60)
        print(f"⚠️ 价格大幅变动预警 | 商品ID：{goods_id}")
        print(f"上次价格：{last_price}  当前价格：{curr_price}  降幅：{round(drop_rate*100,2)}%")
        print("="*60)

def goods_monitor_task():
    """单次监控主任务：循环采集所有商品、入库、预警判断"""
    print(f"\n===== 新一轮监控启动 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} =====")
    for goods_id in MONITOR_GOODS_IDS:
        item_info = fetch_single_goods(goods_id)
        if not item_info:
            continue
        # 价格变动预警检测
        check_price_alert(item_info)
        # 数据存入数据库
        save_data_to_db(item_info)
        time.sleep(1) # 轻微延时，降低通道压力

def export_goods_trend(goods_id, day_count=7):
    """数据分析函数：导出单商品近N天价格趋势Excel报表"""
    conn = sqlite3.connect(DB_NAME)
    df = pd.read_sql(f'''
        SELECT collect_time, current_price, original_price, stock_num, sales_volume 
        FROM goods_data 
        WHERE goods_id = '{goods_id}' 
        ORDER BY collect_time ASC
    ''', conn)
    conn.close()
    if len(df) == 0:
        print(f"商品{goods_id}暂无历史数据，无法导出报表")
        return
    save_path = f"商品{goods_id}_价格趋势报表.xlsx"
    df.to_excel(save_path, index=False)
    print(f"趋势报表已导出至：{save_path}，共{len(df)}条历史记录")
    return df

if __name__ == "__main__":
    # 初始化数据库
    init_sqlite_db()
    # 立即执行一次监控
    goods_monitor_task()
    # 定时循环调度
    schedule.every(POLL_SECONDS).seconds.do(goods_monitor_task)
    print(f"\n定时监控已启动，每{POLL_SECONDS/60}分钟自动执行一次，关闭程序终止监控")
    while True:
        schedule.run_pending()
        time.sleep(1)

import sqlite3
import pandas as pd

DB_NAME = "goods_monitor.db"

def batch_export_all_goods():
    """导出所有监控商品最新一轮数据，横向对比竞品"""
    conn = sqlite3.connect(DB_NAME)
    # 取每个商品最新采集记录
    df_all = pd.read_sql('''
        SELECT * FROM goods_data gd1
        WHERE collect_time = (
            SELECT MAX(collect_time) FROM goods_data gd2 WHERE gd2.goods_id = gd1.goods_id
        )
    ''', conn)
    conn.close()
    save_file = "全商品竞品对比表.xlsx"
    df_all.to_excel(save_file, index=False)
    print(f"竞品对比表导出完成，路径：{save_file}")

if __name__ == "__main__":
    # 导出全部商品最新数据对比
    batch_export_all_goods()
    # 单独导出某款商品7日价格趋势
    from monitor_jd_goods import export_goods_trend
    export_goods_trend("10335871600", day_count=7)

# 钉钉推送示例，填入自己的机器人webhook地址
ding_webhook = "你的钉钉机器人链接"
push_data = {
    "msgtype": "text",
    "text": {"content": f"商品大幅降价预警：{goods_id}，降幅{round(drop_rate*100,2)}%"}
}
requests.post(ding_webhook, json=push_data)