使用Python實(shí)現(xiàn)網(wǎng)站可用性監(jiān)控腳本

更新時(shí)間：2025年12月05日 08:30:22 作者：零日失眠者

這篇文章主要為大家詳細(xì)介紹了如果使用Python實(shí)現(xiàn)一個(gè)網(wǎng)站可用性監(jiān)控腳本,能夠持續(xù)監(jiān)控一個(gè)或多個(gè)網(wǎng)站的在線狀態(tài)和響應(yīng)時(shí)間,需要的朋友可以參考下

功能介紹

這是一個(gè)專業(yè)的網(wǎng)站可用性監(jiān)控腳本，能夠持續(xù)監(jiān)控一個(gè)或多個(gè)網(wǎng)站的在線狀態(tài)和響應(yīng)時(shí)間。該腳本具備以下核心功能：

多網(wǎng)站并發(fā)監(jiān)控：同時(shí)監(jiān)控多個(gè)網(wǎng)站，提高監(jiān)控效率
響應(yīng)時(shí)間測(cè)量：精確測(cè)量網(wǎng)站的響應(yīng)時(shí)間，包括DNS解析、連接建立、內(nèi)容傳輸?shù)雀麟A段耗時(shí)
狀態(tài)碼檢測(cè)：檢查HTTP狀態(tài)碼，識(shí)別網(wǎng)站是否正常響應(yīng)
智能告警機(jī)制：當(dāng)網(wǎng)站出現(xiàn)故障或響應(yīng)時(shí)間超過(guò)閾值時(shí)自動(dòng)發(fā)送告警
歷史數(shù)據(jù)記錄：將監(jiān)控?cái)?shù)據(jù)保存到CSV文件，便于后續(xù)分析
可視化報(bào)表生成：生成響應(yīng)時(shí)間趨勢(shì)圖和可用性統(tǒng)計(jì)圖
靈活配置選項(xiàng)：支持自定義監(jiān)控間隔、超時(shí)時(shí)間、告警閾值等參數(shù)

場(chǎng)景應(yīng)用

1. 企業(yè)IT運(yùn)維

監(jiān)控公司官網(wǎng)、內(nèi)部系統(tǒng)、客戶門(mén)戶等關(guān)鍵業(yè)務(wù)系統(tǒng)的可用性
及時(shí)發(fā)現(xiàn)系統(tǒng)故障，縮短故障響應(yīng)時(shí)間
為SLA考核提供客觀數(shù)據(jù)支撐

2. 電商平臺(tái)運(yùn)營(yíng)

實(shí)時(shí)監(jiān)控購(gòu)物網(wǎng)站、支付接口、物流查詢等核心服務(wù)
預(yù)防因網(wǎng)站不可用導(dǎo)致的訂單流失
分析網(wǎng)站性能瓶頸，優(yōu)化用戶體驗(yàn)

3. 內(nèi)容網(wǎng)站管理

監(jiān)控新聞網(wǎng)站、博客平臺(tái)、視頻網(wǎng)站等內(nèi)容服務(wù)
確保高并發(fā)訪問(wèn)時(shí)的穩(wěn)定性和響應(yīng)速度
為CDN優(yōu)化和服務(wù)器擴(kuò)容提供數(shù)據(jù)依據(jù)

4. 個(gè)人站長(zhǎng)維護(hù)

監(jiān)控個(gè)人博客、作品集網(wǎng)站等個(gè)人項(xiàng)目的運(yùn)行狀態(tài)
及時(shí)發(fā)現(xiàn)并處理網(wǎng)站異常，保障訪問(wèn)體驗(yàn)
積累網(wǎng)站運(yùn)行數(shù)據(jù)，為優(yōu)化提供參考

報(bào)錯(cuò)處理

1. 網(wǎng)絡(luò)連接異常

try:
    response = requests.get(url, timeout=timeout)
except requests.exceptions.ConnectionError:
    log_error(f"無(wú)法連接到 {url}，網(wǎng)絡(luò)連接異常")
    send_alert(f"網(wǎng)站 {url} 連接失敗")
