告别人工盯品！Open Claw 结合淘宝高级商品 API，一键搭建自动化监控选品系统（附完整可运行 Python 代码）

pip install requests schedule pandas

import requests
import json
import time
import pandas as pd
from schedule import every, run_pending

# ===================== 基础配置区（自行修改）=====================
API_GATEWAY = "https://api-gw.onebound.cn/taobao/item_get_pro"
API_KEY = "你的调用key"
API_SECRET = "你的调用secret"
# 待监控商品ID列表，批量添加多个商品实现同时监控
MONITOR_ITEM_IDS = ["520813250866"]
# 监控轮询周期，单位分钟
MONITOR_INTERVAL = 60
# 本地历史数据存储文件
HISTORY_FILE = "taobao_goods_monitor.csv"
# 价格下跌预警阈值，单位元，低于该数值触发标记
PRICE_ALERT_THRESHOLD = 30
# =================================================================

def get_taobao_goods_info(num_iid: str):
    """
    调用taobao.item_get_pro高级接口获取商品完整数据
    :param num_iid: 淘宝商品ID
    :return: 商品结构化字典数据
    """
    try:
        # 公共请求参数
        params = {
            "key": API_KEY,
            "secret": API_SECRET,
            "api_name": "item_get_pro",
            "num_iid": num_iid,
            "cache": "yes",
            "result_type": "json"
        }
        headers = {
            "Accept-Encoding": "gzip",
            "Connection": "close",
            "User-Agent": "OpenClaw-Monitor/1.0"
        }
        resp = requests.get(url=API_GATEWAY, params=params, headers=headers, timeout=15)
        resp.raise_for_status()
        res_data = resp.json()

        # 捕获接口异常
        if res_data.get("error") or res_data.get("error_code") != "0000":
            print(f"商品{num_iid}查询失败：{res_data.get('reason','未知错误')}")
            return None
        
        item_data = res_data.get("item", {})
        if not item_data:
            print(f"商品{num_iid}无商品数据，可能已下架")
            return None
        
        # 提取核心选品字段，精简冗余内容
        core_info = {
            "query_time": time.strftime("%Y-%m-%d %H:%M:%S"),
            "num_iid": item_data.get("num_iid"),
            "title": item_data.get("title"),
            "price": float(item_data.get("price", 0)),
            "original_price": float(item_data.get("orginal_price", 0)),
            "total_sales": int(item_data.get("total_sold", 0)),
            "stock": int(item_data.get("num", 0)),
            "shop_name": item_data.get("nick"),
            "shop_score": item_data.get("seller_info", {}).get("item_score", 0),
            "location": item_data.get("location"),
            "is_tmall": item_data.get("tmall"),
            "main_img": item_data.get("pic_url")
        }
        # SKU规格提取
        sku_list = item_data.get("skus", {}).get("sku", [])
        sku_text = ""
        for sku in sku_list:
            sku_text += f"{sku.get('properties_name')}|售价:{sku.get('price')}|库存:{sku.get('quantity')};"
        core_info["sku_detail"] = sku_text

        # 价格预警标记
        if core_info["price"] <= PRICE_ALERT_THRESHOLD:
            core_info["price_alert"] = "是"
        else:
            core_info["price_alert"] = "否"
        return core_info

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

def batch_monitor_goods():
    """批量执行商品监控，落地本地CSV历史数据"""
    print(f"\n===== 开始执行新一轮监控 {time.strftime('%Y-%m-%d %H:%M:%S')} =====")
    all_goods_data = []
    for item_id in MONITOR_ITEM_IDS:
        goods_data = get_taobao_goods_info(item_id)
        if goods_data:
            all_goods_data.append(goods_data)
            print(f"✅ 成功采集：{goods_data['title']} 当前价{goods_data['price']}元")
        # 间隔1秒防限流
        time.sleep(1)
    
    if len(all_goods_data) == 0:
        print("本轮未获取到有效商品数据")
        return
    
    # 写入本地历史文件，无文件则新建，有文件追加
    new_df = pd.DataFrame(all_goods_data)
    try:
        old_df = pd.read_csv(HISTORY_FILE)
        merge_df = pd.concat([old_df, new_df], ignore_index=True)
    except FileNotFoundError:
        merge_df = new_df
    merge_df.to_csv(HISTORY_FILE, index=False, encoding="utf-8-sig")
    print(f"📊 本轮{len(all_goods)}条数据已存入本地文件{HISTORY_FILE}")

    # 筛选预警商品输出提示
    alert_goods = new_df[new_df["price_alert"] == "是"]
    if len(alert_goods) > 0:
        print("\n⚠️ 检测到价格预警商品：")
        for _, row in alert_goods.iterrows():
            print(f"商品ID{row['num_iid']} {row['title']} 现价{row['price']}元低于预警线")

def start_openclaw_task():
    """对接Open Claw定时调度，循环执行监控任务"""
    # 设置定时任务：每MONITOR_INTERVAL分钟执行一次监控
    every(MONITOR_INTERVAL).minutes.do(batch_monitor_goods)
    print(f"监控任务已启动，每{MONITOR_INTERVAL}分钟自动采集一次商品数据，Ctrl+C停止")
    while True:
        run_pending()
        time.sleep(1)

if __name__ == "__main__":
    start_openclaw_task()