用 Python + Playwright 做一个网页变动监控小工具

你盯着一款球鞋等了两周，到手前一天突然涨价了。或者你关注的课程突然降价，你却不在电脑前。

这类痛点太常见了。与其每天手动刷新页面，不如写个脚本替你盯着。

今天做一个基于 Playwright 的网页监控工具，支持价格、库存、文案变动的自动检测，发现变化后通过 Server 酱推送到微信。

项目背景

网页监控有几个典型场景：

  • 电商降价提醒：蹲限量鞋、显卡、演唱会门票
  • 竞品价格跟踪：监控对手网站的定价策略
  • 课程/资料更新：关注某个知识付费平台的课程上线
  • 招聘信息监控：特定岗位一上架就收到通知

手动刷网页效率太低，而且容易错过窗口期。自动化方案的核心思路是：定时抓取目标页面，提取关键信息，和上次结果比对，有变化就推送通知。

技术选型

组件选择理由
浏览器自动化Playwright支持 JS 渲染页面，比 requests 更稳定
调度器APScheduler轻量级，不需要额外服务
通知渠道Server 酱免费推送到微信，接入简单
数据存储SQLite零配置，存历史快照足够

实现步骤

第一步：安装依赖

pip install playwright apscheduler
playwright install chromium

第二步：核心监控脚本

创建 monitor.py：

import sqlite3
import time
import hashlib
from pathlib import Path
from playwright.sync_api import sync_playwright
from apscheduler.schedulers.blocking import BlockingScheduler

DB_PATH = Path(__file__).parent / "monitor.db"
SERVER_CHAN_TOKEN = "YOUR_SERVERCHAN_KEY"  # 替换为你的 Server 酱 key


