IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

自动对比网站内容变化的脚本

wen 实用脚本 3

本文目录导读：

自动对比网站内容变化的脚本

  1. 完整脚本实现
  2. 配置文件示例 (config.json)
  3. 使用说明

变化的Python脚本,这个脚本会定期检查指定网页的内容变化，并在检测到变化时发送通知。

完整脚本实现

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""监控脚本
功能：自动检测网站内容变化，支持邮件通知和日志记录
"""
import hashlib
import time
import logging
import smtplib
import json
import os
from email.mime.text import MIMEText
from email.header import Header
from datetime import datetime
from urllib.parse import urlparse
import requests
from bs4 import BeautifulSoup
import schedule
from difflib import unified_diff
import argparse
class SiteMonitor:
    def __init__(self, config_file='config.json'):
        self.config_file = config_file
        self.config = self.load_config()
        self.setup_logging()
    def load_config(self):
        """加载配置文件"""
        if os.path.exists(self.config_file):
            with open(self.config_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        else:
            # 默认配置
            return {
                "sites": [
                    {
                        "url": "https://example.com",
                        "name": "Example Site",
                        "check_interval": 300,
                        "compare_html": True
                    }
                ],
                "email": {
                    "enabled": False,
                    "smtp_server": "smtp.gmail.com",
                    "smtp_port": 587,
                    "username": "your_email@gmail.com",
                    "password": "your_app_password",
                    "to": ["recipient@example.com"]
                },
                "log_file": "site_monitor.log"
            }
    def setup_logging(self):
        """设置日志系统"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(self.config.get('log_file', 'site_monitor.log')),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def fetch_page(self, url):
        """获取网页内容"""
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
        try:
            response = requests.get(url, headers=headers, timeout=30)
            response.raise_for_status()
            return response.text
        except Exception as e:
            self.logger.error(f"获取页面失败 {url}: {str(e)}")
            return None
    def hash_content(self, content):
        """计算内容哈希值"""
        return hashlib.sha256(content.encode('utf-8')).hexdigest()
    def extract_content(self, html, compare_html=True):
        """提取比较内容"""
        if compare_html:
            return html
        else:
            # 提取纯文本内容
            soup = BeautifulSoup(html, 'html.parser')
            for script in soup(["script", "style"]):
                script.decompose()
            return soup.get_text(separator='\n', strip=True)
    def diff_content(self, old_content, new_content):
        """比较内容差异"""
        old_lines = old_content.splitlines()
        new_lines = new_content.splitlines()
        diff = list(unified_diff(old_lines, new_lines, lineterm='', n=2))
        return '\n'.join(diff)
    def save_snapshot(self, site, content):
        """保存内容快照"""
        snapshot_dir = 'snapshots'
        if not os.path.exists(snapshot_dir):
            os.makedirs(snapshot_dir)
        site_hash = hashlib.md5(site.encode()).hexdigest()[:8]
        filename = f"{snapshot_dir}/{site_hash}_latest.html"
        with open(filename, 'w', encoding='utf-8') as f:
            f.write(content)
        return filename
    def load_snapshot(self, site):
        """加载上次快照"""
        snapshot_dir = 'snapshots'
        site_hash = hashlib.md5(site.encode()).hexdigest()[:8]
        filename = f"{snapshot_dir}/{site_hash}_latest.html"
        if os.path.exists(filename):
            with open(filename, 'r', encoding='utf-8') as f:
                return f.read()
        return None
    def send_email(self, subject, content, site):
        """发送邮件通知"""
        if not self.config.get('email', {}).get('enabled', False):
            return False
        email_config = self.config['email']
        msg = MIMEText(content, 'html', 'utf-8')
        msg['Subject'] = Header(subject, 'utf-8')
        msg['From'] = email_config['username']
        msg['To'] = ', '.join(email_config['to'])
        try:
            server = smtplib.SMTP(email_config['smtp_server'], email_config['smtp_port'])
            server.starttls()
            server.login(email_config['username'], email_config['password'])
            server.sendmail(email_config['username'], email_config['to'], msg.as_string())
            server.quit()
            self.logger.info(f"邮件发送成功: {subject}")
            return True
        except Exception as e:
            self.logger.error(f"邮件发送失败: {str(e)}")
            return False
    def check_site(self, site):
        """检查单个网站"""
        url = site['url']
        name = site.get('name', url)
        compare_html = site.get('compare_html', True)
        self.logger.info(f"检查网站: {name} ({url})")
        # 获取新内容
        new_content = self.fetch_page(url)
        if not new_content:
            return False
        # 提取比较内容
        new_compare = self.extract_content(new_content, compare_html)
        new_hash = self.hash_content(new_compare)
        # 加载上次快照
        old_compare = self.load_snapshot(url)
        if old_compare is None:
            # 首次检查，保存快照
            self.save_snapshot(url, new_content)
            self.logger.info(f"首次检查 {name}，已保存基准快照")
            return True
        # 计算旧哈希
        old_hash = self.hash_content(old_compare)
        if old_hash != new_hash:
            # 内容发生变化
            self.logger.warning(f"网站内容发生变化: {name}")
            # 生成差异报告
            diff_text = self.diff_content(old_compare, new_compare)
            # 保存新快照
            self.save_snapshot(url, new_content)
            # 发送通知
            if self.config.get('email', {}).get('enabled', False):
                subject = f"网站内容变更: {name}"
                content = f"""
                <html>
                <body>
                    <h2 style="color: #e74c3c;">网站内容变更通知</h2>
                    <p><strong>网站名称:</strong> {name}</p>
                    <p><strong>URL:</strong> <a href="{url}">{url}</a></p>
                    <p><strong>检测时间:</strong> {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
                    <p><strong>变更摘要:</strong></p>
                    <pre style="background-color: #f4f4f4; padding: 10px; border: 1px solid #ddd">
{diff_text}
                    </pre>
                </body>
                </html>
                """
                self.send_email(subject, content, url)
            return True
        else:
            self.logger.info(f"网站 {name} 内容无变化")
            return False
    def run_all(self):
        """运行所有网站检查"""
        for site in self.config['sites']:
            self.check_site(site)
    def schedule_checks(self):
        """计划任务调度"""
        for site in self.config['sites']:
            interval = site.get('check_interval', 300)
            schedule.every(interval).seconds.do(self.check_site, site)
        self.logger.info(f"开始计划任务，共 {len(self.config['sites'])} 个网站")
        while True:
            schedule.run_pending()
            time.sleep(1)