except requests.exceptions.Timeout:
    log_error(f"請(qǐng)求 {url} 超時(shí)")
    send_alert(f"網(wǎng)站 {url} 響應(yīng)超時(shí)")

2. DNS解析失敗

except requests.exceptions.InvalidURL:
    log_error(f"無(wú)效的URL地址: {url}")
    send_alert(f"網(wǎng)站 {url} URL格式錯(cuò)誤")
except socket.gaierror:
    log_error(f"DNS解析失敗: {url}")
    send_alert(f"網(wǎng)站 {url} DNS解析失敗")

3. HTTP狀態(tài)碼異常

if response.status_code >= 500:
    log_error(f"服務(wù)器錯(cuò)誤 {response.status_code} for {url}")
    send_alert(f"網(wǎng)站 {url} 服務(wù)器內(nèi)部錯(cuò)誤")
elif response.status_code >= 400:
    log_warning(f"客戶端錯(cuò)誤 {response.status_code} for {url}")

4. SSL證書(shū)驗(yàn)證失敗

except requests.exceptions.SSLError as e:
    log_error(f"SSL證書(shū)驗(yàn)證失敗: {url} - {str(e)}")
    send_alert(f"網(wǎng)站 {url} SSL證書(shū)問(wèn)題")

5. 文件操作異常

try:
    df.to_csv(csv_file, mode='a', header=not os.path.exists(csv_file), index=False)
except PermissionError:
    log_error(f"無(wú)權(quán)限寫(xiě)入文件: {csv_file}")
except IOError as e:
    log_error(f"文件寫(xiě)入錯(cuò)誤: {str(e)}")

