告别手动盯竞品！OpenClaw 淘宝商品 API 搭建自动化监控选品系统（附完整 Python 代码）

pip install requests sqlite3 schedule

import requests
import sqlite3
import json
import time
import schedule

# ====================== 配置区（自行修改）======================
# OpenClaw接口密钥（平台注册获取）
API_KEY = "你的接口密钥"
# 待监控商品ID列表，可批量添加多个淘系商品id
MONITOR_ITEM_IDS = ["723456123456", "723456789012"]
# 钉钉/企业微信Webhook告警地址，不需要告警可留空
ALERT_WEBHOOK = ""
# 监控扫描间隔，单位分钟
SCAN_INTERVAL = 30
# 爆款判定阈值
SALE_RISE_THRESHOLD = 30  # 销量涨幅超30判定异动
PRICE_CHANGE_THRESHOLD = 5  # 价格变动超5元触发提醒
# ===========================================================

# 1. 初始化本地数据库，存储商品历史数据
def init_db():
    conn = sqlite3.connect("taobao_monitor.db")
    cursor = conn.cursor()
    # 商品数据表
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS item_data (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            item_id TEXT NOT NULL,
            title TEXT,
            price REAL,
            coupon_price REAL,
            monthly_sale INTEGER,
            total_sale INTEGER,
            good_rate REAL,
            shop_score REAL,
            update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    conn.commit()
    conn.close()

# 2. 调用OpenClaw taobao.item_get_pro接口获取商品数据
def get_taobao_item_info(item_id):
    url = "https://open.onebound.cn/api/taobao/item_get_pro"
    params = {
        "key": API_KEY,
        "item_id": item_id
    }
    try:
        resp = requests.get(url, params=params, timeout=15)
        res_json = resp.json()
        if res_json.get("code") != 0:
            print(f"商品{item_id}接口请求失败：{res_json.get('msg')}")
            return None
        item_info = res_json.get("data", {})
        return item_info
    except Exception as e:
        print(f"接口请求异常：{str(e)}")
        return None

# 3. 写入数据库，返回上一次历史数据用于对比
def save_and_get_old_data(item_id, item_info):
    conn = sqlite3.connect("taobao_monitor.db")
    cursor = conn.cursor()
    # 查询上一条记录
    cursor.execute('''
        SELECT price, coupon_price, monthly_sale, total_sale FROM item_data 
        WHERE item_id = ? ORDER BY update_time DESC LIMIT 1
    ''', (item_id,))
    old_data = cursor.fetchone()
    # 插入新数据
    insert_sql = '''
        INSERT INTO item_data 
        (item_id, title, price, coupon_price, monthly_sale, total_sale, good_rate, shop_score)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    '''
    data_tuple = (
        item_id,
        item_info.get("title", ""),
        float(item_info.get("price", 0)),
        float(item_info.get("coupon_price", 0)),
        int(item_info.get("monthly_sale", 0)),
        int(item_info.get("total_sale", 0)),
        float(item_info.get("good_rate", 0)),
        float(item_info.get("shop_score", 0))
    )
    cursor.execute(insert_sql, data_tuple)
    conn.commit()
    conn.close()
    return old_data

# 4. 数据对比，判断是否触发选品预警
def check_data_change(item_id, old_data, new_data):
    alert_msg_list = []
    if not old_data:
        alert_msg_list.append(f"【新品加入监控】商品ID:{item_id}，首次采集数据，开始持续跟踪")
        return alert_msg_list

    old_price, old_coupon, old_month_sale, old_total_sale = old_data
    new_price = float(new_data.get("price", 0))
    new_coupon = float(new_data.get("coupon_price", 0))
    new_month_sale = int(new_data.get("monthly_sale", 0))

    # 价格变动检测
    price_diff = abs(new_coupon - old_coupon)
    if price_diff >= PRICE_CHANGE_THRESHOLD:
        alert_msg_list.append(
            f"【价格异动】商品{item_id} 券后价变动{price_diff}元，原价{old_coupon}→现价{new_coupon}"
        )
    # 销量暴涨检测
    sale_rise = new_month_sale - old_month_sale
    if sale_rise >= SALE_RISE_THRESHOLD:
        alert_msg_list.append(
            f"【爆款预警】商品{item_id} 周期新增销量{sale_rise}，月销{new_month_sale}，具备爆款潜力"
        )
    return alert_msg_list

# 5. 推送告警消息
def send_alert(msg_list):
    if not ALERT_WEBHOOK or len(msg_list) == 0:
        return
    full_msg = "\n".join(msg_list)
    post_data = {
        "msgtype": "text",
        "text": {"content": f"淘宝选品监控告警：\n{full_msg}"}
    }
    try:
        requests.post(ALERT_WEBHOOK, json=post_data, timeout=10)
    except Exception as e:
        print("告警推送失败：", str(e))

# 6. 单次监控主逻辑
def monitor_task():
    print(f"===== 开始一轮商品监控，时间：{time.strftime('%Y-%m-%d %H:%M:%S')} =====")
    all_alert = []
    for item_id in MONITOR_ITEM_IDS:
        item_info = get_taobao_item_info(item_id)
        if not item_info:
            continue
        old_record = save_and_get_old_data(item_id, item_info)
        alert_texts = check_data_change(item_id, old_record, item_info)
        all_alert.extend(alert_texts)
        time.sleep(1)  # 接口请求间隔，避免高频调用
    send_alert(all_alert)
    print(f"本轮监控完成，共{len(all_alert)}条预警信息\n")

if __name__ == "__main__":
    init_db()
    # 定时循环任务
    schedule.every(SCAN_INTERVAL).minutes.do(monitor_task)
    print(f"监控程序已启动，每{SCAN_INTERVAL}分钟扫描一次竞品商品")
    while True:
        schedule.run_pending()
        time.sleep(10)