author: 专注Python实战，分享爬虫与数据分析干货
title: Python爬虫实战㉖｜综合实战1，电商商品价格监控系统
update: 2026-04-26
tags: Python,爬虫实战,电商,价格监控,定时爬虫,数据对比,降价提醒

作者：专注Python实战，分享爬虫与数据分析干货
更新时间：2026年4月
适合人群：已学完全部基础、想做完整项目的开发者

前言：价格监控 = 爬虫 + 分析 + 通知

网购时想等降价再买？手动刷太累。用Python自动监控：

  1. 定时爬取商品价格
  2. 对比历史价格
  3. 降价时自动通知

本篇把之前学的所有知识串起来，做一个完整项目！

一、项目架构

price_monitor/
├── config.py          # 配置文件
├── models.py          # 数据模型
├── crawler.py         # 爬虫模块
├── analyzer.py        # 分析模块
├── notifier.py        # 通知模块
├── main.py            # 入口
└── data/              # 数据目录
    └── prices.db      # SQLite数据库

二、配置文件

# config.py

# 监控商品列表
PRODUCTS = [
    {"name": "商品A", "url": "https://example.com/product/1", "selector": ".price"},
    {"name": "商品B", "url": "https://example.com/product/2", "selector": ".price-now"},
    {"name": "商品C", "url": "https://example.com/product/3", "selector": "#price"},
]

# 爬虫配置
CRAWL_DELAY = 3          # 请求间隔（秒）
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0.0.0",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 Chrome/119.0.0.0",
]

# 降价阈值
PRICE_DROP_THRESHOLD = 0.05  # 降价5%触发通知

# 数据库
DB_PATH = "data/prices.db"

# 通知方式
NOTIFY_METHOD = "console"  # console / email / webhook

三、数据模型

# models.py
import sqlite3

class PriceDB:
    """价格数据库"""

    def __init__(self, db_path="data/prices.db"):
        self.db_path = db_path
        import os
        os.makedirs(os.path.dirname(db_path), exist_ok=True)
        self.conn = sqlite3.connect(db_path)
        self._create_table()

    def _create_table(self):
        self.conn.execute("""
        CREATE TABLE IF NOT EXISTS price_history (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            product_name TEXT NOT NULL,
            url TEXT,
            price REAL NOT NULL,
            crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        """)
        self.conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_product ON price_history(product_name)"
        )
        self.conn.commit()

    def insert_price(self, product_name, url, price):
        self.conn.execute(
            "INSERT INTO price_history (product_name, url, price) VALUES (?, ?, ?)",
            (product_name, url, price),
        )
        self.conn.commit()

    def get_latest_price(self, product_name):
        cursor = self.conn.execute(
            "SELECT price FROM price_history "
            "WHERE product_name=? ORDER BY crawl_time DESC LIMIT 1",
            (product_name,),
        )
        row = cursor.fetchone()
        return row[0] if row else None

    def get_price_history(self, product_name, days=30):
        cursor = self.conn.execute("""
        SELECT price, crawl_time FROM price_history
        WHERE product_name=? AND crawl_time >= datetime('now', ?)
        ORDER BY crawl_time
        """, (product_name, f"-{days} days"))
        return cursor.fetchall()

    def close(self):
        self.conn.close()

四、爬虫模块

# crawler.py
import requests
from bs4 import BeautifulSoup
import random
import time
import re

