📉 淘宝商品价格监控系统 — TOP API + APScheduler + 企业微信推送（Python）

# taobao_price_monitor.py
"""
淘宝商品价格监控系统
- 定时调 taobao.item.get(fields=price) 取一口价
- SQLite 存上次快照，变化幅度 ≥ threshold 推企业微信 Bot
- 支持 APScheduler 常驻 / 单次运行
依赖:
  pip install requests apscheduler
"""
import hashlib, time, requests, sqlite3, logging
from typing import Dict, Optional
from datetime import datetime

# ── 配置区 ──────────────────────────────────────
APP_KEY         = "YOUR_TOP_APP_KEY"
APP_SECRET      = "YOUR_TOP_APP_SECRET"
SANDBOX         = False                       # 生产=False
CHECK_INTERVAL  = 1800                        # 秒(30分钟)
THRESHOLD_PCT   = 0.05                        # 涨跌≥5%触发
WECOM_WEBHOOK   = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
WATCH_LIST      = [                           # 监控商品
    {"num_iid": "654321098765", "alias": "304不锈钢保温杯 500ml"},
    # {"num_iid": "612345678901", "alias": "其它商品"},
]
# ─────────────────────────────────────────────────
封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex

# ============ TOP Client (内联) ============
class TopClient:
    GW = "https://gw.api.taobao.com/router/rest"
    SB = "https://gw.api.tbsandbox.com/router/rest"

    def __init__(self, ak, ask, sandbox=False):
        self.ak, self.ask = ak, ask
        self.gw = self.SB if sandbox else self.GW

    def _sign(self, p: Dict) -> str:
        filt = sorted((k, v) for k, v in p.items()
                       if v is not None and str(v).strip() != '' and k != 'sign')
        qs = ''.join(f"{k}{v}" for k, v in filt)
        return hashlib.md5(f"{self.ask}{qs}{self.ask}".encode()
                          ).hexdigest().upper()

    def get_price(self, num_iid: str) -> float:
        resp = self._call("taobao.item.get", {
            "num_iid": num_iid,
            "fields": "num_iid,title,price,approve_status"
        })
        item = resp.get("item", {})
        if item.get("approve_status") != "onsale":
            raise ValueError(f"商品{num_iid} 已下架/仓库")
        return float(item.get("price") or 0)

    def _call(self, method, biz):
        p = {"method": method, "app_key": self.ak,
             "timestamp": str(int(time.time() * 1000)),
             "format": "json", "v": "2.0", "sign_method": "md5"}
        p.update(biz)
        p["sign"] = self._sign(p)
        r = requests.post(self.gw, data=p, timeout=15)
        r.raise_for_status()
        d = r.json()
        if "error_response" in d:
            err = d["error_response"]
            raise Exception(f"TOP[{err.get('code')}]: {err.get('msg')} {err.get('sub_msg','')}")
        return d.get(list(d.keys() - {"error_response"})[0], {})


# ============ SQLite 快照 ============
def init_db(db="price_monitor.db"):
    conn = sqlite3.connect(db)
    conn.execute("""CREATE TABLE IF NOT EXISTS snap(
        num_iid TEXT PRIMARY KEY, last_price REAL, updated TEXT)""")
    conn.commit()
    return conn


def load_price(conn, nid: str) -> Optional[float]:
    cur = conn.execute("SELECT last_price FROM snap WHERE num_iid=?", (nid,))
    row = cur.fetchone()
    return row[0] if row else None


def save_price(conn, nid: str, price: float):
    conn.execute("""INSERT INTO snap(num_iid,last_price,updated)
        VALUES(?,?,?) ON CONFLICT(num_iid) DO UPDATE SET
        last_price=excluded.last_price, updated=excluded.updated""",
        (nid, price, datetime.now().strftime("%Y-%m-%d %H:%M:%S")))
    conn.commit()


# ============ 企业微信推送 ============
def push_wecom(alias, old, new, nid):
    if not WECOM_WEBHOOK: return
    direction = "📉 降价" if new < old else "📈 涨价"
    pct = abs(new - old) / old * 100 if old else 0
    content = (f"## 🛒 淘宝商品价格变动\n"
               f"**商品**：{alias}\n"
               f"**ID**：`{nid}`\n"
               f"**{direction}**：¥{old:.2f} → **¥{new:.2f}**  (±{pct:.1f}%)\n"
               f"> {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    try:
        requests.post(WECOM_WEBHOOK, json={
            "msgtype": "markdown",
            "markdown": {"content": content}
        }, timeout=10)
    except Exception as e:
        logging.warning(f"推送失败: {e}")


# ============ 核心逻辑 ============
def monitor_once():
    cli = TopClient(APP_KEY, APP_SECRET, sandbox=SANDBOX)
    conn = init_db()
    for w in WATCH_LIST:
        nid = str(w["num_iid"])
        alias = w.get("alias") or nid
        try:
            cur = cli.get_price(nid)
            last = load_price(conn, nid)
            if last is None:
                save_price(conn, nid, cur)
                print(f"[INIT] {alias} 基准价 ¥{cur:.2f}")
                continue
            ratio = abs(cur - last) / last if last else 0
            if ratio >= THRESHOLD_PCT:
                print(f"[ALERT] {alias} ¥{last:.2f}→¥{cur:.2f} (Δ{ratio*100:.1f}%)")
                push_wecom(alias, last, cur, nid)
            else:
                print(f"[OK] {alias} 稳定 ¥{cur:.2f}")
            save_price(conn, nid, cur)
        except ValueError as ve:
            print(f"[WARN] {alias}: {ve}")
        except Exception as e:
            print(f"[ERR] {alias}: {e}")


# ============ 入口 ============
if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s")
    print(f"▶ 淘宝价格监控启动 interval={CHECK_INTERVAL}s threshold={THRESHOLD_PCT*100}%")
    monitor_once()           # 立即执行一次

    # ---- 常驻（取消注释）----
    # from apscheduler.schedulers.blocking import BlockingScheduler
    # sched = BlockingScheduler()
    # sched.add_job(monitor_once, 'interval', seconds=CHECK_INTERVAL, id='tb_price_mon')
    # try: sched.start()
    # except (KeyboardInterrupt, SystemExit): sched.shutdown()

使用步骤

1. pip install requests apscheduler

2. 填 APP_KEY/ APP_SECRET/ WECOM_WEBHOOK/ WATCH_LIST[num_iid]

3. 首次运行记录基准价，后续价格变动 ±5% 推企业微信群

4. 取消尾部注释 BlockingScheduler设为常驻服务（nohup python taobao_price_monitor.py &）

避坑