Open Claw 搭配淘宝 item\_get\_pro 接口，5 分钟搭建自动化竞品监控 & 爆款选品系统（完整可运行 Python 教程）

pip install requests open-claw python-dotenv

API_KEY=你的平台key
API_SECRET=你的平台secret
API_GATEWAY=https://open.onebound.cn/api/taobao/item_get_pro
# 监控配置
MONITOR_INTERVAL=300  # 轮询间隔，单位秒，5分钟查一次
PRICE_ALERT_THRESHOLD=1  # 价格变动超过1元触发预警
# 需要监控的商品ID列表，自行新增
MONITOR_ITEM_IDS=["723456789012","723456789013","723456789014"]

# -*- coding: utf-8 -*-
import os
import time
import requests
from dotenv import load_dotenv
from open_claw import ClawParser

# 加载环境变量
load_dotenv()
API_KEY = os.getenv("API_KEY")
API_SECRET = os.getenv("API_SECRET")
API_GATEWAY = os.getenv("API_GATEWAY")

# 初始化Open Claw解析器，自动标准化淘系商品数据
claw_parser = ClawParser(platform="taobao")

def get_single_item_detail(num_iid: str) -> dict:
    """
    调用taobao.item_get_pro接口获取商品完整数据
    :param num_iid: 淘宝商品ID
    :return: 结构化商品字典
    """
    # 组装请求参数
    req_params = {
        "key": API_KEY,
        "secret": API_SECRET,
        "num_iid": num_iid,
        "api_name": "item_get_pro",
        "result_type": "json"
    }
    try:
        resp = requests.get(url=API_GATEWAY, params=req_params, timeout=12)
        resp.raise_for_status()
        raw_data = resp.json()
        # 使用Open Claw清洗冗余字段，统一数据结构
        parse_result = claw_parser.parse_item(raw_data)
        return parse_result
    except requests.exceptions.RequestException as e:
        print(f"商品{num_iid}接口请求失败：{str(e)}")
        return {}

def extract_select_fields(item_data: dict) -> dict:
    """提取选品核心指标，过滤无用字段"""
    if not item_data:
        return {}
    return {
        "item_id": item_data.get("num_iid"),
        "title": item_data.get("title"),
        "current_price": float(item_data.get("price", 0)),
        "original_price": float(item_data.get("org_price", 0)),
        "month_sales": int(item_data.get("month_sales", 0)),
        "total_sales": int(item_data.get("sales", 0)),
        "stock": int(item_data.get("quantity", 0)),
        "good_rate": item_data.get("good_rate"),
        "location": item_data.get("location"),
        "shop_name": item_data.get("nick"),
        "online_status": item_data.get("is_online")
    }

# 单品测试入口
if __name__ == "__main__":
    test_item_id = "723456789012"
    full_data = get_single_item_detail(test_item_id)
    core_info = extract_select_fields(full_data)
    print("=====单品选品分析数据=====")
    for k, v in core_info.items():
        print(f"{k}: {v}")

# -*- coding: utf-8 -*-
import os
import time
from dotenv import load_dotenv
from tb_item_api import get_single_item_detail, extract_select_fields

load_dotenv()
MONITOR_INTERVAL = int(os.getenv("MONITOR_INTERVAL"))
PRICE_THRESHOLD = float(os.getenv("PRICE_ALERT_THRESHOLD"))
MONITOR_LIST = eval(os.getenv("MONITOR_ITEM_IDS"))

# 全局缓存：存储上一轮商品数据，用于对比异动
item_cache = {}

def alert_message(item_info: dict, alert_type: str, old_val, new_val):
    """异动预警输出，可扩展消息推送"""
    print("\n====================【监控预警】====================")
    print(f"商品ID：{item_info['item_id']} | 标题：{item_info['title']}")
    print(f"预警类型：{alert_type} | 历史值：{old_val} → 当前值：{new_val}")
    print("====================================================\n")

def monitor_loop():
    print("商品监控程序启动，监控列表：", MONITOR_LIST)
    global item_cache
    while True:
        for item_id in MONITOR_LIST:
            item_raw = get_single_item_detail(item_id)
            item_info = extract_select_fields(item_raw)
            if not item_info:
                continue
            # 首次采集，存入缓存不对比
            if item_id not in item_cache:
                item_cache[item_id] = item_info
                print(f"初始化商品{item_id}数据完成")
                continue
            # 读取历史数据
            old_info = item_cache[item_id]
            # 价格异动检测
            price_diff = abs(item_info["current_price"] - old_info["current_price"])
            if price_diff >= PRICE_THRESHOLD:
                alert_message(item_info, "价格大幅变动", old_info["current_price"], item_info["current_price"])
            # 销量暴涨检测（月销增长超50触发预警）
            sales_diff = item_info["month_sales"] - old_info["month_sales"]
            if sales_diff >= 50:
                alert_message(item_info, "月销量暴涨", old_info["month_sales"], item_info["month_sales"])
            # 商品下架检测
            if item_info["online_status"] != old_info["online_status"] and item_info["online_status"] == "false":
                alert_message(item_info, "商品已下架", old_info["online_status"], item_info["online_status"])
            # 更新缓存
            item_cache[item_id] = item_info
        print(f"本轮监控完成，等待{MONITOR_INTERVAL}秒进入下一轮...")
        time.sleep(MONITOR_INTERVAL)

if __name__ == "__main__":
    try:
        monitor_loop()
    except KeyboardInterrupt:
        print("监控程序手动停止，缓存数据已保存")

import csv
def save_to_csv(item_info):
    with open("tb_monitor_data.csv", "a", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=item_info.keys())
        writer.writerow(item_info)