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

pip install requests  # 网络请求库，用于调用API
pip install json      # 内置库，无需额外安装，解析返回数据

import requests
import json

# ==================== 基础配置（自行修改）====================
# 接口请求地址
API_URL = "https://open.onebound.cn/taobao/item_get_pro"
# 你的接口密钥（替换为自己的Appkey）
APP_KEY = "你的Appkey"
# 待监控的淘宝商品ID（可在商品链接中获取）
ITEM_ID = "789654123000"
# ==========================================================

def get_taobao_item_info(item_id):
    """
    调用淘宝商品详情API，获取商品数据
    :param item_id: 淘宝商品ID
    :return: 结构化商品数据/None（调用失败）
    """
    # 组装请求参数
    params = {
        "key": APP_KEY,
        "itemid": item_id,
        "format": "json"  # 指定返回格式为JSON
    }

    try:
        # 发送GET请求调用接口
        response = requests.get(url=API_URL, params=params, timeout=10)
        # 状态码判断
        if response.status_code != 200:
            print(f"接口请求失败，状态码：{response.status_code}")
            return None
        
        # 解析JSON数据
        res_data = json.loads(response.text)
        # 判断接口返回状态
        if res_data.get("code") != 200:
            print(f"数据获取失败：{res_data.get('msg', '未知错误')}")
            return None
        
        # 提取核心商品数据
        item_data = res_data.get("data", {})
        return item_data

    except Exception as e:
        print(f"请求异常：{str(e)}")
        return None

# 主程序执行
if __name__ == "__main__":
    item_result = get_taobao_item_info(ITEM_ID)
    if item_result:
        print("===== 淘宝商品核心选品数据 =====")
        print(f"商品标题：{item_result.get('title')}")
        print(f"现价：{item_result.get('price')} 元")
        print(f"原价：{item_result.get('org_price')} 元")
        print(f"累计销量：{item_result.get('sales')}")
        print(f"库存数量：{item_result.get('stock')}")
        print(f"商品类目：{item_result.get('category')}")
        print(f"商品状态：{'正常在售' if item_result.get('status') == '1' else '下架/违规'}")
    else:
        print("未获取到商品数据，请检查商品ID和接口密钥")

import requests
import json
import time

# ==================== 自定义配置区（重点修改）====================
API_URL = "https://open.onebound.cn/taobao/item_get_pro"
APP_KEY = "你的Appkey"
# 批量监控的商品ID列表，可无限添加
MONITOR_ITEM_LIST = [
    "789654123000",
    "123456789000",
    "456789123000"
]
# 监控规则配置
PRICE_CHANGE_RATIO = 0.1  # 价格波动阈值：涨跌超过10%触发告警
SALES_SURGE_THRESHOLD = 500  # 单次轮询销量增加500+判定为爆款苗头
# 本地缓存历史数据（简单内存缓存，重启后清空，进阶可接入本地文件/数据库）
history_data = {}
# =================================================================

def get_single_item(item_id):
    """获取单个商品数据"""
    params = {"key": APP_KEY, "itemid": item_id, "format": "json"}
    try:
        res = requests.get(API_URL, params=params, timeout=10)
        data = json.loads(res.text)
        if data.get("code") == 200:
            return data.get("data", {})
        return None
    except:
        return None

def monitor_rule_check(item_id, current_data):
    """
    选品监控规则校验
    :param item_id: 商品ID
    :param current_data: 当前商品数据
    :return: 告警信息列表
    """
    alert_list = []
    current_price = float(current_data.get("price", 0))
    current_sales = int(current_data.get("sales", 0))
    item_status = current_data.get("status")
    item_title = current_data.get("title", "未知商品")

    # 规则1：商品下架/违规告警
    if item_status != "1":
        alert_list.append(f"【告警】商品《{item_title}》已下架或违规！ID：{item_id}")

    # 规则2：对比历史数据，判断价格、销量异常
    if item_id in history_data:
        old_data = history_data[item_id]
        old_price = float(old_data.get("price", 0))
        old_sales = int(old_data.get("sales", 0))

        # 价格波动检测
        if old_price > 0:
            price_diff = abs(current_price - old_price) / old_price
            if price_diff > PRICE_CHANGE_RATIO:
                alert_list.append(
                    f"【价格异动】商品《{item_title}》价格波动超{PRICE_CHANGE_RATIO*100}% "
                    f"原价：{old_price}元，现价：{current_price}元"
                )

        # 销量暴涨检测（爆款挖掘）
        sales_increase = current_sales - old_sales
        if sales_increase >= SALES_SURGE_THRESHOLD:
            alert_list.append(
                f"【爆款预警】商品《{item_title}》销量激增！本轮新增销量：{sales_increase}"
            )

    # 更新历史数据
    history_data[item_id] = current_data
    return alert_list

def batch_monitor():
    """批量商品监控主函数（OpenClaw定时调用入口）"""
    print(f"\n========== 开始批量选品监控 时间：{time.strftime('%Y-%m-%d %H:%M:%S')} ==========")
    total_alert = []

    for item_id in MONITOR_ITEM_LIST:
        item_info = get_single_item(item_id)
        if not item_info:
            total_alert.append(f"【异常】商品ID：{item_id} 数据获取失败")
            continue
        
        # 执行规则校验
        alerts = monitor_rule_check(item_id, item_info)
        total_alert.extend(alerts)

    # 统一输出所有告警信息
    if total_alert:
        for alert in total_alert:
            print(alert)
    else:
        print("本次监控：所有商品数据正常，无异常告警")

# 程序入口（OpenClaw直接调用此函数即可）
if __name__ == "__main__":
    # 单次运行批量监控
    batch_monitor()