用 Python 搭建自动化竞品网站监控工具，价格变动第一时间通知

公司上线了一款新产品，但不知道竞品最近有没有降价。

手动每天去翻对方网站？太累了。

写个脚本自动监控，有变化就发消息提醒你。这才是正确做法。

这个项目用 Python + Requests + SQLite 搭一个轻量级竞品监控工具。它定期抓取目标网页，对比内容变化，发现差异后通过邮件或微信推送通知。

项目背景

做电商、做 SaaS、做内容创业，竞品动态就是你的风向标。

但市面上的竞品监控工具要么太贵，要么功能太重。你只需要知道几个核心指标变了没有——价格、标题、库存状态。

这个工具的特点：

纯 Python 实现，不用装重型框架
SQLite 存储历史快照，零外部依赖
支持 HTML 指定区域提取，只关注你要的部分
变化通知支持邮件和 Server 酱微信推送
一天跑一次就够了，资源消耗几乎为零

技术选型

组件	选型	理由
HTTP 请求	Requests	简单好用，支持重试
HTML 解析	BeautifulSoup4	精准定位页面元素
数据存储	SQLite	内置于 Python，无需额外部署
定时任务	schedule 库 / crontab	灵活选择，推荐 crontab
通知方式	SMTP 邮件 + Server 酱	免费且可靠

实现步骤

第一步：安装依赖

pip install requests beautifulsoup4 schedule

三个库，不到十兆。

第二步：编写监控脚本

创建 monitor.py：

import requests
import sqlite3
from bs4 import BeautifulSoup
import hashlib
import smtplib
from email.mime.text import MIMEText
import serverchan  # 可选，用于微信推送
import time
import schedule

# ============ 配置区 ============
TARGET_URLS = {
    "竞品A-价格页": "https://example-competitor-a.com/pricing",
    "竞品B-首页": "https://example-competitor-b.com",
}

# 每个页面的监控规则
RULES = {
    "竞品A-价格页": {
        "selector": ".pricing-card .price",  # CSS 选择器
        "extract_text": True,
    },
    "竞品B-首页": {
        "selector": "h1.hero-title",
        "extract_text": True,
    },
}

# 通知配置
EMAIL_CONFIG = {
    "smtp_server": "smtp.qq.com",
    "smtp_port": 465,
    "sender": "your@qq.com",
    "password": "your_smtp_password",  # QQ邮箱是授权码
    "receiver": "you@gmail.com",
}

SERVERCHAN_KEY = "YOUR_SERVERCHAN_KEY"  # server酱Token

DB_PATH = "monitor.db"


def init_db(conn):
    """初始化数据库"""
    conn.execute("""
        CREATE TABLE IF NOT EXISTS snapshots (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            page_name TEXT NOT NULL,
            content_hash TEXT NOT NULL,
            raw_content TEXT,
            detected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS rules (
            page_name TEXT PRIMARY KEY,
            selector TEXT,
            extract_text BOOLEAN DEFAULT 1
        )
    """)
    # 写入规则
    for name, rule in RULES.items():
        conn.execute(
            "INSERT OR REPLACE INTO rules VALUES (?, ?, ?)",
            (name, rule["selector"], rule.get("extract_text", True)),
        )
    conn.commit()


def fetch_page(url):
    """获取网页内容"""
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                       "AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36"
    }
    resp = requests.get(url, headers=headers, timeout=15)
    resp.encoding = resp.apparent_encoding
    return resp.text


def extract_content(html, selector, extract_text=True):
    """根据 CSS 选择器提取内容"""
    soup = BeautifulSoup(html, "html.parser")
    elements = soup.select(selector)
    if not elements:
        return None
    if extract_text:
        return "\n".join(el.get_text(strip=True) for el in elements)
    return str(elements[0])


def compute_hash(content):
    """计算内容哈希值"""
    if content is None:
        return "empty"
    return hashlib.sha256(content.encode("utf-8")).hexdigest()


def send_email_notification(page_name, old_hash, new_hash, new_content):
    """发送邮件通知"""
    subject = f"[监控告警] {page_name} 内容发生变化"
    body = f"""页面名称: {page_name}
旧内容哈希: {old_hash}
新内容哈希: {new_hash}

最新内容:
{new_content}

请及时查看。"""

    msg = MIMEText(body, "plain", "utf-8")
    msg["Subject"] = subject
    msg["From"] = EMAIL_CONFIG["sender"]
    msg["To"] = EMAIL_CONFIG["receiver"]

    with smtplib.SMTP_SSL(
        EMAIL_CONFIG["smtp_server"], EMAIL_CONFIG["smtp_port"]
    ) as server:
        server.login(EMAIL_CONFIG["sender"], EMAIL_CONFIG["password"])
        server.sendmail(
            EMAIL_CONFIG["sender"],
            [EMAIL_CONFIG["receiver"]],
            msg.as_string(),
        )


