Python 如何实现一个网站可用性监控系统

我来给你一个完整的实现方案。这个监控系统会定期检查网站是否可访问，并记录响应时间和状态码。

import requests
import time
import logging
from datetime import datetime
import schedule
import json
from typing import Dict, Optional
import smtplib
from email.mime.text import MIMEText

class WebsiteMonitor:
    def __init__(self, config_file: str = 'config.json'):
        """
        初始化网站监控器
        
        Args:
            config_file: 配置文件路径，包含要监控的网站列表
        """
        self.load_config(config_file)
        self.setup_logging()
        
    def load_config(self, config_file: str):
        """加载配置文件"""
        try:
            with open(config_file, 'r') as f:
                config = json.load(f)
                self.websites = config.get('websites', [])
                self.check_interval = config.get('check_interval', 60)  # 默认60秒
                self.timeout = config.get('timeout', 10)  # 默认10秒超时
                self.alert_email = config.get('alert_email')
                self.smtp_config = config.get('smtp', {})
        except FileNotFoundError:
            # 默认配置
            self.websites = [
                {"url": "https://www.google.com", "name": "Google"},
                {"url": "https://github.com", "name": "GitHub"}
            ]
            self.check_interval = 60
            self.timeout = 10
            self.alert_email = None
            
    def setup_logging(self):
        """设置日志"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('website_monitor.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
        
    def check_website(self, website: Dict) -> Dict:
        """
        检查单个网站
        
        Returns:
            包含检查结果的字典
        """
        url = website['url']
        name = website.get('name', url)
        
        result = {
            'name': name,
            'url': url,
            'timestamp': datetime.now().isoformat(),
            'status': 'down',
            'response_time': None,
            'status_code': None,
            'error': None
        }
        
        try:
            start_time = time.time()
            response = requests.get(
                url, 
                timeout=self.timeout,
                headers={'User-Agent': 'WebsiteMonitor/1.0'}
            )
            response_time = (time.time() - start_time) * 1000  # 转换为毫秒
            
            result.update({
                'status': 'up' if response.status_code < 400 else 'down',
                'response_time': round(response_time, 2),
                'status_code': response.status_code
            })
            
            self.logger.info(f"{name} ({url}) - Status: {result['status']}, "
                           f"Response Time: {response_time:.2f}ms, "
                           f"Code: {response.status_code}")
            
        except requests.exceptions.Timeout:
            result['error'] = 'Timeout'
            self.logger.error(f"{name} ({url}) - Timeout after {self.timeout}s")
            
        except requests.exceptions.ConnectionError:
            result['error'] = 'Connection Error'
            self.logger.error(f"{name} ({url}) - Connection Error")
            
        except Exception as e:
            result['error'] = str(e)
            self.logger.error(f"{name} ({url}) - Error: {str(e)}")
            
        return result
    
    def send_alert(self, website: Dict, result: Dict):
        """发送警报邮件"""
        if not self.alert_email or not self.smtp_config:
            return
            
        subject = f"⚠️ Website Alert: {website['name']} is DOWN"
        body = f"""
        Website Monitoring Alert:
        
        Website: {website['name']}
        URL: {website['url']}
        Status: {result['status']}
        Time: {result['timestamp']}
        Error: {result.get('error', 'N/A')}
        Status Code: {result.get('status_code', 'N/A')}
        Response Time: {result.get('response_time', 'N/A')}ms
        """
        
        try:
            msg = MIMEText(body)
            msg['Subject'] = subject
            msg['From'] = self.smtp_config.get('from')
            msg['To'] = self.alert_email
            
            with smtplib.SMTP(self.smtp_config.get('host'), 
                            self.smtp_config.get('port', 587)) as server:
                if self.smtp_config.get('use_tls', True):
                    server.starttls()
                if self.smtp_config.get('username'):
                    server.login(self.smtp_config['username'], 
                               self.smtp_config['password'])
                server.send_message(msg)
                
            self.logger.info(f"Alert email sent for {website['name']}")
            
        except Exception as e:
            self.logger.error(f"Failed to send alert email: {str(e)}")
    
    def check_all_websites(self):
        """检查所有网站"""
        self.logger.info("Starting website checks...")
        
        for website in self.websites:
            result = self.check_website(website)
            
            # 如果网站宕机，发送警报
            if result['status'] == 'down':
                self.send_alert(website, result)
                
            # 这里可以添加将结果保存到数据库的逻辑
            # self.save_to_database(result)
    
    def run(self):
        """运行监控器"""
        self.logger.info(f"Website Monitor started. Checking {len(self.websites)} websites every {self.check_interval} seconds.")
        
        # 立即执行一次检查
        self.check_all_websites()
        
        # 设置定时任务
        schedule.every(self.check_interval).seconds.do(self.check_all_websites)
        
        try:
            while True:
                schedule.run_pending()
                time.sleep(1)
        except KeyboardInterrupt:
            self.logger.info("Website Monitor stopped by user.")

# 配置文件示例 (config.json)
"""
{
    "websites": [
        {"url": "https://www.google.com", "name": "Google"},
        {"url": "https://github.com", "name": "GitHub"},
        {"url": "https://www.example.com", "name": "Example"}
    ],
    "check_interval": 60,
    "timeout": 10,
    "alert_email": "admin@example.com",
    "smtp": {
        "host": "smtp.gmail.com",
        "port": 587,
        "use_tls": true,
        "username": "your-email@gmail.com",
        "password": "your-password",
        "from": "your-email@gmail.com"
    }
}
"""

if __name__ == "__main__":
    # 使用方式
    monitor = WebsiteMonitor('config.json')  # 传入配置文件
    monitor.run()
    
    # 或者使用默认配置
    # monitor = WebsiteMonitor()
    # monitor.run()

这个实现包含了核心功能：

1. 配置管理：通过JSON文件配置要监控的网站
2. HTTP检查：使用requests库检查网站状态
3. 定时监控：使用schedule库定期执行检查
4. 日志记录：记录所有检查结果到文件和控制台
5. 警报系统：网站宕机时发送邮件通知
6. 错误处理：处理超时、连接错误等异常情况

要运行这个系统，你需要安装依赖：

pip install requests schedule

然后创建配置文件，运行监控器即可。系统会持续监控所有配置的网站，并在网站不可用时发送警报。

这个方案足够简单直接，你可以根据需要扩展功能，比如添加数据库存储、Web界面、更多监控指标等。