实操教程｜OpenClaw 对接淘宝商品 API，低成本搭建全天候监控选品系统

pip install openclaw requests pandas schedule

# -*- coding: utf-8 -*-
import time
import schedule
import pandas as pd
from openclaw import OpenClawClient

# ===================== 配置区（自行修改）=====================
APP_KEY = "你的appkey"
APP_SECRET = "你的appsecret"
API_URL = "https://api.onebound.cn/taobao/item_get_pro"
# 需要监控的商品ID列表，可批量导入
MONITOR_ITEM_IDS = [
    "723456123456",
    "723456789012",
    "723456345678"
]
# 选品筛选阈值
SALE_THRESHOLD = 300  # 单日销量大于300标记潜力款
PRICE_DIFF_RATE = 0.2  # 价格跌幅超过20%触发告警
POLL_INTERVAL = 60  # 监控轮询间隔，单位分钟
# ==========================================================

# 初始化OpenClaw客户端
claw_client = OpenClawClient(
    app_key=APP_KEY,
    app_secret=APP_SECRET,
    timeout=10,
    retry_times=2
)

def get_taobao_item_detail(item_id):
    """调用taobao.item_get_pro接口获取商品完整数据"""
    params = {
        "api": "taobao.item_get_pro",
        "num_iid": item_id,
        "appkey": APP_KEY,
        "appsecret": APP_SECRET
    }
    try:
        # OpenClaw统一发送请求，自动处理签名校验
        resp = claw_client.get(url=API_URL, params=params)
        res_json = resp.json()
        if res_json.get("code") != 0:
            print(f"商品{item_id}请求失败：{res_json.get('msg')}")
            return None
        return res_json.get("result", {})
    except Exception as e:
        print(f"接口请求异常{item_id}: {str(e)}")
        return None

def filter_potential_product(item_data):
    """选品筛选逻辑，返回是否为潜力爆款"""
    if not item_data:
        return False, {}
    title = item_data.get("title", "")
    real_price = float(item_data.get("price", 0))
    origin_price = float(item_data.get("original_price", real_price))
    sales = int(item_data.get("sales", 0))
    is_new = item_data.get("is_new_item", False)
    shop_type = item_data.get("shop_type")

    # 价格跌幅计算
    discount_rate = (origin_price - real_price) / origin_price if origin_price > 0 else 0
    alert_msg = ""

    # 筛选规则1：销量达标新品
    if sales >= SALE_THRESHOLD and is_new:
        alert_msg += f"【潜力新品】销量{sales}，新品标"
    # 筛选规则2：大幅降价竞品
    if discount_rate >= PRICE_DIFF_RATE:
        alert_msg += f"【降价告警】原价{origin_price},现价{real_price},降幅{round(discount_rate*100,1)}%"
    # 筛选规则3：天猫高销量商品
    if shop_type == "tmall" and sales >= SALE_THRESHOLD*0.8:
        alert_msg += "【天猫爆款】"

    flag = len(alert_msg) > 0
    item_info = {
        "item_id": item_data.get("num_iid"),
        "title": title,
        "shop_type": shop_type,
        "real_price": real_price,
        "origin_price": origin_price,
        "sales": sales,
        "is_new": is_new,
        "discount_rate": round(discount_rate,3),
        "alert": alert_msg,
        "crawl_time": time.strftime("%Y-%m-%d %H:%M:%S")
    }
    return flag, item_info

def monitor_task():
    """定时监控主任务"""
    print(f"\n===== 开始执行商品监控任务 {time.strftime('%Y-%m-%d %H:%M:%S')} =====")
    all_item_data = []
    alert_list = []

    for item_id in MONITOR_ITEM_IDS:
        item_detail = get_taobao_item_detail(item_id)
        if not item_detail:
            continue
        is_hot, info = filter_potential_product(item_detail)
        all_item_data.append(info)
        if is_hot:
            alert_list.append(info)
            print(f"🔥 预警商品：{info['title']} | {info['alert']}")
        # 简单限流，避免接口高频调用
        time.sleep(1.2)

    # 数据持久化写入Excel
    df = pd.DataFrame(all_item_data)
    df.to_excel("淘宝商品监控记录表.xlsx", index=False, mode="a", header=False)
    print(f"本次共抓取{len(all_item_data)}款商品，预警商品{len(alert_list)}款，数据已存入本地表格")

if __name__ == "__main__":
    # 配置定时任务
    schedule.every(POLL_INTERVAL).minutes.do(monitor_task)
    print(f"监控程序启动成功，每{POLL_INTERVAL}分钟自动执行一次抓取")
    # 循环常驻运行
    while True:
        schedule.run_pending()
        time.sleep(10)