代碼實(shí)現(xiàn)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
網(wǎng)站可用性監(jiān)控腳本
功能：監(jiān)控網(wǎng)站可用性、響應(yīng)時(shí)間、生成報(bào)告和告警
作者：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)  # 監(jiān)控間隔（秒）
        self.timeout = config.get('timeout', 10)    # 請(qǐng)求超時(shí)時(shí)間（秒）
        self.response_threshold = config.get('response_threshold', 5.0)  # 響應(yīng)時(shí)間閾值（秒）
        self.csv_file = config.get('csv_file', 'monitor_data.csv')
        self.email_config = config.get('email', {})
        self.data_buffer = []  # 數(shù)據(jù)緩存
        
    def check_website(self, url):
        """檢查單個(gè)網(wǎng)站的可用性"""
        start_time = time.time()
        timestamp = datetime.now()
        
        try:
            # 發(fā)送HTTP請(qǐng)求
            response = requests.get(
                url, 
                timeout=self.timeout,
                headers={'User-Agent': 'Website-Monitor/1.0'}
            )
            
            end_time = time.time()
            response_time = end_time - start_time
            
            # 記錄監(jiān)控?cái)?shù)據(jù)
            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)
            }
            
            # 檢查響應(yīng)時(shí)間和狀態(tài)碼
            if response_time > self.response_threshold:
                logger.warning(f"網(wǎng)站 {url} 響應(yīng)時(shí)間過(guò)長(zhǎng): {response_time:.3f}s")
                self.send_alert(f"網(wǎng)站 {url} 響應(yīng)時(shí)間過(guò)長(zhǎng) ({response_time:.3f}s)")
                
            if not data['is_available']:
                logger.error(f"網(wǎng)站 {url} 不可用，狀態(tài)碼: {response.status_code}")
                self.send_alert(f"網(wǎng)站 {url} 不可用 (狀態(tài)碼: {response.status_code})")
                
            logger.info(f"網(wǎng)站 {url} 檢查完成 - 狀態(tài)碼: {response.status_code}, 響應(yīng)時(shí)間: {response_time:.3f}s")
            return data
            
        except requests.exceptions.ConnectionError:
            logger.error(f"無(wú)法連接到網(wǎng)站 {url}")
            self.send_alert(f"網(wǎng)站 {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"網(wǎng)站 {url} 請(qǐng)求超時(shí)")
            self.send_alert(f"網(wǎng)站 {url} 響應(yīng)超時(shí)")
            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"檢查網(wǎng)站 {url} 時(shí)發(fā)生未知錯(cuò)誤: {str(e)}")
            self.send_alert(f"網(wǎng)站 {url} 監(jiān)控異常: {str(e)}")
            return {
                'timestamp': timestamp,
                'url': url,
                'status_code': None,
                'response_time': None,
                'is_available': False,
                'content_length': 0
            }
    
    def monitor_all_websites(self):
        """并發(fā)監(jiān)控所有網(wǎng)站"""
        logger.info("開(kāi)始監(jiān)控所有網(wǎng)站...")
        
        with ThreadPoolExecutor(max_workers=5) as executor:
            # 提交所有監(jiān)控任務(wù)
            future_to_url = {
                executor.submit(self.check_website, url): url 
                for url in self.urls
            }
            
            # 收集結(jié)果
            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"處理網(wǎng)站 {url} 的結(jié)果時(shí)出錯(cuò): {str(e)}")
                    
        return results
    
    def save_to_csv(self, data):
        """保存數(shù)據(jù)到CSV文件"""
        try:
            # 確保目錄存在
            os.makedirs(os.path.dirname(self.csv_file) if os.path.dirname(self.csv_file) else '.', exist_ok=True)
            
            # 寫(xiě)入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"數(shù)據(jù)已保存到 {self.csv_file}")
        except Exception as e:
            logger.error(f"保存數(shù)據(jù)到CSV文件時(shí)出錯(cuò): {str(e)}")
    
    def generate_report(self, days=7):
        """生成監(jiān)控報(bào)告"""
        try:
            # 讀取CSV數(shù)據(jù)
            if not os.path.exists(self.csv_file):
                logger.warning("監(jiān)控?cái)?shù)據(jù)文件不存在")
                return
                
            df = pd.read_csv(self.csv_file)
            df['timestamp'] = pd.to_datetime(df['timestamp'])
            
            # 篩選最近N天的數(shù)據(jù)
            cutoff_date = datetime.now() - pd.Timedelta(days=days)
            recent_data = df[df['timestamp'] >= cutoff_date]
            
            if recent_data.empty:
                logger.warning("最近沒(méi)有監(jiān)控?cái)?shù)據(jù)")
                return
                
            # 生成可用性統(tǒng)計(jì)
            availability_stats = recent_data.groupby('url').agg({
                'is_available': ['count', 'sum'],
                'response_time': ['mean', 'min', 'max']
            }).round(3)
            
            print("\n=== 網(wǎng)站可用性監(jiān)控報(bào)告 ===")
            print(f"統(tǒng)計(jì)周期: 最近{days}天")
            print(f"數(shù)據(jù)總量: {len(recent_data)} 條記錄")
            print("\n各網(wǎng)站統(tǒng)計(jì):")
            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網(wǎng)站: {url}")
                print(f"  總檢查次數(shù): {total_checks}")
                print(f"  可用次數(shù): {int(available_checks)}")
                print(f"  可用率: {availability:.2f}%")
                print(f"  平均響應(yīng)時(shí)間: {avg_response:.3f}s")
                
            # 生成響應(yīng)時(shí)間趨勢(shì)圖
            self.plot_response_trend(recent_data)
            
        except Exception as e:
            logger.error(f"生成報(bào)告時(shí)出錯(cuò): {str(e)}")
    
    def plot_response_trend(self, data):
        """繪制響應(yīng)時(shí)間趨勢(shì)圖"""
        try:
            plt.figure(figsize=(12, 8))
            
            # 為每個(gè)網(wǎng)站繪制響應(yīng)時(shí)間趨勢(shì)
            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('時(shí)間')
            plt.ylabel('響應(yīng)時(shí)間 (秒)')
            plt.title('網(wǎng)站響應(yīng)時(shí)間趨勢(shì)')
            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"響應(yīng)時(shí)間趨勢(shì)圖已保存到 {chart_file}")
            
        except Exception as e:
            logger.error(f"繪制響應(yīng)時(shí)間趨勢(shì)圖時(shí)出錯(cuò): {str(e)}")
    
    def send_alert(self, message):
        """發(fā)送告警郵件"""
        if not self.email_config:
            return
            
        try:
            # 創(chuàng)建郵件內(nèi)容
            msg = MIMEMultipart()
            msg['From'] = self.email_config['sender']
            msg['To'] = self.email_config['recipient']
            msg['Subject'] = f"網(wǎng)站監(jiān)控告警 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
            
            body = f"""
網(wǎng)站監(jiān)控系統(tǒng)告警:

{message}

時(shí)間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
            """
            msg.attach(MIMEText(body, 'plain'))
            
            # 發(fā)送郵件
            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("告警郵件發(fā)送成功")
            
        except Exception as e:
            logger.error(f"發(fā)送告警郵件時(shí)出錯(cuò): {str(e)}")
    
    def run_continuous_monitoring(self):
        """持續(xù)監(jiān)控模式"""
        logger.info(f"啟動(dòng)持續(xù)監(jiān)控模式，監(jiān)控間隔: {self.interval}秒")
        
        try:
            while True:
                self.monitor_all_websites()
                logger.info(f"等待 {self.interval} 秒后進(jìn)行下次監(jiān)控...")
                time.sleep(self.interval)
                
        except KeyboardInterrupt:
            logger.info("監(jiān)控已停止")

