OpenClaw 对接淘宝商品 API，低成本实现全天候选品监控｜附可运行 Python 实操代码

# 安装接口请求、数据处理、表格存储依赖
pip install requests pandas openpyxl
# OpenClaw一键安装命令（官方安装脚本）
npx clawhub@latest install EcommerceCrawlSkill PriceMonitorSkill

openclaw config set bind_ip 127.0.0.1
openclaw restart

import requests
import pandas as pd
import json
from datetime import datetime

# ==========用户自定义配置区【只改这里参数即可】==========
APP_KEY = "你的接口申请的appkey"  # 替换成自己密钥
API_URL = "https://api.open.onebound.cn/taobao/item_get_pro"
# 待监控商品ID列表，可手动填写，也可从excel批量读取
ITEM_ID_LIST = ["725896325412", "693258741259", "598741236985"]
SAVE_EXCEL_PATH = "./淘宝选品监控数据表.xlsx"  # 数据保存路径
# ======================================================

def get_taobao_item_info(item_id):
    """调用item_get_pro高级接口，获取单品全量详情数据"""
    params = {
        "appkey": APP_KEY,
        "item_id": item_id,
        "lang": "zh-CN"
    }
    try:
        resp = requests.get(url=API_URL, params=params, timeout=15)
        result = resp.json()
        # 接口异常捕获
        if result.get("code") != 0:
            print(f"商品{item_id}查询失败：{result.get('msg')}")
            return None
        item_data = result["data"]
        # 筛选选品核心字段，剔除无用冗余数据
        filter_data = {
            "抓取时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "商品ID": item_id,
            "商品标题": item_data.get("title", ""),
            "现价": item_data.get("price", 0),
            "原价": item_data.get("orig_price", 0),
            "月销量": item_data.get("month_sold", 0),
            "总销量": item_data.get("total_sold", 0),
            "佣金比例%": item_data.get("tk_rate", 0),
            "好评率": item_data.get("good_rate", 0),
            "店铺评分": item_data.get("shop_dsr", ""),
            "优惠券金额": item_data.get("coupon_price", 0),
            "商品链接": item_data.get("item_url", "")
        }
        return filter_data
    except Exception as e:
        print(f"接口请求异常：{str(e)}")
        return None

def batch_get_item():
    """批量遍历商品ID，全量采集数据并保存Excel"""
    save_list = []
    for item in ITEM_ID_LIST:
        item_info = get_taobao_item_info(item)
        if item_info:
            save_list.append(item_info)
    # 数据写入Excel，追加存储历史数据
    if save_list:
        df_new = pd.DataFrame(save_list)
        try:
            df_old = pd.read_excel(SAVE_EXCEL_PATH)
            df_all = pd.concat([df_old, df_new], ignore_index=True)
        except:
            df_all = df_new
        df_all.to_excel(SAVE_EXCEL_PATH, index=False)
        print(f"本轮成功采集{len(save_list)}个商品，数据已存入表格")
    return save_list

if __name__ == "__main__":
    batch_get_item()

from openclaw import scheduler  # OpenClaw内置定时调度模块
from taobao_item_api import batch_get_item
import pandas as pd

# ==========自定义选品阈值（按需修改）==========
MIN_MONTH_SOLD = 300    # 最低月销
MAX_MONTH_SOLD = 5000   # 最高月销（避开头部红海爆款）
MIN_COMMISSION = 15     # 最低淘客佣金%
MIN_GOOD_RATE = 0.92    # 最低好评率
DISCOUNT_RATE = 0.6     # 低于原价6折
GOOD_ITEM_SAVE = "./潜力爆款清单.xlsx"
# ============================================

def select_good_product():
    """筛选符合条件的潜力商品"""
    item_data = batch_get_item()
    good_item_list = []
    for data in item_data:
        # 计算折扣：现价/原价
        try:
            discount = float(data["现价"]) / float(data["原价"]) if float(data["原价"]) > 0 else 1
        except:
            discount = 1
        # 选品逻辑判断
        if (MIN_MONTH_SOLD <= int(data["月销量"]) <= MAX_MONTH_SOLD
                and float(data["佣金比例%"]) >= MIN_COMMISSION
                and float(data["好评率"]) >= MIN_GOOD_RATE
                and discount <= DISCOUNT_RATE):
            good_item_list.append(data)
            print(f"✅挖到潜力品：{data['商品标题']}，商品ID：{data['商品ID']}")
    # 优质商品单独存表
    if good_item_list:
        df_good = pd.DataFrame(good_item_list)
        try:
            df_history = pd.read_excel(GOOD_ITEM_SAVE)
            df_final = pd.concat([df_history, df_good], ignore_index=True)
        except:
            df_final = df_good
        df_final.to_excel(GOOD_ITEM_SAVE, index=False)
        # OpenClaw内置消息推送，可对接钉钉/企业微信
        scheduler.msg_send(title="新发现淘宝潜力选品", content=f"本轮筛选出{len(good_item_list)}个优质货源，请查看表格")

# OpenClaw配置定时：每4小时自动执行一次全量监控
scheduler.add_job(
    func=select_good_product,
    trigger="interval",
    hours=4,
    id="taobao_product_monitor",
    replace_existing=True
)

# 启动定时任务
if __name__ == "__main__":
    print("OpenClaw淘宝选品监控服务启动成功，每4小时自动轮询筛选...")
    scheduler.start()