基于淘宝商品详情 API 的实时价格监控系统设计与实现

graph TD
    A[用户交互层] --> B[核心业务层]
    B --> C[数据持久层]
    B --> D[提醒通知层]
    B --> E[API调用层]
    E --> F[淘宝商品详情API]
    C --> G[本地数据库/SQLite]

# 核心依赖
pip install requests python-dotenv schedule sqlite3 smtplib beautifulsoup4

# API配置
TAOBAO_APP_KEY=你的AppKey
TAOBAO_APP_SECRET=你的AppSecret
TAOBAO_API_URL=https://eco.taobao.com/router/rest

# 邮件提醒配置
SMTP_SERVER=smtp.qq.com
SMTP_PORT=587
SMTP_USER=你的邮箱@qq.com
SMTP_PASS=邮箱授权码
TO_EMAIL=接收提醒的邮箱

# 系统配置
CHECK_INTERVAL=30  # 价格检查间隔（分钟）
DB_PATH=price_monitor.db

import sqlite3
import os
from dotenv import load_dotenv

# 加载配置
load_dotenv()
DB_PATH = os.getenv("DB_PATH")

def init_db():
    """初始化数据库，创建商品表和价格历史表"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    
    # 商品信息表
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS products (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        item_id TEXT UNIQUE NOT NULL,
        title TEXT,
        url TEXT,
        target_price FLOAT NOT NULL,
        create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
    ''')
    
    # 价格历史表
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS price_history (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        item_id TEXT NOT NULL,
        price FLOAT NOT NULL,
        check_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
        FOREIGN KEY (item_id) REFERENCES products (item_id)
    )
    ''')
    
    conn.commit()
    conn.close()

if __name__ == "__main__":
    init_db()
    print("数据库初始化完成")

import requests
import time
import hashlib
import json
from urllib.parse import urlencode, urlparse
import os
from dotenv import load_dotenv

load_dotenv()

# API配置
APP_KEY = os.getenv("TAOBAO_APP_KEY")
APP_SECRET = os.getenv("TAOBAO_APP_SECRET")
API_URL = os.getenv("TAOBAO_API_URL")

def get_sign(params, app_secret):
    """生成API签名（淘宝API要求）"""
    # 按参数名升序排序
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    # 拼接字符串
    sign_str = app_secret + ''.join([f"{k}{v}" for k, v in sorted_params]) + app_secret
    # MD5加密并转大写
    sign = hashlib.md5(sign_str.encode()).hexdigest().upper()
    return sign

def parse_item_id(url):
    """从淘宝商品链接解析商品ID"""
    parsed_url = urlparse(url)
    # 处理不同格式的商品链接
    if "item.taobao.com" in parsed_url.netloc:
        query_params = dict([i.split('=') for i in parsed_url.query.split('&') if '=' in i])
        return query_params.get('id')
    elif "taobao.com" in parsed_url.netloc:
        # 短链接处理
        return parsed_url.path.split('/')[-1]
    else:
        raise ValueError("无效的淘宝商品链接")

def get_item_detail(item_id):
    """调用淘宝商品详情API获取商品信息"""
    timestamp = str(int(time.time() * 1000))
    
    # 构造请求参数
    params = {
        'method': 'taobao.item.get',
        'app_key': APP_KEY,
        'format': 'json',
        'v': '2.0',
        'timestamp': timestamp,
        'sign_method': 'md5',
        'fields': 'num_iid,title,price,item_price,stock',
        'num_iid': item_id
    }
    
    # 生成签名
    params['sign'] = get_sign(params, APP_SECRET)
    
    try:
        response = requests.get(API_URL, params=params, timeout=10)
        result = response.json()
        
        if 'error_response' in result:
            raise Exception(f"API调用失败：{result['error_response']['msg']}")
        
        item = result['item_get_response']['item']
        return {
            'item_id': item['num_iid'],
            'title': item['title'],
            'current_price': float(item['price']),
            'stock': item.get('stock', 0)
        }
    except Exception as e:
        print(f"获取商品详情失败：{e}")
        return None

if __name__ == "__main__":
    # 测试
    item_url = "https://item.taobao.com/item.htm?id=1234567890"
    item_id = parse_item_id(item_url)
    print(f"解析的商品ID：{item_id}")
    
    detail = get_item_detail(item_id)
    if detail:
        print(f"商品信息：{json.dumps(detail, ensure_ascii=False, indent=2)}")

import schedule
import time
import sqlite3
import smtplib
from email.mime.text import MIMEText
from email.header import Header
from dotenv import load_dotenv
import os
from api_client import get_item_detail

load_dotenv()

# 邮件配置
SMTP_SERVER = os.getenv("SMTP_SERVER")
SMTP_PORT = int(os.getenv("SMTP_PORT"))
SMTP_USER = os.getenv("SMTP_USER")
SMTP_PASS = os.getenv("SMTP_PASS")
TO_EMAIL = os.getenv("TO_EMAIL")
CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL"))
DB_PATH = os.getenv("DB_PATH")

def send_email_alert(item_info, current_price):
    """发送价格提醒邮件"""
    subject = f"【价格提醒】{item_info['title']} 价格已低于阈值！"
    content = f"""
    <h3>商品价格提醒</h3>
    <p>商品名称：{item_info['title']}</p>
    <p>商品ID：{item_info['item_id']}</p>
    <p>目标价格：{item_info['target_price']} 元</p>
    <p>当前价格：{current_price} 元</p>
    <p>购买链接：<a href="{item_info['url']}">{item_info['url']}</a></p>
    <p>提醒时间：{time.strftime('%Y-%m-%d %H:%M:%S')}</p>
    """
    
    msg = MIMEText(content, 'html', 'utf-8')
    msg['From'] = Header(f"价格监控系统<{SMTP_USER}>", 'utf-8')
    msg['To'] = Header(TO_EMAIL, 'utf-8')
    msg['Subject'] = Header(subject, 'utf-8')
    
    try:
        server = smtplib.SMTP(SMTP_SERVER, SMTP_PORT)
        server.starttls()  # 开启TLS加密
        server.login(SMTP_USER, SMTP_PASS)
        server.sendmail(SMTP_USER, TO_EMAIL, msg.as_string())
        server.quit()
        print(f"提醒邮件已发送至 {TO_EMAIL}")
    except Exception as e:
        print(f"邮件发送失败：{e}")

def save_price_history(item_id, price):
    """保存价格历史数据"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute('''
    INSERT INTO price_history (item_id, price) VALUES (?, ?)
    ''', (item_id, price))
    conn.commit()
    conn.close()

def check_price_task():
    """定时检查商品价格任务"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    
    # 获取所有监控的商品
    cursor.execute('SELECT item_id, title, url, target_price FROM products')
    products = cursor.fetchall()
    conn.close()
    
    if not products:
        print("暂无监控商品")
        return
    
    for product in products:
        item_id, title, url, target_price = product
        print(f"正在检查商品：{title}（ID：{item_id}）")
        
        # 获取实时价格
        item_detail = get_item_detail(item_id)
        if not item_detail:
            continue
        
        current_price = item_detail['current_price']
        # 保存价格历史
        save_price_history(item_id, current_price)
        
        # 价格对比，触发提醒
        if current_price <= target_price:
            print(f"价格触发阈值！{title} 当前价格：{current_price} 元，目标价格：{target_price} 元")
            send_email_alert({
                'item_id': item_id,
                'title': title,
                'url': url,
                'target_price': target_price
            }, current_price)
        else:
            print(f"{title} 当前价格：{current_price} 元，高于目标价格 {target_price} 元")

def add_product(item_url, target_price):
    """添加监控商品"""
    from api_client import parse_item_id
    try:
        item_id = parse_item_id(item_url)
        # 获取商品标题
        item_detail = get_item_detail(item_id)
        if not item_detail:
            print("获取商品信息失败，添加失败")
            return False
        
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
        cursor.execute('''
        INSERT OR REPLACE INTO products (item_id, title, url, target_price)
        VALUES (?, ?, ?, ?)
        ''', (item_id, item_detail['title'], item_url, target_price))
        conn.commit()
        conn.close()
        print(f"商品添加成功：{item_detail['title']}")
        return True
    except Exception as e:
        print(f"商品添加失败：{e}")
        return False

if __name__ == "__main__":
    # 示例：添加商品并启动监控
    # add_product("https://item.taobao.com/item.htm?id=1234567890", 99.0)
    
    # 定时任务：每隔指定分钟检查一次价格
    schedule.every(CHECK_INTERVAL).minutes.do(check_price_task)
    
    print(f"价格监控系统已启动，每隔{CHECK_INTERVAL}分钟检查一次价格...")
    while True:
        schedule.run_pending()
        time.sleep(60)

from db_init import init_db
from price_monitor import add_product, check_price_task, schedule
import time
import os

def main():
    # 初始化数据库
    init_db()
    
    # 交互式添加商品
    while True:
        choice = input("请选择操作：1-添加监控商品 2-启动监控系统 0-退出\n")
        if choice == "1":
            item_url = input("请输入淘宝商品链接：")
            target_price = float(input("请输入目标价格（阈值）："))
            add_product(item_url, target_price)
        elif choice == "2":
            print("启动价格监控系统...")
            # 立即执行一次检查
            check_price_task()
            # 启动定时任务
            while True:
                schedule.run_pending()
                time.sleep(60)
        elif choice == "0":
            print("退出系统")
            break
        else:
            print("无效选择，请重新输入")

if __name__ == "__main__":
    main()