def main():
    parser = argparse.ArgumentParser(description='網(wǎng)站可用性監(jiān)控工具')
    parser.add_argument('urls', nargs='+', help='要監(jiān)控的網(wǎng)站URL列表')
    parser.add_argument('-i', '--interval', type=int, default=60, help='監(jiān)控間隔（秒）')
    parser.add_argument('-t', '--timeout', type=int, default=10, help='請(qǐng)求超時(shí)時(shí)間（秒）')
    parser.add_argument('-r', '--threshold', type=float, default=5.0, help='響應(yīng)時(shí)間閾值（秒）')
    parser.add_argument('-c', '--csv', default='monitor_data.csv', help='CSV數(shù)據(jù)文件路徑')
    parser.add_argument('-m', '--mode', choices=['once', 'continuous'], default='once', help='監(jiān)控模式')
    parser.add_argument('--report', action='store_true', help='生成監(jiān)控報(bào)告')
    parser.add_argument('--days', type=int, default=7, help='報(bào)告統(tǒng)計(jì)天數(shù)')
    
    args = parser.parse_args()
    
    # 配置監(jiān)控參數(shù)
    config = {
        'urls': args.urls,
        'interval': args.interval,
        'timeout': args.timeout,
        'response_threshold': args.threshold,
        'csv_file': args.csv
    }
    
    # 創(chuàng)建監(jiān)控實(shí)例
    monitor = WebsiteMonitor(config)
    
    if args.report:
        # 生成報(bào)告模式
        monitor.generate_report(args.days)
    elif args.mode == 'continuous':
        # 持續(xù)監(jiān)控模式
        monitor.run_continuous_monitoring()
    else:
        # 單次監(jiān)控模式
        monitor.monitor_all_websites()

if __name__ == '__main__':
    main()

使用說(shuō)明

1. 基本使用

# 監(jiān)控單個(gè)網(wǎng)站
python website_monitor.py https://www.google.com

# 監(jiān)控多個(gè)網(wǎng)站
python website_monitor.py https://www.google.com https://www.github.com https://www.stackoverflow.com

# 設(shè)置監(jiān)控間隔為30秒
python website_monitor.py https://www.google.com -i 30

# 設(shè)置超時(shí)時(shí)間為5秒
python website_monitor.py https://www.google.com -t 5

# 設(shè)置響應(yīng)時(shí)間閾值為2秒
python website_monitor.py https://www.google.com -r 2.0

