用Python做商品价格监控：降价自动提醒，再也不怕买贵了

项目背景

你有没有这种经历：看中一个商品，加了购物车等了半个月，结果某天发现它降价了——而且降了不少。更气人的是，有些商品会先涨价再打折，你一不小心就踩坑。

手动刷价格太累，第三方比价平台又经常不准。不如自己写一个：盯住你想买的商品，价格一降就通知你。

这个项目的目标是做一个轻量级价格监控脚本。支持自定义商品URL、设定心理价位、记录历史价格曲线，降价到阈值后自动推送消息。

全部用 Python 标准库 + requests + BeautifulSoup 搞定，不需要数据库后端，数据存本地 SQLite 就行。

技术选型

组件	选型	理由
语言	Python 3.10+	生态丰富，写脚本快
HTTP	requests	简单好用
HTML解析	BeautifulSoup4	兼容各种商品页面结构
数据存储	SQLite3	Python内置，零配置
通知	邮件 / Telegram Bot / Webhook	灵活选择
定时任务	Cron	系统级定时，稳定可靠

不需要数据库服务，不需要前端框架。纯脚本，丢到VPS或树莓派上就能跑。

实现步骤

第一步：安装依赖

pip install requests beautifulsoup4

就这两个包。SQLite、smtplib、urllib 全是 Python 内置模块。

第二步：创建价格数据库

新建 price_monitor.py，先写数据库初始化逻辑：

import sqlite3
from datetime import datetime

DB_PATH = "prices.db"

