本文目录导读:

- Python 基础监控脚本
- 更高级的监控(支持选择器)
- 命令行脚本(使用curl和diff)
- 使用现有工具 - changedetection.io
- Node.js 脚本示例
- 实际部署建议
- 关键优化点
- 完整示例(带通知)
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
关键优化点
- 减少请求频率:添加合理的时间间隔
- 条件监控:只关注特定内容变化
- 离线存储:保存快照以便对比
- 智能通知:
- 邮件(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
这些脚本可以根据你的具体需求进行定制,需要我详细解释某个特定部分吗?