电商比价小助手：定时抓取 + 降价提醒，告别错过折扣

pip install requests beautifulsoup4 schedule

import requests
import sqlite3
import smtplib
import schedule
import time
from bs4 import BeautifulSoup
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime
import re

# ============ 配置区 ============

# 商品列表：每个商品一个字典
PRODUCTS = [
    {
        "name": "Sony WH-1000XM5 耳机",
        "url": "https://example-product-page-1.com",
        "target_price": 1500,  # 目标价（元）
        "price_css": ".price-class",  # 价格元素的 CSS 选择器
    },
    {
        "name": "Kindle Paperwhite 2024",
        "url": "https://example-product-page-2.com",
        "target_price": 800,
        "price_css": "#price-block",
    },
]

# 邮件配置
SMTP_HOST = "smtp.qq.com"
SMTP_PORT = 465
EMAIL_USER = "your_email@qq.com"
EMAIL_PASS = "your_smtp_token"
EMAIL_TO = "your_email@qq.com"

# 数据库
DB_PATH = "price_history.db"

# ============ 数据库初始化 ============

def init_db():
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    c.execute("""
        CREATE TABLE IF NOT EXISTS prices (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            product_name TEXT,
            price REAL,
            target_price REAL,
            checked_at TEXT
        )
    """)
    conn.commit()
    conn.close()

# ============ 价格抓取 ============

def get_price(product):
    """从商品页面抓取当前价格"""
    headers = {
        "User-Agent": "Mozilla/5.0 (compatible; PriceBot/1.0)"
    }
    try:
        resp = requests.get(product["url"], headers=headers, timeout=10)
        resp.encoding = resp.apparent_encoding
        soup = BeautifulSoup(resp.text, "html.parser")
        
        # 方式一：CSS 选择器提取
        price_elem = soup.select_one(product["price_css"])
        if price_elem:
            text = price_elem.get_text(strip=True)
            # 提取数字
            match = re.search(r'[\d,]+\.?\d*', text.replace(",", ""))
            if match:
                return float(match.group())
        
        # 方式二：如果 CSS 选择器不匹配，尝试通用价格模式
        all_prices = soup.find_all(string=re.compile(r'¥?\s*[\d,]+\.?\d*'))
        for p in all_prices:
            text = p.strip()
            match = re.search(r'[\d,]+\.?\d*', text.replace(",", "").replace("¥", ""))
            if match:
                val = float(match.group())
                if 100 < val < 50000:  # 过滤掉不合理的小数值
                    return val
        return None
    except Exception as e:
        print(f"  [抓取失败] {product['name']}: {e}")
        return None

# ============ 价格记录 ============

def save_price(product_name, price, target_price):
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    c.execute(
        "INSERT INTO prices (product_name, price, target_price, checked_at) VALUES (?, ?, ?, ?)",
        (product_name, price, target_price, datetime.now().isoformat())
    )
    conn.commit()
    conn.close()

# ============ 邮件通知 ============

def send_notification(product_name, current_price, target_price):
    subject = f"🎉 价格触达提醒：{product_name}"
    body = f"""
价格触达提醒

商品名称：{product_name}
当前价格：¥{current_price:.2f}
目标价格：¥{target_price:.2f}

当前价格已降至目标价以下，可以入手了！

—— 电商比价小助手
"""
    msg = MIMEMultipart()
    msg["From"] = EMAIL_USER
    msg["To"] = EMAIL_TO
    msg["Subject"] = subject
    msg.attach(MIMEText(body, "plain", "utf-8"))

    with smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT) as server:
        server.login(EMAIL_USER, EMAIL_PASS)
        server.sendmail(EMAIL_USER, EMAIL_TO, msg.as_string())
    print(f"  [已通知] {product_name}")

# ============ 主检查逻辑 ============

def check_all():
    print(f"[{datetime.now().strftime('%H:%M')}] 开始检查...")
    for p in PRODUCTS:
        price = get_price(p)
        if price is None:
            print(f"  [跳过] {p['name']}：无法获取价格")
            continue
        
        save_price(p["name"], price, p["target_price"])
        print(f"  [{p['name']}] 当前价格: ¥{price:.2f}  | 目标价: ¥{p['target_price']:.2f}")
        
        if price <= p["target_price"]:
            # 查一下今天是否已经通知过
            send_notification(p["name"], price, p["target_price"])

# ============ 定时任务 ============

def main():
    init_db()
    print("=== 电商比价小助手启动 ===")
    schedule.every(6).hours.do(check_all)  # 每6小时检查一次
    check_all()  # 首次立即检查
    while True:
        schedule.run_pending()
        time.sleep(60)

if __name__ == "__main__":
    main()

nohup python3 price_monitor.py > monitor.log 2>&1 &

sudo tee /etc/systemd/system/price-monitor.service <<EOF
[Unit]
Description=Price Monitor
After=network.target

[Service]
Type=simple
WorkingDirectory=/home/ubuntu/price-monitor
ExecStart=/usr/bin/python3 /home/ubuntu/price-monitor/price_monitor.py
Restart=always
RestartSec=30

[Install]
WantedBy=multi-user.target
EOF

sudo systemctl enable price-monitor
sudo systemctl start price-monitor

import sqlite3
conn = sqlite3.connect("price_history.db")
c = conn.cursor()
c.execute("SELECT product_name, price, checked_at FROM prices ORDER BY checked_at DESC LIMIT 20")
for row in c.fetchall():
    print(f"{row[0]} | ¥{row[1]:.2f} | {row[2]}")
conn.close()