def send_wechat_notification(page_name, new_content):
    """通过 Server 酱发送微信通知"""
    text = f"🔔 竞品监控: {page_name}"
    desp = f"内容有变化:\n{new_content[:500]}"
    url = f"https://sctapi.ftqq.com/{SERVERCHAN_KEY}.send"
    requests.post(url, data={"text": text, "desp": desp})


def monitor():
    """执行一次监控"""
    conn = sqlite3.connect(DB_PATH)
    init_db(conn)

    cursor = conn.execute("SELECT page_name, selector, extract_text FROM rules")
    rules = {row[0]: {"selector": row[1], "extract_text": row[2]} for row in cursor}

    for page_name, url in TARGET_URLS.items():
        if page_name not in rules:
            continue

        rule = rules[page_name]
        print(f"正在检查: {page_name} -> {url}")

        try:
            html = fetch_page(url)
            content = extract_content(html, rule["selector"], rule["extract_text"])
            new_hash = compute_hash(content)
        except Exception as e:
            print(f"  ⚠ 请求失败: {e}")
            continue

        # 查询上次记录
        row = conn.execute(
            "SELECT content_hash FROM snapshots WHERE page_name = ? ORDER BY id DESC LIMIT 1",
            (page_name,),
        ).fetchone()

        if row is None:
            # 首次记录
            conn.execute(
                "INSERT INTO snapshots (page_name, content_hash, raw_content) VALUES (?, ?, ?)",
                (page_name, new_hash, content),
            )
            conn.commit()
            print(f"  ✓ 首次记录完成")
        elif row[0] != new_hash:
            # 检测到变化
            print(f"  🔔 发现变化！")
            try:
                send_email_notification(page_name, row[0], new_hash, content)
                print("  📧 邮件通知已发送")
            except Exception as e:
                print(f"  ✖ 邮件发送失败: {e}")

            try:
                send_wechat_notification(page_name, content)
                print("  📱 微信通知已发送")
            except Exception as e:
                print(f"  ✖ 微信通知失败: {e}")

            # 更新记录
            conn.execute(
                "INSERT INTO snapshots (page_name, content_hash, raw_content) VALUES (?, ?, ?)",
                (page_name, new_hash, content),
            )
            conn.commit()
        else:
            print(f"  ✓ 无变化")

    conn.close()


if __name__ == "__main__":
    # 开发调试时可以取消注释下面这行，每小时执行一次
    # schedule.every().hour.do(monitor)
    # while True:
    #     schedule.run_pending()
    #     time.sleep(60)

    # 生产环境直接用 crontab 调度
    monitor()
    print("监控任务完成")

第三步：配置你的监控目标

修改脚本顶部的 TARGET_URLS 字典，填入你要监控的竞品网址。

RULES 里定义每个页面用哪个 CSS 选择器来提取内容。比如：

.price 提取价格
.stock-status 提取库存状态
h1 提取主标题

用浏览器开发者工具就能找到对应的选择器。

第四步：设置定时任务

推荐用 crontab，服务器不需要一直挂着 Python 进程。

# 每天早上9点执行一次监控
crontab -e
# 添加这一行：
0 9 * * * cd ~/competitor-monitor && python3 monitor.py >> monitor.log 2>&1

如果你本地开发，可以用 schedule 库替代 crontab。

第五步：测试验证

先改一下监控规则，指向一个你能控制变化的页面，比如自己的博客。确认通知能正常收到，再切回真正的竞品地址。

运行效果

脚本每次执行会输出类似这样的日志：

正在检查: 竞品A-价格页 -> https://example-competitor-a.com/pricing
✓ 无变化
正在检查: 竞品B-首页 -> https://example-competitor-b.com
🔔 发现变化！
📧 邮件通知已发送
📱 微信通知已发送
监控任务完成

收到邮件或微信消息后，你就能知道竞品什么时候改了价格、换了标语。

优化方向

这个基础版本已经能用了，后续可以根据需要扩展：

JS 渲染页面：遇到 Vue/React 渲染的页面，Requests 抓不到内容。换成 Playwright，几行代码就能拿到渲染后的 HTML。
图片对比：有些竞品不改动文字，只换 Banner 图。加一个图片哈希对比模块，用 imagehash 库做感知哈希。
结构化数据提取：如果竞品价格藏在复杂 HTML 里，可以用正则表达式或 XPath 精确定位。
Webhook 集成：把通知推送到钉钉、飞书或 Slack，方便团队协作。
历史记录可视化：把 SQLite 数据导出到 CSV，用 Excel 或 Grafana 画趋势图，看清竞品价格变化规律。
反爬对抗：加随机延迟、代理 IP 池、Cookie 轮换，降低被封的概率。

一个脚本搞定的事，别花几百块买 SaaS 工具。

选一个你最关心的竞品页面，今天就把脚本跑起来。

你监控的主要是哪个平台的竞品？评论区聊聊。

这篇文章对你有帮助吗？