def init_db():
    """初始化数据库，存储历史快照"""
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS snapshots (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            url TEXT NOT NULL,
            fingerprint TEXT NOT NULL,
            timestamp REAL NOT NULL,
            content_hash TEXT NOT NULL
        )
    """)
    conn.commit()
    return conn


def get_last_fingerprint(conn, url):
    """查询指定 URL 的最新指纹"""
    row = conn.execute(
        "SELECT content_hash FROM snapshots WHERE url = ? ORDER BY id DESC LIMIT 1",
        (url,)
    ).fetchone()
    return row[0] if row else None


def compute_fingerprint(page_content):
    """计算页面内容的哈希指纹，用于快速比对"""
    return hashlib.md5(page_content.encode()).hexdigest()


def extract_price(content):
    """从页面文本中提取价格信息（适配常见格式）"""
    import re
    prices = re.findall(r'¥?(\d+(?:,\d{3})*(?:\.\d+)?)', content)
    return prices[0] if prices else None


def send_notification(title, message):
    """通过 Server 酱发送微信通知"""
    import requests
    url = f"https://sctapi.ftqq.com/{SERVER_CHAN_TOKEN}.send"
    data = {
        "title": title,
        "desp": message,
    }
    resp = requests.post(url, data=data, timeout=10)
    return resp.json()


def monitor_page(url, label, check_interval=300):
    """
    监控单个页面的核心函数
    
    Args:
        url: 目标网页地址
        label: 标签名，用于通知时区分
        check_interval: 检查间隔（秒），默认5分钟
    """
    conn = init_db()
    
    def check():
        with sync_playwright() as p:
            browser = p.chromium.launch(headless=True)
            page = browser.new_page()
            
            try:
                page.goto(url, wait_until="networkidle", timeout=30000)
                # 等待价格或关键元素出现
                page.wait_for_selector("body", timeout=10000)
                content = page.content()
                fingerprint = compute_fingerprint(content)
                
                last_fp = get_last_fingerprint(conn, url)
                
                if last_fp and last_fp != fingerprint:
                    # 页面发生了变化，提取关键信息
                    price = extract_price(content)
                    msg_parts = [f"🔔 [{label}] 页面已更新！"]
                    if price:
                        msg_parts.append(f"💰 检测到价格: ¥{price}")
                    msg_parts.append(f"\n📎 链接: {url}")
                    
                    result = send_notification("网页监控告警", "\n".join(msg_parts))
                    if result.get("code") == 0:
                        print(f"[{time.strftime('%H:%M:%S')}] 已发送通知: {label}")
                    else:
                        print(f"[{time.strftime('%H:%M:%S')}] 通知发送失败: {result}")
                else:
                    print(f"[{time.strftime('%H:%M:%S')}] [{label}] 无变化")
                
                # 保存快照
                conn.execute(
                    "INSERT INTO snapshots (url, fingerprint, timestamp, content_hash) VALUES (?, ?, ?, ?)",
                    (url, fingerprint, time.time(), fingerprint)
                )
                conn.commit()
                
            except Exception as e:
                print(f"[{time.strftime('%H:%M:%S')}] [{label}] 出错: {e}")
            finally:
                browser.close()
    
    return check


def main():
    scheduler = BlockingScheduler()
    
    # 配置要监控的目标
    targets = [
        ("https://example.com/product/123", "限量球鞋"),
        ("https://example.com/course/456", "Python课程"),
        ("https://example.com/ticket", "演唱会门票"),
    ]
    
    for url, label in targets:
        check_fn = monitor_page(url, label, check_interval=300)
        scheduler.add_job(check_fn, "interval", minutes=5, id=f"job_{label}")
    
    print("⏰ 监控已启动，按 Ctrl+C 停止")
    print(f"📋 监控目标数: {len(targets)}")
    scheduler.start()


if __name__ == "__main__":
    main()

第三步：配置监控目标

修改 main() 函数中的 targets 列表，填入你要监控的真实 URL。比如：

targets = [
    ("https://www.taobao.com/item/12345678", "Air Jordan 1"),
    ("https://www.example.com/course/python-advanced", "进阶Python课"),
]

第四步：获取 Server 酱通知 Key

  1. 访问 https://sct.ftqq.com ，用微信登录
  2. 在"推送设置"中绑定微信
  3. 复制你的 SendKey（形如 SCT123456xxxxxxxxxxxxx）
  4. 替换脚本中的 SERVER_CHAN_TOKEN

第五步：运行

python monitor.py

首次运行会自动创建 monitor.db 数据库并记录基准快照。之后每隔 5 分钟检查一次。

如果想长期后台运行，可以用 systemd 或 nohup：

nohup python monitor.py > monitor.log 2>&1 &

进阶：只监控页面中的某个区域

如果整个页面经常变但你要关注的区域不变，可以精确提取：

def extract_element_text(page, selector):
    """提取页面中某个 CSS 选择器的文本内容"""
    element = page.query_selector(selector)
    if element:
        return element.inner_text()
    return None

# 使用示例：只监控价格元素
price_text = extract_element_text(page, ".product-price")
fingerprint = compute_fingerprint(price_text)

这样即使页面广告位变了，也不会触发误报。

运行效果

启动后终端输出类似：

⏰ 监控已启动，按 Ctrl+C 停止
📋 监控目标数: 3
[14:05:00] [限量球鞋] 无变化
[14:10:00] [Python课程] 无变化
[14:15:00] [演唱会门票] 无变化
[14:20:00] [限量球鞋] 已发送通知: 限量球鞋

当目标页面发生变化时，微信会收到一条推送：

🔔 [限量球鞋] 页面已更新！
💰 检测到价格: ¥1,299
📎 链接: https://...

优化方向

这个基础版本已经能用，后续可以从这几个方向继续完善：

  • 智能去重：用 difflib 对比页面差异，只在关键区域变化时才通知，减少误报
  • 多通知渠道：同时支持钉钉机器人、Telegram Bot、邮件推送，不依赖单一平台
  • 可视化面板：加一个 Flask 或 Streamlit 前端，管理监控目标和查看历史变化记录
  • 代理池：高频请求容易被封 IP，接入代理池提高稳定性
  • 截图存档：每次变化自动截取当前页面图片，方便追溯
  • 条件过滤：支持设置价格阈值，只有降价到指定金额才通知，避免无效推送

监控的本质就是用代码换时间。花半小时写好脚本，之后每天替你省下几十次手动刷新。

你现在最想监控哪个网站？留言告诉我，帮你写对应的提取逻辑。