未命名

# 一键安装依赖
pip install requests

import requests
import json
import time

# 接口基础配置
API_HOST = "https://open.onebound.cn/api/taobao/item_get_pro"
APP_KEY = "你的专属appkey"  # 替换自己密钥

def get_taobao_item_detail(num_iid: str) -> dict:
    """
    调用淘宝item_get_pro接口获取商品完整数据
    :param num_iid: 商品ID
    :return: 商品结构化字典
    """
    params = {
        "key": APP_KEY,
        "num_iid": num_iid,
        "get_same": 1,      # 开启同款商品查询
        "get_review": 1,    # 拉取评价数据
        "get_sku": 1        # 获取全部SKU价格
    }
    try:
        resp = requests.get(url=API_HOST, params=params, timeout=15)
        resp_data = resp.json()
        # 判断接口返回状态
        if resp_data.get("code") != 0:
            print(f"商品{num_iid}采集失败：{resp_data.get('msg')}")
            return {}
        item_info = resp_data.get("result", {})
        return item_info
    except Exception as e:
        print(f"接口请求异常：{str(e)}")
        return {}

if __name__ == "__main__":
    # 测试单商品采集
    test_item_id = "723456123456"
    res = get_taobao_item_detail(test_item_id)
    print(json.dumps(res, ensure_ascii=False, indent=2))

import sqlite3
import datetime

def init_db():
    # 创建商品监控数据表
    conn = sqlite3.connect("taobao_monitor.db")
    cursor = conn.cursor()
    create_sql = """
    CREATE TABLE IF NOT EXISTS item_monitor (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        item_id TEXT NOT NULL,
        title TEXT,
        price REAL,
        min_price REAL,
        sales INTEGER,
        review_rate REAL,
        shop_name TEXT,
        collect_time DATETIME
    )
    """
    cursor.execute(create_sql)
    conn.commit()
    conn.close()

def save_item_data(item_data: dict):
    """存储单条商品采集数据"""
    if not item_data:
        return
    conn = sqlite3.connect("taobao_monitor.db")
    cursor = conn.cursor()
    item_id = item_data.get("num_iid", "")
    title = item_data.get("title", "")
    price = float(item_data.get("price", 0))
    min_price = float(item_data.get("min_price", 0))
    sales = int(item_data.get("sales", 0))
    review_rate = float(item_data.get("good_review_rate", 0))
    shop_name = item_data.get("shop_info", {}).get("shop_name", "")
    now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    insert_sql = """
    INSERT INTO item_monitor
    (item_id, title, price, min_price, sales, review_rate, shop_name, collect_time)
    VALUES (?,?,?,?,?,?,?,?)
    """
    cursor.execute(insert_sql, (item_id, title, price, min_price, sales, review_rate, shop_name, now))
    conn.commit()
    conn.close()

def get_last_record(item_id: str):
    """查询商品上一次采集记录，用于对比异动"""
    conn = sqlite3.connect("taobao_monitor.db")
    cursor = conn.cursor()
    cursor.execute("""
        SELECT price, sales FROM item_monitor
        WHERE item_id = ? ORDER BY collect_time DESC LIMIT 1
    """, (item_id,))
    res = cursor.fetchone()
    conn.close()
    return res if res else None

from taobao_api import get_taobao_item_detail
from db_storage import init_db, save_item_data, get_last_record
import time

# 待监控商品ID列表，批量添加竞品
MONITOR_ITEM_LIST = [
    "723456123456",
    "723456789012",
    "723456987654"
]

# 选品阈值配置（可自行调整）
PRICE_DOWN_ALERT = 0.08  # 价格下跌8%触发告警
SALES_UP_THRESHOLD = 30  # 单次销量上涨30+标记潜力款
LOW_COMPETITION_SALES = 200  # 月销低于200判定蓝海小类目

def analysis_item(item_data: dict):
    """商品数据筛选，输出优质选品标记"""
    sales = int(item_data.get("sales", 0))
    review_rate = float(item_data.get("good_review_rate", 0))
    min_price = float(item_data.get("min_price", 0))
    tips = []
    # 蓝海品判定
    if sales < LOW_COMPETITION_SALES and review_rate > 0.9:
        tips.append("【蓝海潜力款】低销量+高好评，竞争小")
    # 低价爆款判定
    if sales > 1000 and min_price < 30:
        tips.append("【低价爆款】高销量低客单，适合铺货")
    # 高利润款
    original_price = float(item_data.get("original_price", 0))
    if original_price > 0 and (original_price - min_price) / original_price > 0.4:
        tips.append("【高利润款】溢价空间超40%")
    return tips

def monitor_single_item(item_id: str):
    """单商品监控流程"""
    item_info = get_taobao_item_detail(item_id)
    if not item_info:
        return
    # 存入数据库
    save_item_data(item_info)
    # 获取上次数据对比
    last_data = get_last_record(item_id)
    curr_price = float(item_info.get("min_price", 0))
    curr_sales = int(item_info.get("sales", 0))
    alert_msg = []
    if last_data:
        last_price, last_sales = last_data
        # 价格下跌告警
        if last_price > 0 and (last_price - curr_price) / last_price >= PRICE_DOWN_ALERT:
            alert_msg.append(f"价格异动：原价{last_price}，现价{curr_price}，降价幅度超8%")
        # 销量暴涨告警
        if curr_sales - last_sales >= SALES_UP_THRESHOLD:
            alert_msg.append(f"销量爆发：新增销量{curr_sales - last_sales}，市场需求上涨")
    # 选品标签
    select_tips = analysis_item(item_info)
    # 输出监控日志，OpenClaw可对接企业微信/钉钉推送消息
    print("="*60)
    print(f"商品：{item_info.get('title')}")
    print(f"当前售价：{curr_price} 月销量：{curr_sales} 好评率：{item_info.get('good_review_rate')}")
    if alert_msg:
        print("【异动告警】")
        for msg in alert_msg:
            print(msg)
    if select_tips:
        print("【选品标签】")
        for tip in select_tips:
            print(tip)
    print("="*60 + "\n")

def run_monitor_task():
    """批量监控入口，对接OpenClaw定时调度"""
    init_db()
    print("=====淘宝竞品监控任务启动=====")
    for item_id in MONITOR_ITEM_LIST:
        monitor_single_item(item_id)
        time.sleep(2)  # 接口请求间隔，避免超限

if __name__ == "__main__":
    run_monitor_task()