def main():
    parser = argparse.ArgumentParser(description='网站内容监控脚本')
    parser.add_argument('--config', default='config.json', help='配置文件路径')
    parser.add_argument('--once', action='store_true', help='只运行一次')
    parser.add_argument('--add-site', nargs=3, metavar=('URL', 'NAME', 'INTERVAL'), 
                       help='添加网站到配置')
    args = parser.parse_args()
    monitor = SiteMonitor(args.config)
    if args.add_site:
        # 添加网站到配置
        url, name, interval = args.add_site
        # 读取现有配置
        config = monitor.config
        config['sites'].append({
            'url': url,
            'name': name,
            'check_interval': int(interval),
            'compare_html': True
        })
        # 保存配置
        with open(args.config, 'w', encoding='utf-8') as f:
            json.dump(config, f, indent=2, ensure_ascii=False)
        print(f"已添加网站: {name} ({url})")
        return
    if args.once:
        # 只运行一次检查
        monitor.run_all()
    else:
        # 计划任务模式
        monitor.schedule_checks()
if __name__ == "__main__":
    main()

配置文件示例 (config.json)

{
  "sites": [
    {
      "url": "https://news.ycombinator.com/",
      "name": "Hacker News",
      "check_interval": 300,
      "compare_html": false
    },
    {
      "url": "https://github.com/trending",
      "name": "GitHub Trending",
      "check_interval": 600,
      "compare_html": true
    }
  ],
  "email": {
    "enabled": false,
    "smtp_server": "smtp.gmail.com",
    "smtp_port": 587,
    "username": "your_email@gmail.com",
    "password": "your_app_password",
    "to": ["recipient@example.com"]
  },
  "log_file": "site_monitor.log"
}

使用说明

安装依赖

pip install requests beautifulsoup4 schedule

基础用法

只检查一次：

python site_monitor.py --once

持续监控：

python site_monitor.py

添加新网站：

python site_monitor.py --add-site "https://example.com" "Example" 300

功能特点

  • ✅ 多网站监控：支持同时监控多个网站
  • ✅ 内容比较：支持HTML全文或纯文本比较
  • ✅ 差异报告：生成详细的变更差异报告
  • ✅ 邮件通知：可选邮件通知功能
  • ✅ 日志记录：完整的日志记录
  • ✅ 定时检查：灵活设置检查间隔

高级配置

邮件通知设置

  1. 在 config.json 中启用邮件功能
  2. 对于Gmail,需要生成App密码
  3. 可以配置多个接收邮箱

自定义检查策略

  • compare_html：设为true比较完整HTML，false只比较文本内容
  • check_interval：设置检查频率（秒），最小可设置30秒

注意事项

  1. 尊重robots.txt：检查网站的robots.txt文件
  2. 合理设置时间间隔：避免对服务器造成压力
  3. 代理支持：如果需要代理，可以在 fetch_page 方法中添加
  4. 数据存储：快照保存在 snapshots/ 目录

这个脚本提供了完整的网站内容监控功能,你可以根据实际需求进行修改和扩展。

上一篇如何写一个脚本自动更新脚本本身

下一篇如何写一个脚本远程协助

相关文章

抱歉，评论功能暂时关闭!