告别手动翻页！基于淘宝商品接口 + Open Claw 实现自动化选品与实时监控（附完整 Python 代码）

pip install requests open-claw json5

# 导入所需库
import time
import requests
from open_claw import ClawParser  # 导入Open Claw解析工具

# ===================== 配置区（自行修改参数）=====================
APP_KEY = "你的appkey"       # 替换为你的接口密钥
APP_SECRET = "你的secret"   # 替换为你的加密密钥
API_URL = "https://open.onebound.cn/taobao/item_get_pro"  # 淘宝商品接口地址
REQUEST_INTERVAL = 1        # 请求间隔（秒），防限流
# =================================================================

def get_taobao_item_info(item_id: str) -> dict:
    """
    调用淘宝商品接口，获取单品完整数据
    :param item_id: 淘宝商品ID
    :return: 解析后的商品字典数据
    """
    # 构造接口请求参数
    params = {
        "appkey": APP_KEY,
        "secret": APP_SECRET,
        "itemid": item_id,
        "format": "json",
        "lang": "zh-CN"
    }

    try:
        # 发起接口请求
        response = requests.get(url=API_URL, params=params, timeout=10)
        response.raise_for_status()  # 捕获HTTP请求异常

        # 使用Open Claw清洗、解析返回数据，过滤无效字符
        claw = ClawParser()
        raw_data = claw.clean_html(response.text)  # 清洗多余标签、特殊符号
        item_data = claw.parse_json(raw_data)      # 解析JSON数据

        # 接口状态校验
        if item_data.get("status") != 0:
            print(f"接口请求失败：{item_data.get('msg', '未知错误')}")
            return {}

        # 提取核心商品数据（选品重点字段）
        item_info = item_data.get("data", {})
        result = {
            "商品ID": item_id,
            "商品标题": item_info.get("title", ""),
            "当前售价": item_info.get("price", ""),
            "原价": item_info.get("original_price", ""),
            "累计销量": item_info.get("sales", 0),
            "库存数量": item_info.get("stock", 0),
            "所属类目": item_info.get("category_name", ""),
            "发货地": item_info.get("location", ""),
            "好评数": item_info.get("good_rate", ""),
            "商品链接": item_info.get("item_url", "")
        }
        return result

    except requests.exceptions.Timeout:
        print("请求超时，请检查网络或接口地址")
        return {}
    except Exception as e:
        print(f"数据解析异常：{str(e)}")
        return {}

# 主程序运行
if __name__ == "__main__":
    # 替换为你需要查询的淘宝商品ID
    target_item_id = "68xxxxxxxxx"
    print("========== 开始抓取商品数据 ==========")
    item_result = get_taobao_item_info(target_item_id)
    if item_result:
        # 格式化输出选品数据
        for key, value in item_result.items():
            print(f"{key}：{value}")
    time.sleep(REQUEST_INTERVAL)

import time
from open_claw import ClawParser
import requests

# ===================== 全局配置（按需修改）=====================
APP_KEY = "你的appkey"
APP_SECRET = "你的secret"
API_URL = "https://open.onebound.cn/taobao/item_get_pro"
MONITOR_LIST = [
    "68xxxxxxxxx",  # 监控商品ID1
    "69xxxxxxxxx",  # 监控商品ID2
    "70xxxxxxxxx"   # 监控商品ID3
]
MONITOR_CYCLE = 300  # 监控周期（秒），此处设置为5分钟轮询一次
PRICE_CHANGE_THRESHOLD = 0.5  # 价格异动阈值（元），超过则预警
SALES_JUMP_THRESHOLD = 100    # 销量突增阈值，增量超过则预警
# =================================================================

# 全局字典：存储上一轮商品数据，用于对比异动
last_item_data = {}
claw = ClawParser()  # 初始化Open Claw解析器

def fetch_single_item(item_id: str) -> dict:
    """获取单个商品原始数据"""
    params = {
        "appkey": APP_KEY,
        "secret": APP_SECRET,
        "itemid": item_id,
        "format": "json"
    }
    try:
        resp = requests.get(API_URL, params=params, timeout=10)
        clean_data = claw.clean_html(resp.text)
        parse_data = claw.parse_json(clean_data)
        if parse_data.get("status") != 0:
            return {}
        return parse_data.get("data", {})
    except:
        return {}

def check_data_change(item_id: str, current_data: dict):
    """对比新旧数据，检测价格、销量、库存异动"""
    global last_item_data
    # 首次监控，初始化历史数据，不做对比
    if item_id not in last_item_data:
        last_item_data[item_id] = {
            "price": current_data.get("price", 0),
            "sales": current_data.get("sales", 0),
            "stock": current_data.get("stock", 0)
        }
        print(f"【初始化监控】商品{item_id} 数据录入完成")
        return

    # 读取上一轮数据
    old_data = last_item_data[item_id]
    current_price = float(current_data.get("price", 0))
    old_price = float(old_data.get("price", 0))
    current_sales = int(current_data.get("sales", 0))
    old_sales = int(old_data.get("sales", 0))
    current_stock = int(current_data.get("stock", 0))
    old_stock = int(old_data.get("stock", 0))

    # 1. 价格异动检测
    price_diff = abs(current_price - old_price)
    if price_diff >= PRICE_CHANGE_THRESHOLD:
        if current_price > old_price:
            print(f"⚠️【价格上涨预警】商品{item_id} | 原价格：{old_price}元 → 现价格：{current_price}元")
        else:
            print(f"⚠️【价格下跌预警】商品{item_id} | 原价格：{old_price}元 → 现价格：{current_price}元")

    # 2. 销量突增检测（爆款预警）
    sales_diff = current_sales - old_sales
    if sales_diff >= SALES_JUMP_THRESHOLD:
        print(f"🔥【爆款预警】商品{item_id} 销量突增！新增销量：{sales_diff} | 总销量：{current_sales}")

    # 3. 库存预警（断货风险）
    stock_diff = old_stock - current_stock
    if stock_diff > 50 and current_stock < 200:
        print(f"📦【库存预警】商品{item_id} 库存紧张！剩余库存：{current_stock}")

    # 更新历史数据
    last_item_data[item_id] = {
        "price": current_price,
        "sales": current_sales,
        "stock": current_stock
    }

def main_monitor():
    """主监控循环"""
    print("===== 淘宝商品自动化监控已启动 =====")
    print(f"监控商品列表：{MONITOR_LIST}")
    print(f"轮询周期：{MONITOR_CYCLE}秒\n")
    while True:
        for item_id in MONITOR_LIST:
            item_data = fetch_single_item(item_id)
            if item_data:
                check_data_change(item_id, item_data)
            time.sleep(1)  # 单个商品请求间隔
        print(f"\n本轮监控完成，等待{MONITOR_CYCLE}秒后下一轮...\n")
        time.sleep(MONITOR_CYCLE)

if __name__ == "__main__":
    main_monitor()