2. 持續(xù)監(jiān)控模式

# 啟動(dòng)持續(xù)監(jiān)控
python website_monitor.py https://www.google.com https://www.github.com -m continuous -i 60

3. 生成報(bào)告

# 生成最近7天的監(jiān)控報(bào)告
python website_monitor.py --report

# 生成最近30天的監(jiān)控報(bào)告
python website_monitor.py --report --days 30

4. 配置郵件告警

在腳本中配置郵箱參數(shù)：

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

高級(jí)特性

1. 數(shù)據(jù)分析

腳本會(huì)自動(dòng)將監(jiān)控?cái)?shù)據(jù)保存到CSV文件中，便于后續(xù)分析：

import pandas as pd

# 讀取監(jiān)控?cái)?shù)據(jù)
df = pd.read_csv('monitor_data.csv')
df['timestamp'] = pd.to_datetime(df['timestamp'])

# 分析網(wǎng)站可用性
availability = df.groupby('url')['is_available'].mean() * 100
print("網(wǎng)站可用率:")
print(availability)

2. 自定義告警

可以通過(guò)修改腳本中的告警條件來(lái)自定義告警規(guī)則：

# 添加自定義告警條件
if response_time > 10.0:
    send_alert("網(wǎng)站響應(yīng)極慢")
elif status_code == 500:
    send_alert("服務(wù)器內(nèi)部錯(cuò)誤")

3. 集成到監(jiān)控系統(tǒng)

可以將腳本集成到現(xiàn)有的監(jiān)控系統(tǒng)中，定期執(zhí)行并收集數(shù)據(jù)：

# 添加到crontab中每分鐘執(zhí)行一次
* * * * * /usr/bin/python3 /path/to/website_monitor.py https://yourwebsite.com >> /var/log/website_monitor.log 2>&1

性能優(yōu)化

1. 并發(fā)處理

腳本使用線程池并發(fā)處理多個(gè)網(wǎng)站的監(jiān)控，提高監(jiān)控效率。

2. 內(nèi)存管理

合理控制數(shù)據(jù)緩存大小，避免內(nèi)存泄漏。

3. 錯(cuò)誤恢復(fù)

完善的異常處理機(jī)制確保腳本在遇到錯(cuò)誤時(shí)能夠繼續(xù)運(yùn)行。

安全考慮

1. 日志安全

敏感信息不會(huì)記錄到日志文件中。

2. 網(wǎng)絡(luò)安全

使用HTTPS協(xié)議進(jìn)行網(wǎng)絡(luò)通信。

3. 權(quán)限控制

腳本運(yùn)行所需的最小權(quán)限原則。

擴(kuò)展建議

數(shù)據(jù)庫(kù)存儲(chǔ)：將監(jiān)控?cái)?shù)據(jù)存儲(chǔ)到數(shù)據(jù)庫(kù)中，便于大規(guī)模數(shù)據(jù)分析
Web界面：開(kāi)發(fā)Web界面展示監(jiān)控?cái)?shù)據(jù)和圖表
移動(dòng)端告警：集成短信、微信等移動(dòng)端告警方式
機(jī)器學(xué)習(xí)預(yù)測(cè)：使用機(jī)器學(xué)習(xí)算法預(yù)測(cè) 網(wǎng)站故障
分布式部署：在多個(gè)地理位置部署監(jiān)控節(jié)點(diǎn)，獲得更全面的監(jiān)控視角

這個(gè)網(wǎng)站可用性監(jiān)控腳本是一個(gè)功能完整、易于使用且高度可定制的監(jiān)控工具，能夠滿足從小型個(gè)人網(wǎng)站到大型企業(yè)系統(tǒng)的各種監(jiān)控需求。

以上就是使用Python實(shí)現(xiàn)網(wǎng)站可用性監(jiān)控腳本的詳細(xì)內(nèi)容，更多關(guān)于Python網(wǎng)站監(jiān)控的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: