用 Python 做网站变化监控，改版第一时间收到通知

pip install playwright pillow numpy apscheduler pyyaml
playwright install chromium

mkdir -p site_monitor/{screenshots,logs}
touch monitor.py config.yaml

# 监控目标列表
targets:
  - url: "https://example.com"
    name: "示例网站"
    interval: 3600  # 每小时检查一次
    notify: true    # 有变化时通知
    
  - url: "https://competitor.com/pricing"
    name: "竞品价格页"
    interval: 7200  # 每两小时检查
    notify: true

# 通知设置
notification:
  type: "wechat"  # wechat / email / webhook
  wechat:
    webhook_url: "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
  email:
    smtp_server: "smtp.qq.com"
    smtp_port: 465
    sender: "your@qq.com"
    password: "your_password"
    receiver: "target@qq.com"

# 截图配置
screenshot:
  width: 1440
  height: 900
  compare_threshold: 0.05  # 像素差异超过5%才触发告警

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import yaml
import time
from datetime import datetime
from pathlib import Path
from PIL import Image
import numpy as np
from playwright.sync_api import sync_playwright
from apscheduler.schedulers.blocking import BlockingScheduler
import requests

# 加载配置
with open("config.yaml", "r", encoding="utf-8") as f:
    config = yaml.safe_load(f)

SCREENSHOT_DIR = Path("screenshots")
SCREENSHOT_DIR.mkdir(exist_ok=True)

def take_screenshot(url, name, width=1440, height=900):
    """抓取网页截图"""
    filename = f"{SCREENSHOT_DIR}/{name}_{int(time.time())}.png"
    
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page(viewport={"width": width, "height": height})
        page.goto(url, wait_until="networkidle")
        page.screenshot(path=filename, full_page=True)
        browser.close()
    
    return filename

def compare_screenshots(old_path, new_path, threshold=0.05):
    """对比两张截图，返回差异率"""
    img1 = np.array(Image.open(old_path).convert("L"))
    img2 = np.array(Image.open(new_path).convert("L"))
    
    # 调整尺寸到一致
    if img1.shape != img2.shape:
        img2 = np.array(Image.fromarray(img2).resize((img1.shape[1], img1.shape[0])))
    
    # 计算像素差异率
    diff = np.abs(img1.astype(float) - img2.astype(float)) / 255.0
    diff_rate = np.mean(diff)
    
    return diff_rate > threshold, diff_rate

def send_wechat_notification(url, name, diff_rate):
    """发送微信通知"""
    webhook = config["notification"]["wechat"]["webhook_url"]
    message = f"🔔 网站变化监控\n\n网站：{name}\n链接：{url}\n变化率：{diff_rate:.2%}\n时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
    
    payload = {
        "msgtype": "text",
        "text": {
            "content": message
        }
    }
    
    requests.post(webhook, json=payload, timeout=10)
    print(f"✅ 已发送微信通知：{name}")

def check_site(target):
    """检查单个网站是否有变化"""
    name = target["name"]
    url = target["url"]
    notify = target.get("notify", True)
    
    print(f"\n🔍 正在检查：{name} ({url})")
    
    # 获取最新截图
    latest = take_screenshot(url, name, 
                             config["screenshot"]["width"],
                             config["screenshot"]["height"])
    
    # 找上一次截图
    screenshots = sorted(SCREENSHOT_DIR.glob(f"{name}_*.png"))
    
    if len(screenshots) < 2:
        print(f"  ⏳ 首次监控，已保存截图")
        return
    
    old_path = screenshots[-2]
    new_path = latest
    
    # 删除旧格式截图（避免混淆）
    for s in screenshots[:-2]:
        s.unlink()
    
    # 对比
    changed, diff_rate = compare_screenshots(old_path, new_path,
                                             config["screenshot"]["compare_threshold"])
    
    if changed:
        print(f"  ⚠️ 检测到变化！差异率：{diff_rate:.2%}")
        if notify:
            send_wechat_notification(url, name, diff_rate)
    else:
        print(f"  ✅ 无变化（差异率：{diff_rate:.2%}）")
    
    # 清理本次截图（保留最新一份）
    latest.unlink(missing_ok=True)

def main():
    """主程序"""
    scheduler = BlockingScheduler()
    
    for target in config["targets"]:
        scheduler.add_job(
            check_site,
            "interval",
            seconds=target["interval"],
            args=[target],
            id=target["name"],
            misfire_grace_time=60
        )
        print(f"📌 已添加监控任务：{target['name']} (每{target['interval']}秒)")
    
    print("\n🚀 网站变化监控启动，按 Ctrl+C 退出\n")
    
    try:
        scheduler.start()
    except KeyboardInterrupt:
        print("\n👋 监控已停止")

if __name__ == "__main__":
    main()

python monitor.py