怎样用脚本监控网站变化？

本文目录导读：

1. Python 基础监控脚本
2. 更高级的监控（支持选择器）
3. 命令行脚本（使用curl和diff）
4. 使用现有工具 - changedetection.io
5. Node.js 脚本示例
6. 实际部署建议
7. 关键优化点
8. 完整示例（带通知）

Python 基础监控脚本

import requests
import hashlib
import time
import smtplib
from email.mime.text import MIMEText
class WebsiteMonitor:
    def __init__(self, url, check_interval=300):
        self.url = url
        self.check_interval = check_interval
        self.last_hash = None
    def get_content_hash(self):
        """获取页面内容的哈希值"""
        try:
            response = requests.get(self.url, timeout=10)
            # 可以只取特定部分的内容
            return hashlib.md5(response.text.encode()).hexdigest()
        except Exception as e:
            print(f"Error accessing {self.url}: {e}")
            return None
    def check_change(self):
        """检查页面是否变化"""
        current_hash = self.get_content_hash()
        if current_hash and current_hash != self.last_hash:
            if self.last_hash is not None:  # 忽略第一次检查
                print(f"变化 detected: {self.url}")
                self.notify()
            self.last_hash = current_hash
            return True
        return False
    def notify(self):
        """发送通知（示例：邮件）"""
        # 实现邮件发送逻辑
        pass
    def run(self):
        """运行监控"""
        print(f"开始监控: {self.url}")
        while True:
            self.check_change()
            time.sleep(self.check_interval)
# 使用示例
monitor = WebsiteMonitor("https://example.com", check_interval=300)
monitor.run()

更高级的监控（支持选择器）

import requests
from bs4 import BeautifulSoup
import hashlib
import time
class SmartMonitor:
    def __init__(self, url, css_selector=None):
        self.url = url
        self.css_selector = css_selector
        self.last_hash = None
    def get_target_content(self):
        """获取指定区域的内容"""
        response = requests.get(self.url)
        soup = BeautifulSoup(response.text, 'html.parser')
        if self.css_selector:
            # 只监控特定区域
            elements = soup.select(self.css_selector)
            content = ''.join([str(elem) for elem in elements])
        else:
            content = soup.get_text()  # 只保留文本
        return hashlib.md5(content.encode()).hexdigest()
    def monitor_loop(self, interval=60):
        while True:
            current_hash = self.get_target_content()
            if current_hash != self.last_hash:
                if self.last_hash:
                    print(f"内容变化！{self.url}")
                    # 可以在这里添加通知逻辑
                self.last_hash = current_hash
            time.sleep(interval)

命令行脚本（使用curl和diff）

#!/bin/bash
URL="https://example.com"
INTERVAL=300  # 5分钟
TEMP_FILE="/tmp/website_snapshot"
# 首次获取
curl -s "$URL" > "$TEMP_FILE"
while true; do
    NEW_FILE="/tmp/website_new_$$"
    curl -s "$URL" > "$NEW_FILE"
    if ! diff "$TEMP_FILE" "$NEW_FILE" > /dev/null; then
        echo "网站内容发生变化！"
        # 显示差异
        diff "$TEMP_FILE" "$NEW_FILE"
        # 更新快照
        cp "$NEW_FILE" "$TEMP_FILE"
    fi
    rm "$NEW_FILE"
    sleep $INTERVAL
done

使用现有工具 - changedetection.io

这不是脚本,但非常实用：

• 开源项目：https://github.com/dgtlmoon/changedetection.io
• 支持 Docker 部署
• 提供 Web 界面
• 支持多种通知方式

Node.js 脚本示例

const axios = require('axios');
const crypto = require('crypto');
const nodemailer = require('nodemailer');
class Monitor {
    constructor(url, interval = 300) {
        this.url = url;
        this.interval = interval * 1000;
        this.lastHash = null;
    }
    async fetchPage() {
        const response = await axios.get(this.url);
        return response.data;
    }
    hashContent(content) {
        return crypto.createHash('md5').update(content).digest('hex');
    }
    async check() {
        try {
            const content = await this.fetchPage();
            const hash = this.hashContent(content);
            if (hash !== this.lastHash) {
                if (this.lastHash) {
                    console.log(`变化 detected: ${this.url}`);
                    await this.notify();
                }
                this.lastHash = hash;
            }
        } catch (error) {
            console.error(error);
        }
    }
    async notify() {
        // 实现通知逻辑
    }
    start() {
        this.check();
        setInterval(() => this.check(), this.interval);
    }
}

实际部署建议

使用 systemd 服务（Linux）

[Unit]
Description=Website Monitor
After=network.target
[Service]
ExecStart=/usr/bin/python3 /path/to/monitor.py
Restart=always
User=your_user
[Install]
WantedBy=multi-user.target

使用 Docker Compose

version: '3'
services:
  monitor:
    image: python:3.9
    volumes:
      - ./monitor.py:/app/monitor.py
    command: python /app/monitor.py
    restart: always

关键优化点

1. 减少请求频率：添加合理的时间间隔
2. 条件监控：只关注特定内容变化
3. 离线存储：保存快照以便对比
4. 智能通知：
   • 邮件（SMTP）
   • 即时通讯（微信、Telegram、Slack）
   • 短信（Twilio）
   • 桌面通知

完整示例（带通知）

import requests
import hashlib
import time
import json
from datetime import datetime
class CompleteMonitor:
    def __init__(self, config_file='sites.json'):
        with open(config_file) as f:
            self.sites = json.load(f)
        self.history = {}
    def send_telegram(self, message):
        # 使用Telegram Bot API通知
        bot_token = "YOUR_BOT_TOKEN"
        chat_id = "YOUR_CHAT_ID"
        url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
        requests.post(url, json={'chat_id': chat_id, 'text': message})
    def monitor(self):
        for site in self.sites:
            url = site['url']
            selector = site.get('selector')
            current = self.get_content(url, selector)
            key = f"{url}_{selector}"
            if key in self.history and current != self.history[key]:
                msg = f"检测到变化: {url}\n时间: {datetime.now()}"
                self.send_telegram(msg)
            self.history[key] = current

这些脚本可以根据你的具体需求进行定制,需要我详细解释某个特定部分吗？