OpenClaw 对接淘宝商品 API，零爬虫风险搭建自动化选品监控系统（附完整可运行代码）

# 全局安装clawhub技能管理器
npm install -g clawhub
# 安装电商API对接技能、定时任务、消息推送三个核心插件
clawhub install ecommerce-api-task price-alert csv-storage

# OpenClaw定时监控任务配置
task_name: taobao_goods_monitor
cron: "0 */2 * * *" # 每2小时轮询一次商品池，可自定义定时规则
api:
  url: "https://api域名/taobao/item_get_pro"
  appkey: "替换为你的appkey"
  appsecret: "替换为你的appsecret"
goods_pool:
  # 批量填入需要监控的淘宝商品ID，可批量扩充
  - "728956123456"
  - "659874123698"
  - "523698741258"
filter_rule:
  min_sale: 300
  max_sale: 3000
  max_comment: 200
  min_price: 19.9
  max_price: 99
alert:
  channel: dingtalk
  webhook: "钉钉机器人推送链接"
storage:
  save_path: "./taobao_goods_data.csv"

import requests
import csv
import json
from datetime import datetime

# 接口基础配置
API_URL = "https://api域名/taobao/item_get_pro"
APPKEY = "你的appkey"
APPSECRET = "你的appsecret"
# 待监控商品ID列表
GOODS_ID_LIST = ["728956123456","659874123698","523698741258"]
# 选品筛选规则
RULE = {
    "min_sale":300,"max_sale":3000,"max_comment":200,"min_price":19.9,"max_price":99
}
# 存储文件
CSV_FILE = "./taobao_goods_data.csv"

# 初始化CSV表头
def init_csv():
    headers = ["抓取时间","商品ID","商品标题","售价","月销量","评价数","发货地","是否新品","运费险"]
    with open(CSV_FILE,"w",encoding="utf-8-sig",newline="") as f:
        writer = csv.writer(f)
        writer.writerow(headers)

# 调用item_get_pro接口获取商品详情
def get_taobao_goods(num_iid):
    params = {
        "appkey":APPKEY,
        "secret":APPSECRET,
        "num_iid":num_iid
    }
    try:
        res = requests.get(API_URL,params=params,timeout=15)
        data = res.json()
        if data.get("code") != 0:
            print(f"商品{num_iid}接口异常：{data.get('msg')}")
            return None
        return data["result"]
    except Exception as e:
        print(f"请求失败：{str(e)}")
        return None

# 蓝海商品筛选逻辑
def filter_blue_ocean(goods_info):
    price = float(goods_info.get("price",0))
    month_sale = int(goods_info.get("month_sold",0))
    comment_num = int(goods_info.get("comment_count",0))
    # 匹配预设选品条件
    if (RULE["min_sale"]<=month_sale<=RULE["max_sale"] and
        RULE["min_price"]<=price<=RULE["max_price"] and comment_num<=RULE["max_comment"]):
        return True
    return False

# 主执行函数
def run_monitor():
    init_csv()
    alert_list = [] # 存储符合条件的蓝海商品，用于推送告警
    now_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    for gid in GOODS_ID_LIST:
        goods = get_taobao_goods(gid)
        if not goods:
            continue
        title = goods.get("title","无标题")
        price = goods.get("price","0")
        sale = goods.get("month_sold","0")
        comment = goods.get("comment_count","0")
        ship_area = goods.get("from","未知")
        is_new = goods.get("is_new_item","否")
        has_freight = "是" if goods.get("free_shipping_insurance") ==1 else "否"
        # 写入全量数据
        row = [now_time,gid,title,price,sale,comment,ship_area,is_new,has_freight]
        with open(CSV_FILE,"a",encoding="utf-8-sig",newline="") as f:
            csv.writer(f).writerow(row)
        # 筛选蓝海款
        if filter_blue_ocean(goods):
            alert_list.append(f"【蓝海新品】{title}|售价：{price}|月销：{sale}|商品ID：{gid}")
    # 触发OpenClaw消息推送
    if alert_list:
        alert_content = "\n".join(alert_list)
        from clawhub import send_dingtalk
        send_dingtalk(alert_content)
        print("发现蓝海商品，已推送钉钉！")

if __name__ == "__main__":
    run_monitor()

# 加载配置文件，启动自动选品监控任务
openclaw task start taobao_goods_monitor --config openclaw_config.yaml
# 查看运行日志
openclaw task log taobao_goods_monitor