class PriceCrawler:
    """价格爬虫"""

    def __init__(self):
        self.session = requests.Session()

    def fetch_price(self, url, selector):
        """抓取商品价格"""
        from config import CRAWL_DELAY, USER_AGENTS
        headers = {"User-Agent": random.choice(USER_AGENTS)}

        try:
            response = self.session.get(url, headers=headers, timeout=15)
            response.raise_for_status()
            response.encoding = response.apparent_encoding

            soup = BeautifulSoup(response.text, "html.parser")
            price_element = soup.select_one(selector)

            if price_element:
                price_text = price_element.get_text(strip=True)
                price = self._parse_price(price_text)
                return price
            else:
                print(f"  未找到价格元素: {selector}")
                return None

        except Exception as e:
            print(f"  爬取失败: {e}")
            return None
        finally:
            time.sleep(CRAWL_DELAY)

    def _parse_price(self, text):
        """解析价格文本"""
        clean = re.sub(r"[^\d.]", "", text)
        try:
            return float(clean)
        except ValueError:
            return None

    def crawl_all(self, products):
        """爬取所有商品"""
        results = []
        for product in products:
            print(f"  抓取: {product['name']}...")
            price = self.fetch_price(product["url"], product["selector"])
            if price:
                results.append({
                    "name": product["name"],
                    "url": product["url"],
                    "price": price,
                })
                print(f"    价格: ¥{price}")
            else:
                print(f"    价格获取失败")
        return results

五、分析模块

# analyzer.py
from config import PRICE_DROP_THRESHOLD

class PriceAnalyzer:
    """价格分析器"""

    def __init__(self, db):
        self.db = db

    def check_price_drop(self, product_name, current_price):
        """检查是否降价"""
        last_price = self.db.get_latest_price(product_name)

        if last_price is None:
            return {"is_drop": False, "reason": "首次记录", "last_price": None}

        if current_price < last_price:
            drop_pct = (last_price - current_price) / last_price
            is_significant = drop_pct >= PRICE_DROP_THRESHOLD

            return {
                "is_drop": True,
                "is_significant": is_significant,
                "drop_pct": drop_pct * 100,
                "last_price": last_price,
                "current_price": current_price,
                "saved": last_price - current_price,
                "reason": (
                    f"降价{drop_pct*100:.1f}%！省¥{last_price - current_price:.0f}"
                    if is_significant
                    else f"微降{drop_pct*100:.1f}%"
                ),
            }

        elif current_price > last_price:
            rise_pct = (current_price - last_price) / last_price * 100
            return {"is_drop": False, "last_price": last_price,
                    "reason": f"涨价{rise_pct:.1f}%"}

        return {"is_drop": False, "last_price": last_price, "reason": "价格不变"}

    def get_price_summary(self, product_name):
        """获取价格摘要"""
        history = self.db.get_price_history(product_name, days=30)
        if not history:
            return None

        prices = [h[0] for h in history]
        return {
            "当前价格": prices[-1],
            "30天最高": max(prices),
            "30天最低": min(prices),
            "30天均价": round(sum(prices) / len(prices), 2),
            "记录次数": len(prices),
        }

六、通知模块

# notifier.py

class Notifier:
    """通知器"""

    def send(self, message):
        """发送通知（控制台输出）"""
        print(f"\n🔔 通知: {message}")

    def notify_price_drop(self, product_name, info):
        """降价通知"""
        msg = (
            f"📢 【降价提醒】{product_name}\n"
            f"  原价: ¥{info['last_price']}\n"
            f"  现价: ¥{info['current_price']}\n"
            f"  降幅: {info['drop_pct']:.1f}%\n"
            f"  省: ¥{info['saved']:.0f}"
        )
        self.send(msg)

    def notify_daily_summary(self, summary):
        """每日摘要"""
        msg = "📋 每日价格摘要\n" + "=" * 30
        for name, info in summary.items():
            if info:
                msg += (
                    f"\n  {name}: ¥{info['当前价格']} "
                    f"(低¥{info['30天最低']} 高¥{info['30天最高']})"
                )
        self.send(msg)

    def send_email(self, to, subject, body):
        """邮件通知（扩展用）"""
        import smtplib
        from email.mime.text import MIMEText
        # 邮件发送逻辑（参考专栏第17-21篇）
        pass

    def send_webhook(self, url, data):
        """Webhook通知（企业微信/钉钉）"""
        import requests
        try:
            requests.post(url, json=data, timeout=10)
        except Exception as e:
            print(f"Webhook发送失败: {e}")

七、主程序

# main.py
from config import PRODUCTS
from models import PriceDB
from crawler import PriceCrawler
from analyzer import PriceAnalyzer
from notifier import Notifier