def init_db():
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS products (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            url TEXT NOT NULL UNIQUE,
            target_price REAL,
            currency TEXT DEFAULT 'CNY',
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS price_history (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            product_id INTEGER,
            price REAL NOT NULL,
            recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (product_id) REFERENCES products(id)
        )
    """)
    conn.commit()
    return conn

建了两张表：products 存商品基本信息，price_history 存每次抓到的价格记录。

第三步：写价格抓取器

不同平台的HTML结构不一样，所以这里设计成可配置的提取规则：

from bs4 import BeautifulSoup
import requests

DEFAULT_HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                  "AppleWebKit/537.36 (KHTML, like Gecko) "
                  "Chrome/120.0.0.0 Safari/537.36",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}

# 可自定义的提取规则
PRICE_SELECTORS = {
    "jd": "span.price",
    "taobao": ".tb-price, .Price--num",
    "generic": "[data-price], .price, .ProductPrice",
}

def fetch_price(url, platform="generic"):
    """抓取单个商品页面的当前价格"""
    resp = requests.get(url, headers=DEFAULT_HEADERS, timeout=15)
    resp.encoding = resp.apparent_encoding
    soup = BeautifulSoup(resp.text, "html.parser")

    # 优先使用 platform 指定的选择器
    selector = PRICE_SELECTORS.get(platform, PRICE_SELECTORS["generic"])

    # 尝试多个候选选择器
    candidates = selector.split(", ")
    for sel in candidates:
        el = soup.select(sel.strip())
        if el:
            text = el[0].get_text(strip=True)
            price = float("".join(c for c in text if c.isdigit() or c == "."))
            return price

    # 兜底：用正则找常见价格格式
    import re
    match = re.search(r'[¥￥]\s*([\d,]+\.?\d*)', resp.text)
    if match:
        return float(match.group(1).replace(",", ""))

    return None

这段代码的核心思路是：先按平台选择器抓，失败就用正则兜底。遇到反爬严格的页面，可以加代理池或换 User-Agent。

第四步：写入历史记录并判断是否降价

def record_price(conn, product_id, price):
    """记录价格到历史表"""
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO price_history (product_id, price) VALUES (?, ?)",
        (product_id, price)
    )
    conn.commit()

def check_alerts(conn):
    """检查所有商品，如果低于目标价则触发提醒"""
    cursor = conn.cursor()
    cursor.execute("SELECT id, name, url, target_price FROM products")
    products = cursor.fetchall()

    for pid, name, url, target_price in products:
        current_price = fetch_price(url)
        if current_price is None:
            print(f"⚠️ {name}: 无法获取价格")
            continue

        record_price(conn, pid, current_price)

        if target_price and current_price <= target_price:
            print(f"🎉 {name} 降价！当前 ¥{current_price}，目标 ¥{target_price}")
            send_alert(name, current_price, target_price, url)
        else:
            print(f"📦 {name}: 当前 ¥{current_price}", end="")
            if target_price:
                diff = current_price - target_price
                print(f" (还差 ¥{diff:.0f})")
            else:
                print()

第五步：接入通知渠道

支持三种通知方式，按需启用：

import smtplib
from email.mime.text import MIMEText

def send_alert(product_name, current_price, target_price, url):
    subject = f"价格提醒：{product_name} 已降至 ¥{current_price}"
    body = f"{product_name}\n当前价格：¥{current_price}\n目标价格：¥{target_price}\n商品链接：{url}"

    # 方式1：邮件通知
    # msg = MIMEText(body, "plain", "utf-8")
    # msg["Subject"] = subject
    # msg["From"] = "your_email@qq.com"
    # msg["To"] = "youremail@qq.com"
    # with smtplib.SMTP_SSL("smtp.qq.com", 465) as server:
    #     server.login("your_email@qq.com", "授权码")
    #     server.send_message(msg)

    # 方式2：Telegram Bot（推荐，免费且实时）
    TELEGRAM_BOT_TOKEN = "YOUR_BOT_TOKEN"
    TELEGRAM_CHAT_ID = "YOUR_CHAT_ID"
    tg_url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    requests.post(tg_url, json={
        "chat_id": TELEGRAM_CHAT_ID,
        "text": f"🔔 降价提醒\n\n{subject}\n\n{body}",
        "parse_mode": "HTML"
    })

    # 方式3：Webhook（支持飞书/钉钉/企业微信）
    # webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/xxx"
    # requests.post(webhook_url, json={
    #     "msg_type": "text",
    #     "content": {"text": subject + "\n" + body}
    # })

Telegram 方案最简单：创建一个 Bot，找 @userinfobot 拿你的 Chat ID，把 token 和 chat_id 填进去就行。

第六步：配置定时任务

crontab -e

每6小时检查一次，白天刷4次足够：

0 */6 * * * cd ~/price-monitor && /usr/bin/python3 price_monitor.py >> monitor.log 2>&1

运行效果

$ python3 price_monitor.py
📦 MacBook Air M3: 当前 ¥8999 (还差 ¥500)
📦 Sony WH-1000XM5: 当前 ¥1899
🎉 Sony WH-1000XM5 降价！当前 ¥1899，目标 ¥2000

Telegram 会收到一条消息：

🔔 降价提醒

Sony WH-1000XM5 已降至 ¥1899

当前价格：¥1899
目标价格：¥2000
商品链接：https://...

历史数据可以用 SQLite 直接查：

SELECT p.name, h.price, h.recorded_at
FROM price_history h JOIN products p ON h.product_id = p.id
WHERE p.name = 'Sony WH-1000XM5'
ORDER BY h.recorded_at DESC LIMIT 10;

优化方向

多平台适配：给京东、淘宝、拼多多各写一套专属选择器，或者用 Selenium 处理动态渲染页面。
价格曲线图：把 SQLite 数据导出为 CSV，用 matplotlib 画折线图，直观看到价格趋势。
智能预测：接入简单的线性回归模型，根据历史价格走势预测下次大促可能降到什么价位。
商品管理界面：用 Flask 做一个极简 Web UI，网页里添加/删除商品、查看历史价格，不用手动改代码。
反爬策略：加入随机请求间隔、代理 IP 轮换、Cookie 池，应对 stricter 的反爬机制。

完整代码已开源在 GitHub。Fork 后修改商品列表和通知配置就能直接用。

这篇文章对你有帮助吗？