使用Python实现网站可用性监控脚本

作者:

文章目录
  • 这是一个专业的网站可用性监控脚本,能够持续监控一个或多个网站的在线状态和响应时间。该脚本具备以下核心功能: 多网站并发监控:同时监控多个网站,提高监控效率 响应时间测量:精确测量网站的响应时间,包括DNS解析、连接建立、内容传输等各阶段耗时 状态码检测:检查HTTP状态码,识别网站是否正常响应 智能告警机制:当网站出现故障或响应时间超过阈值时自动发送告警 历史数据记录:将监控数据保存到CSV文件,便于后续分析 可视化报表生成:生成响应时间趋势图和可用性统计图 灵活配置选项:支持自定义监控间隔、超时时间、告警阈值等参数
  • #!/usr/bin/env python3 # -*- coding: utf-8 -*- “”” 网站可用性监控脚本 功能:监控网站可用性、响应时间、生成报告和告警 作者:Cline 版本:1.0 “”” import requests import time import csv import argparse import logging from datetime import datetime import matplotlib.pyplot as plt import pandas as pd from concurrent.futures import ThreadPoolExecutor, as_completed import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart import ssl import sys import os # 配置日志 logging.basicConfig( level=logging.INFO, format=’%(asctime)s – %(levelname)s – %(message)s’, handlers=[ logging.FileHandler(‘website_monitor.log’), logging.StreamHandler(sys.stdout) ] ) logger = logging.getLogger(__name__) class WebsiteMonitor: def __init__(self, config): self.urls = config[‘urls’] self.interval = config.get(‘interval’, 60) # 监控间隔(秒) self.timeout = config.get(‘timeout’, 10) # 请求超时时间(秒) self.response_threshold = config.get(‘response_threshold’, 5.0) # 响应时间阈值(秒) self.csv_file = config.get(‘csv_file’, ‘monitor_data.csv’) self.email_config = config.get(’email’, {}) self.data_buffer = [] # 数据缓存 def check_website(self, url): “””检查单个网站的可用性””” start_time = time.time() timestamp = datetime.now() try: # 发送HTTP请求 response = requests.get( url, timeout=self.timeout, headers={‘User-Agent’: ‘Website-Monitor/1.0’} ) end_time = time.time() response_time = end_time – start_time # 记录监控数据 data = { ‘timestamp’: timestamp, ‘url’: url, ‘status_code’: response.status_code, ‘response_time’: round(response_time, 3), ‘is_available’: response.status_code < 400, ‘content_length’: len(response.content) } # 检查响应时间和状态码 if response_time > self.response_threshold: logger.warning(f”网站 {url} 响应时间过长: {response_time:.3f}s”) self.send_alert(f”网站 {url} 响应时间过长 ({response_time:.3f}s)”) if not data[‘is_available’]: logger.error(f”网站 {url} 不可用,状态码: {response.status_code}”) self.send_alert(f”网站 {url} 不可用 (状态码: {response.status_code})”) logger.info(f”网站 {url} 检查完成 – 状态码: {response.status_code}, 响应时间: {response_time:.3f}s”) return data except requests.exceptions.ConnectionError: logger.error(f”无法连接到网站 {url}”) self.send_alert(f”网站 {url} 连接失败”) return { ‘timestamp’: timestamp, ‘url’: url, ‘status_code’: None, ‘response_time’: None, ‘is_available’: False, ‘content_length’: 0 } except requests.exceptions.Timeout: logger.error(f”网站 {url} 请求超时”) self.send_alert(f”网站 {url} 响应超时”) return { ‘timestamp’: timestamp, ‘url’: url, ‘status_code’: None, ‘response_time’: self.timeout, ‘is_available’: False, ‘content_length’: 0 } except Exception as e: logger.error(f”检查网站 {url} 时发生未知错误: {str(e)}”) self.send_alert(f”网站 {url} 监控异常: {str(e)}”) return { ‘timestamp’: timestamp, ‘url’: url, ‘status_code’: None, ‘response_time’: None, ‘is_available’: False, ‘content_length’: 0 } def monitor_all_websites(self): “””并发监控所有网站””” logger.info(“开始监控所有网站…”) with ThreadPoolExecutor(max_workers=5) as executor: # 提交所有监控任务 future_to_url = { executor.submit(self.check_website, url): url for url in self.urls } # 收集结果 results = [] for future in as_completed(future_to_url): url = future_to_url[future] try: result = future.result() results.append(result) self.save_to_csv(result) except Exception as e: logger.error(f”处理网站 {url} 的结果时出错: {str(e)}”) return results def save_to_csv(self, data): “””保存数据到CSV文件””” try: # 确保目录存在 os.makedirs(os.path.dirname(self.csv_file) if os.path.dirname(self.csv_file) else ‘.’, exist_ok=True) # 写入CSV文件 file_exists = os.path.exists(self.csv_file) with open(self.csv_file, ‘a’, newline=”, encoding=’utf-8′) as f: writer = csv.DictWriter(f, fieldnames=data.keys()) if not file_exists: writer.writeheader() writer.writerow(data) logger.debug(f”数据已保存到 {self.csv_file}”) except Exception as e: logger.error(f”保存数据到CSV文件时出错: {str(e)}”) def generate_report(self, days=7): “””生成监控报告””” try: # 读取CSV数据 if not os.path.exists(self.csv_file): logger.warning(“监控数据文件不存在”) return df = pd.read_csv(self.csv_file) df[‘timestamp’] = pd.to_datetime(df[‘timestamp’]) # 筛选最近N天的数据 cutoff_date = datetime.now() – pd.Timedelta(days=days) recent_data = df[df[‘timestamp’] >= cutoff_date] if recent_data.empty: logger.warning(“最近没有监控数据”) return # 生成可用性统计 availability_stats = recent_data.groupby(‘url’).agg({ ‘is_available’: [‘count’, ‘sum’], ‘response_time’: [‘mean’, ‘min’, ‘max’] }).round(3) print(“n=== 网站可用性监控报告 ===”) print(f”统计周期: 最近{days}天”) print(f”数据总量: {len(recent_data)} 条记录”) print(“n各网站统计:”) for url in recent_data[‘url’].unique(): url_data = recent_data[recent_data[‘url’] == url] total_checks = len(url_data) available_checks = url_data[‘is_available’].sum() availability = (available_checks / total_checks) * 100 avg_response = url_data[‘response_time’].mean() print(f”n网站: {url}”) print(f” 总检查次数: {total_checks}”) print(f” 可用次数: {int(available_checks)}”) print(f” 可用率: {availability:.2f}%”) print(f” 平均响应时间: {avg_response:.3f}s”) # 生成响应时间趋势图 self.plot_response_trend(recent_data) except Exception as e: logger.error(f”生成报告时出错: {str(e)}”) def plot_response_trend(self, data): “””绘制响应时间趋势图””” try: plt.figure(figsize=(12, 8)) # 为每个网站绘制响应时间趋势 for url in data[‘url’].unique(): url_data = data[data[‘url’] == url] plt.plot(url_data[‘timestamp’], url_data[‘response_time’], marker=’o’, label=url, linewidth=1, markersize=3) plt.xlabel(‘时间’) plt.ylabel(‘响应时间 (秒)’) plt.title(‘网站响应时间趋势’) plt.legend() plt.grid(True, alpha=0.3) plt.xticks(rotation=45) plt.tight_layout() # 保存图表 chart_file = ‘response_trend.png’ plt.savefig(chart_file, dpi=300, bbox_inches=’tight’) plt.close() logger.info(f”响应时间趋势图已保存到 {chart_file}”) except Exception as e: logger.error(f”绘制响应时间趋势图时出错: {str(e)}”) def send_alert(self, message): “””发送告警邮件””” if not self.email_config: return try: # 创建邮件内容 msg = MIMEMultipart() msg[‘From’] = self.email_config[‘sender’] msg[‘To’] = self.email_config[‘recipient’] msg[‘Subject’] = f”网站监控告警 – {datetime.now().strftime(‘%Y-%m-%d %H:%M:%S’)}” body = f””” 网站监控系统告警: {message} 时间: {datetime.now().strftime(‘%Y-%m-%d %H:%M:%S’)} “”” msg.attach(MIMEText(body, ‘plain’)) # 发送邮件 context = ssl.create_default_context() with smtplib.SMTP(self.email_config[‘smtp_server’], self.email_config[‘smtp_port’]) as server: server.starttls(context=context) server.login(self.email_config[‘sender’], self.email_config[‘password’]) server.send_message(msg) logger.info(“告警邮件发送成功”) except Exception as e: logger.error(f”发送告警邮件时出错: {str(e)}”) def run_continuous_monitoring(self): “””持续监控模式””” logger.info(f”启动持续监控模式,监控间隔: {self.interval}秒”) try: while True: self.monitor_all_websites() logger.info(f”等待 {self.interval} 秒后进行下次监控…”) time.sleep(self.interval) except KeyboardInterrupt: logger.info(“监控已停止”) def main(): parser = argparse.ArgumentParser(description=’网站可用性监控工具’) parser.add_argument(‘urls’, nargs=’+’, help=’要监控的网站URL列表’) parser.add_argument(‘-i’, ‘–interval’, type=int, default=60, help=’监控间隔(秒)’) parser.add_argument(‘-t’, ‘–timeout’, type=int, default=10, help=’请求超时时间(秒)’) parser.add_argument(‘-r’, ‘–threshold’, type=float, default=5.0, help=’响应时间阈值(秒)’) parser.add_argument(‘-c’, ‘–csv’, default=’monitor_data.csv’, help=’CSV数据文件路径’) parser.add_argument(‘-m’, ‘–mode’, choices=[‘once’, ‘continuous’], default=’once’, help=’监控模式’) parser.add_argument(‘–report’, action=’store_true’, help=’生成监控报告’) parser.add_argument(‘–days’, type=int, default=7, help=’报告统计天数’) args = parser.parse_args() # 配置监控参数 config = { ‘urls’: args.urls, ‘interval’: args.interval, ‘timeout’: args.timeout, ‘response_threshold’: args.threshold, ‘csv_file’: args.csv } # 创建监控实例 monitor = WebsiteMonitor(config) if args.report: # 生成报告模式 monitor.generate_report(args.days) elif args.mode == ‘continuous’: # 持续监控模式 monitor.run_continuous_monitoring() else: # 单次监控模式 monitor.monitor_all_websites() if __name__ == ‘__main__’: main()
  • 1. 并发处理 脚本使用线程池并发处理多个网站的监控,提高监控效率。 2. 内存管理 合理控制数据缓存大小,避免内存泄漏。 3. 错误恢复 完善的异常处理机制确保脚本在遇到错误时能够继续运行。
  • 1. 日志安全 敏感信息不会记录到日志文件中。 2. 网络安全 使用HTTPS协议进行网络通信。 3. 权限控制 脚本运行所需的最小权限原则。
  • 数据库存储:将监控数据存储到数据库中,便于大规模数据分析 Web界面:开发Web界面展示监控数据和图表 移动端告警:集成短信、微信等移动端告警方式 机器学习预测:使用机器学习算法预测 网站故障 分布式部署:在多个地理位置部署监控节点,获得更全面的监控视角 这个网站可用性监控脚本是一个功能完整、易于使用且高度可定制的监控工具,能够满足从小型个人网站到大型企业系统的各种监控需求。 以上就是使用Python实现网站可用性监控脚本的详细内容,更多关于Python网站监控的资料请关注风君子博客其它相关文章! 您可能感兴趣的文章: Python实现监控网站变化并自动通知 如何利用Python监控别人的网站 用Python监控你的朋友都在浏览哪些网站? Python实时监控网站浏览记录实现过程详解 python实现批量监控网站 利用Python自动监控网站并发送邮件告警的方法 python监控网站运行异常并发送邮件的方法
  • 目录
    • 功能介绍
    • 场景应用
      • 1. 企业IT运维
      • 2. 电商平台运营
      • 3. 内容网站管理
      • 4. 个人站长维护
    • 报错处理
      • 1. 网络连接异常
      • 2. DNS解析失败
      • 3. HTTP状态码异常
      • 4. SSL证书验证失败
      • 5. 文件操作异常
    • 代码实现
      • 使用说明
        • 1. 基本使用
        • 2. 持续监控模式
        • 3. 生成报告
        • 4. 配置邮件告警
      • 高级特性
        • 1. 数据分析
        • 2. 自定义告警
        • 3. 集成到监控系统
      • 性能优化
        • 安全考虑
          • 扩展建议

            这是一个专业的网站可用性监控脚本,能够持续监控一个或多个网站的在线状态和响应时间。该脚本具备以下核心功能:

            • 多网站并发监控:同时监控多个网站,提高监控效率
            • 响应时间测量:精确测量网站的响应时间,包括DNS解析、连接建立、内容传输等各阶段耗时
            • 状态码检测:检查HTTP状态码,识别网站是否正常响应
            • 智能告警机制:当网站出现故障或响应时间超过阈值时自动发送告警
            • 历史数据记录:将监控数据保存到CSV文件,便于后续分析
            • 可视化报表生成:生成响应时间趋势图和可用性统计图
            • 灵活配置选项:支持自定义监控间隔、超时时间、告警阈值等参数

            • 监控公司官网、内部系统、客户门户等关键业务系统的可用性
            • 及时发现系统故障,缩短故障响应时间
            • 为SLA考核提供客观数据支撑

            • 实时监控购物网站、支付接口、物流查询等核心服务
            • 预防因网站不可用导致的订单流失
            • 分析网站性能瓶颈,优化用户体验

            • 监控新闻网站、博客平台、视频网站等内容服务
            • 确保高并发访问时的稳定性和响应速度
            • 为CDN优化和服务器扩容提供数据依据

            • 监控个人博客、作品集网站等个人项目的运行状态
            • 及时发现并处理网站异常,保障访问体验
            • 积累网站运行数据,为优化提供参考

            try:
                response = requests.get(url, timeout=timeout)
            except requests.exceptions.ConnectionError:
                log_error(f"无法连接到 {url},网络连接异常")
                send_alert(f"网站 {url} 连接失败")
            except requests.exceptions.Timeout:
                log_error(f"请求 {url} 超时")
                send_alert(f"网站 {url} 响应超时")
            

            except requests.exceptions.InvalidURL:
                log_error(f"无效的URL地址: {url}")
                send_alert(f"网站 {url} URL格式错误")
            except socket.gaierror:
                log_error(f"DNS解析失败: {url}")
                send_alert(f"网站 {url} DNS解析失败")
            

            if response.status_code >= 500:
                log_error(f"服务器错误 {response.status_code} for {url}")
                send_alert(f"网站 {url} 服务器内部错误")
            elif response.status_code >= 400:
                log_warning(f"客户端错误 {response.status_code} for {url}")
            

            except requests.exceptions.SSLError as e:
                log_error(f"SSL证书验证失败: {url} - {str(e)}")
                send_alert(f"网站 {url} SSL证书问题")
            

            try:
                df.to_csv(csv_file, mode='a', header=not os.path.exists(csv_file), index=False)
            except PermissionError:
                log_error(f"无权限写入文件: {csv_file}")
            except IOError as e:
                log_error(f"文件写入错误: {str(e)}")
            

            #!/usr/bin/env python3
            # -*- coding: utf-8 -*-
            """
            网站可用性监控脚本
            功能:监控网站可用性、响应时间、生成报告和告警
            作者:Cline
            版本:1.0
            """
            
            import requests
            import time
            import csv
            import argparse
            import logging
            from datetime import datetime
            import matplotlib.pyplot as plt
            import pandas as pd
            from concurrent.futures import ThreadPoolExecutor, as_completed
            import smtplib
            from email.mime.text import MIMEText
            from email.mime.multipart import MIMEMultipart
            import ssl
            import sys
            import os
            
            # 配置日志
            logging.basicConfig(
                level=logging.INFO,
                format='%(asctime)s - %(levelname)s - %(message)s',
                handlers=[
                    logging.FileHandler('website_monitor.log'),
                    logging.StreamHandler(sys.stdout)
                ]
            )
            logger = logging.getLogger(__name__)
            
            class WebsiteMonitor:
                def __init__(self, config):
                    self.urls = config['urls']
                    self.interval = config.get('interval', 60)  # 监控间隔(秒)
                    self.timeout = config.get('timeout', 10)    # 请求超时时间(秒)
                    self.response_threshold = config.get('response_threshold', 5.0)  # 响应时间阈值(秒)
                    self.csv_file = config.get('csv_file', 'monitor_data.csv')
                    self.email_config = config.get('email', {})
                    self.data_buffer = []  # 数据缓存
                    
                def check_website(self, url):
                    """检查单个网站的可用性"""
                    start_time = time.time()
                    timestamp = datetime.now()
                    
                    try:
                        # 发送HTTP请求
                        response = requests.get(
                            url, 
                            timeout=self.timeout,
                            headers={'User-Agent': 'Website-Monitor/1.0'}
                        )
                        
                        end_time = time.time()
                        response_time = end_time - start_time
                        
                        # 记录监控数据
                        data = {
                            'timestamp': timestamp,
                            'url': url,
                            'status_code': response.status_code,
                            'response_time': round(response_time, 3),
                            'is_available': response.status_code < 400,
                            'content_length': len(response.content)
                        }
                        
                        # 检查响应时间和状态码
                        if response_time > self.response_threshold:
                            logger.warning(f"网站 {url} 响应时间过长: {response_time:.3f}s")
                            self.send_alert(f"网站 {url} 响应时间过长 ({response_time:.3f}s)")
                            
                        if not data['is_available']:
                            logger.error(f"网站 {url} 不可用,状态码: {response.status_code}")
                            self.send_alert(f"网站 {url} 不可用 (状态码: {response.status_code})")
                            
                        logger.info(f"网站 {url} 检查完成 - 状态码: {response.status_code}, 响应时间: {response_time:.3f}s")
                        return data
                        
                    except requests.exceptions.ConnectionError:
                        logger.error(f"无法连接到网站 {url}")
                        self.send_alert(f"网站 {url} 连接失败")
                        return {
                            'timestamp': timestamp,
                            'url': url,
                            'status_code': None,
                            'response_time': None,
                            'is_available': False,
                            'content_length': 0
                        }
                        
                    except requests.exceptions.Timeout:
                        logger.error(f"网站 {url} 请求超时")
                        self.send_alert(f"网站 {url} 响应超时")
                        return {
                            'timestamp': timestamp,
                            'url': url,
                            'status_code': None,
                            'response_time': self.timeout,
                            'is_available': False,
                            'content_length': 0
                        }
                        
                    except Exception as e:
                        logger.error(f"检查网站 {url} 时发生未知错误: {str(e)}")
                        self.send_alert(f"网站 {url} 监控异常: {str(e)}")
                        return {
                            'timestamp': timestamp,
                            'url': url,
                            'status_code': None,
                            'response_time': None,
                            'is_available': False,
                            'content_length': 0
                        }
                
                def monitor_all_websites(self):
                    """并发监控所有网站"""
                    logger.info("开始监控所有网站...")
                    
                    with ThreadPoolExecutor(max_workers=5) as executor:
                        # 提交所有监控任务
                        future_to_url = {
                            executor.submit(self.check_website, url): url 
                            for url in self.urls
                        }
                        
                        # 收集结果
                        results = []
                        for future in as_completed(future_to_url):
                            url = future_to_url[future]
                            try:
                                result = future.result()
                                results.append(result)
                                self.save_to_csv(result)
                            except Exception as e:
                                logger.error(f"处理网站 {url} 的结果时出错: {str(e)}")
                                
                    return results
                
                def save_to_csv(self, data):
                    """保存数据到CSV文件"""
                    try:
                        # 确保目录存在
                        os.makedirs(os.path.dirname(self.csv_file) if os.path.dirname(self.csv_file) else '.', exist_ok=True)
                        
                        # 写入CSV文件
                        file_exists = os.path.exists(self.csv_file)
                        with open(self.csv_file, 'a', newline='', encoding='utf-8') as f:
                            writer = csv.DictWriter(f, fieldnames=data.keys())
                            if not file_exists:
                                writer.writeheader()
                            writer.writerow(data)
                            
                        logger.debug(f"数据已保存到 {self.csv_file}")
                    except Exception as e:
                        logger.error(f"保存数据到CSV文件时出错: {str(e)}")
                
                def generate_report(self, days=7):
                    """生成监控报告"""
                    try:
                        # 读取CSV数据
                        if not os.path.exists(self.csv_file):
                            logger.warning("监控数据文件不存在")
                            return
                            
                        df = pd.read_csv(self.csv_file)
                        df['timestamp'] = pd.to_datetime(df['timestamp'])
                        
                        # 筛选最近N天的数据
                        cutoff_date = datetime.now() - pd.Timedelta(days=days)
                        recent_data = df[df['timestamp'] >= cutoff_date]
                        
                        if recent_data.empty:
                            logger.warning("最近没有监控数据")
                            return
                            
                        # 生成可用性统计
                        availability_stats = recent_data.groupby('url').agg({
                            'is_available': ['count', 'sum'],
                            'response_time': ['mean', 'min', 'max']
                        }).round(3)
                        
                        print("n=== 网站可用性监控报告 ===")
                        print(f"统计周期: 最近{days}天")
                        print(f"数据总量: {len(recent_data)} 条记录")
                        print("n各网站统计:")
                        for url in recent_data['url'].unique():
                            url_data = recent_data[recent_data['url'] == url]
                            total_checks = len(url_data)
                            available_checks = url_data['is_available'].sum()
                            availability = (available_checks / total_checks) * 100
                            avg_response = url_data['response_time'].mean()
                            
                            print(f"n网站: {url}")
                            print(f"  总检查次数: {total_checks}")
                            print(f"  可用次数: {int(available_checks)}")
                            print(f"  可用率: {availability:.2f}%")
                            print(f"  平均响应时间: {avg_response:.3f}s")
                            
                        # 生成响应时间趋势图
                        self.plot_response_trend(recent_data)
                        
                    except Exception as e:
                        logger.error(f"生成报告时出错: {str(e)}")
                
                def plot_response_trend(self, data):
                    """绘制响应时间趋势图"""
                    try:
                        plt.figure(figsize=(12, 8))
                        
                        # 为每个网站绘制响应时间趋势
                        for url in data['url'].unique():
                            url_data = data[data['url'] == url]
                            plt.plot(url_data['timestamp'], url_data['response_time'], 
                                    marker='o', label=url, linewidth=1, markersize=3)
                        
                        plt.xlabel('时间')
                        plt.ylabel('响应时间 (秒)')
                        plt.title('网站响应时间趋势')
                        plt.legend()
                        plt.grid(True, alpha=0.3)
                        plt.xticks(rotation=45)
                        plt.tight_layout()
                        
                        # 保存图表
                        chart_file = 'response_trend.png'
                        plt.savefig(chart_file, dpi=300, bbox_inches='tight')
                        plt.close()
                        
                        logger.info(f"响应时间趋势图已保存到 {chart_file}")
                        
                    except Exception as e:
                        logger.error(f"绘制响应时间趋势图时出错: {str(e)}")
                
                def send_alert(self, message):
                    """发送告警邮件"""
                    if not self.email_config:
                        return
                        
                    try:
                        # 创建邮件内容
                        msg = MIMEMultipart()
                        msg['From'] = self.email_config['sender']
                        msg['To'] = self.email_config['recipient']
                        msg['Subject'] = f"网站监控告警 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
                        
                        body = f"""
            网站监控系统告警:
            
            {message}
            
            时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
                        """
                        msg.attach(MIMEText(body, 'plain'))
                        
                        # 发送邮件
                        context = ssl.create_default_context()
                        with smtplib.SMTP(self.email_config['smtp_server'], self.email_config['smtp_port']) as server:
                            server.starttls(context=context)
                            server.login(self.email_config['sender'], self.email_config['password'])
                            server.send_message(msg)
                            
                        logger.info("告警邮件发送成功")
                        
                    except Exception as e:
                        logger.error(f"发送告警邮件时出错: {str(e)}")
                
                def run_continuous_monitoring(self):
                    """持续监控模式"""
                    logger.info(f"启动持续监控模式,监控间隔: {self.interval}秒")
                    
                    try:
                        while True:
                            self.monitor_all_websites()
                            logger.info(f"等待 {self.interval} 秒后进行下次监控...")
                            time.sleep(self.interval)
                            
                    except KeyboardInterrupt:
                        logger.info("监控已停止")
            
            def main():
                parser = argparse.ArgumentParser(description='网站可用性监控工具')
                parser.add_argument('urls', nargs='+', help='要监控的网站URL列表')
                parser.add_argument('-i', '--interval', type=int, default=60, help='监控间隔(秒)')
                parser.add_argument('-t', '--timeout', type=int, default=10, help='请求超时时间(秒)')
                parser.add_argument('-r', '--threshold', type=float, default=5.0, help='响应时间阈值(秒)')
                parser.add_argument('-c', '--csv', default='monitor_data.csv', help='CSV数据文件路径')
                parser.add_argument('-m', '--mode', choices=['once', 'continuous'], default='once', help='监控模式')
                parser.add_argument('--report', action='store_true', help='生成监控报告')
                parser.add_argument('--days', type=int, default=7, help='报告统计天数')
                
                args = parser.parse_args()
                
                # 配置监控参数
                config = {
                    'urls': args.urls,
                    'interval': args.interval,
                    'timeout': args.timeout,
                    'response_threshold': args.threshold,
                    'csv_file': args.csv
                }
                
                # 创建监控实例
                monitor = WebsiteMonitor(config)
                
                if args.report:
                    # 生成报告模式
                    monitor.generate_report(args.days)
                elif args.mode == 'continuous':
                    # 持续监控模式
                    monitor.run_continuous_monitoring()
                else:
                    # 单次监控模式
                    monitor.monitor_all_websites()
            
            if __name__ == '__main__':
                main()
            

            # 监控单个网站
            python website_monitor.py https://www.google.com
            
            # 监控多个网站
            python website_monitor.py https://www.google.com https://www.github.com https://www.stackoverflow.com
            
            # 设置监控间隔为30秒
            python website_monitor.py https://www.google.com -i 30
            
            # 设置超时时间为5秒
            python website_monitor.py https://www.google.com -t 5
            
            # 设置响应时间阈值为2秒
            python website_monitor.py https://www.google.com -r 2.0
            

            # 启动持续监控
            python website_monitor.py https://www.google.com https://www.github.com -m continuous -i 60
            

            # 生成最近7天的监控报告
            python website_monitor.py --report
            
            # 生成最近30天的监控报告
            python website_monitor.py --report --days 30
            

            在脚本中配置邮箱参数:

            email_config = {
                'smtp_server': 'smtp.gmail.com',
                'smtp_port': 587,
                'sender': 'your_email@gmail.com',
                'password': 'your_password',
                'recipient': 'admin@company.com'
            }
            

            脚本会自动将监控数据保存到CSV文件中,便于后续分析:

            import pandas as pd
            
            # 读取监控数据
            df = pd.read_csv('monitor_data.csv')
            df['timestamp'] = pd.to_datetime(df['timestamp'])
            
            # 分析网站可用性
            availability = df.groupby('url')['is_available'].mean() * 100
            print("网站可用率:")
            print(availability)
            

            可以通过修改脚本中的告警条件来自定义告警规则:

            # 添加自定义告警条件
            if response_time > 10.0:
                send_alert("网站响应极慢")
            elif status_code == 500:
                send_alert("服务器内部错误")
            

            可以将脚本集成到现有的监控系统中,定期执行并收集数据:

            # 添加到crontab中每分钟执行一次
            * * * * * /usr/bin/python3 /path/to/website_monitor.py https://yourwebsite.com >> /var/log/website_monitor.log 2>&1
            

            1. 并发处理

            脚本使用线程池并发处理多个网站的监控,提高监控效率。

            2. 内存管理

            合理控制数据缓存大小,避免内存泄漏。

            3. 错误恢复

            完善的异常处理机制确保脚本在遇到错误时能够继续运行。

            1. 日志安全

            敏感信息不会记录到日志文件中。

            2. 网络安全

            使用HTTPS协议进行网络通信。

            3. 权限控制

            脚本运行所需的最小权限原则。

            • 数据库存储:将监控数据存储到数据库中,便于大规模数据分析
            • Web界面:开发Web界面展示监控数据和图表
            • 移动端告警:集成短信、微信等移动端告警方式
            • 机器学习预测:使用机器学习算法预测 网站故障
            • 分布式部署:在多个地理位置部署监控节点,获得更全面的监控视角

            这个网站可用性监控脚本是一个功能完整、易于使用且高度可定制的监控工具,能够满足从小型个人网站到大型企业系统的各种监控需求。

            以上就是使用Python实现网站可用性监控脚本的详细内容,更多关于Python网站监控的资料请关注风君子博客其它相关文章!

            您可能感兴趣的文章:

            • Python实现监控网站变化并自动通知
            • 如何利用Python监控别人的网站
            • 用Python监控你的朋友都在浏览哪些网站?
            • Python实时监控网站浏览记录实现过程详解
            • python实现批量监控网站
            • 利用Python自动监控网站并发送邮件告警的方法
            • python监控网站运行异常并发送邮件的方法

            站内搜索