def main():
    print("=" * 50)
    print("  电商商品价格监控系统")
    print("=" * 50)

    # 初始化
    db = PriceDB()
    crawler = PriceCrawler()
    analyzer = PriceAnalyzer(db)
    notifier = Notifier()

    # 1. 爬取价格
    print("\n📡 开始爬取价格...")
    results = crawler.crawl_all(PRODUCTS)

    if not results:
        print("未获取到任何价格数据")
        db.close()
        return

    # 2. 分析与存储
    print("\n📊 分析价格变化...")
    summary = {}

    for item in results:
        name = item["name"]
        price = item["price"]

        # 检查降价
        drop_info = analyzer.check_price_drop(name, price)
        if drop_info.get("is_significant"):
            notifier.notify_price_drop(name, drop_info)

        # 存储到数据库
        db.insert_price(name, item["url"], price)

        # 生成摘要
        summary[name] = analyzer.get_price_summary(name)

        print(f"  {name}: ¥{price} - {drop_info['reason']}")

    # 3. 每日摘要
    notifier.notify_daily_summary(summary)

    db.close()
    print("\n✅ 监控完成")


if __name__ == "__main__":
    main()

八、定时执行

8.1 Windows任务计划

@echo off
cd /d D:\price_monitor
python main.py >> logs\monitor.log 2>&1

通过"任务计划程序"设置每小时执行一次。

8.2 Python定时

# scheduler.py
import schedule
import time
import logging
from main import main

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(message)s",
    handlers=[
        logging.FileHandler("logs/scheduler.log", encoding="utf-8"),
        logging.StreamHandler(),
    ],
)

# 每小时执行
schedule.every().hour.do(main)

# 每天早上9点执行
schedule.every().day.at("09:00").do(main)

if __name__ == "__main__":
    print("定时监控已启动...")
    while True:
        schedule.run_pending()
        time.sleep(60)

九、项目运行效果

==================================================
  电商商品价格监控系统
==================================================

📡 开始爬取价格...
  抓取: 商品A...
    价格: ¥2999
  抓取: 商品B...
    价格: ¥4599
  抓取: 商品C...
    价格: ¥899

📊 分析价格变化...

🔔 通知:
📢 【降价提醒】商品A
  原价: ¥3199
  现价: ¥2999
  降幅: 6.3%
  省: ¥200

  商品A: ¥2999 - 降价6.3%！省¥200
  商品B: ¥4599 - 价格不变
  商品C: ¥899 - 微降1.2%

📋 每日价格摘要
==============================
  商品A: ¥2999 (低¥2899 高¥3199)
  商品B: ¥4599 (低¥4499 高¥4799)
  商品C: ¥899 (低¥849 高¥949)

✅ 监控完成

十、知识卡

模块 说明
config.py 配置集中管理
models.py SQLite数据存储
crawler.py requests+BS4爬虫
analyzer.py 价格对比分析
notifier.py 多渠道通知
scheduler.py 定时执行
PRICE_DROP_THRESHOLD 降价阈值

十一、课后作业

必做题：

  1. 搭建完整的价格监控项目
  2. 实现SQLite存储和降价检测
  3. 添加控制台通知

选做题：

  1. 添加邮件/微信通知
  2. 画出价格趋势图
  3. 部署到服务器定时运行

有问题欢迎评论区留言，大家一起讨论！

标签：Python | 爬虫实战 | 电商 | 价格监控 | 定时爬虫 | 降价提醒

# python # 爬虫 # jvm
Logo
快递鸟一站式物流API解决方案

电商企业物流数字化转型必备！快递鸟 API 接口，72 小时快速完成物流系统集成。全流程实战1V1指导，营造开放的API技术生态圈。

更多推荐

cover

2026年最新AI数字人直播系统高强度效果实测

avatar 快递鸟社区
cover

2026深度实测：vibe coding真实开发优势与落地实战全解析

avatar 快递鸟社区
cover

客服部门为什么越来越贵，却越来越不管用

avatar 快递鸟社区