使用Python實現(xiàn)網(wǎng)站可用性監(jiān)控腳本
功能介紹
這是一個專業(yè)的網(wǎng)站可用性監(jiān)控腳本,能夠持續(xù)監(jiān)控一個或多個網(wǎng)站的在線狀態(tài)和響應時間。該腳本具備以下核心功能:
- 多網(wǎng)站并發(fā)監(jiān)控:同時監(jiān)控多個網(wǎng)站,提高監(jiān)控效率
- 響應時間測量:精確測量網(wǎng)站的響應時間,包括DNS解析、連接建立、內(nèi)容傳輸?shù)雀麟A段耗時
- 狀態(tài)碼檢測:檢查HTTP狀態(tài)碼,識別網(wǎng)站是否正常響應
- 智能告警機制:當網(wǎng)站出現(xiàn)故障或響應時間超過閾值時自動發(fā)送告警
- 歷史數(shù)據(jù)記錄:將監(jiān)控數(shù)據(jù)保存到CSV文件,便于后續(xù)分析
- 可視化報表生成:生成響應時間趨勢圖和可用性統(tǒng)計圖
- 靈活配置選項:支持自定義監(jiān)控間隔、超時時間、告警閾值等參數(shù)
場景應用
1. 企業(yè)IT運維
- 監(jiān)控公司官網(wǎng)、內(nèi)部系統(tǒng)、客戶門戶等關鍵業(yè)務系統(tǒng)的可用性
- 及時發(fā)現(xiàn)系統(tǒng)故障,縮短故障響應時間
- 為SLA考核提供客觀數(shù)據(jù)支撐
2. 電商平臺運營
- 實時監(jiān)控購物網(wǎng)站、支付接口、物流查詢等核心服務
- 預防因網(wǎng)站不可用導致的訂單流失
- 分析網(wǎng)站性能瓶頸,優(yōu)化用戶體驗
3. 內(nèi)容網(wǎng)站管理
- 監(jiān)控新聞網(wǎng)站、博客平臺、視頻網(wǎng)站等內(nèi)容服務
- 確保高并發(fā)訪問時的穩(wěn)定性和響應速度
- 為CDN優(yōu)化和服務器擴容提供數(shù)據(jù)依據(jù)
4. 個人站長維護
- 監(jiān)控個人博客、作品集網(wǎng)站等個人項目的運行狀態(tài)
- 及時發(fā)現(xiàn)并處理網(wǎng)站異常,保障訪問體驗
- 積累網(wǎng)站運行數(shù)據(jù),為優(yōu)化提供參考
報錯處理
1. 網(wǎng)絡連接異常
try:
response = requests.get(url, timeout=timeout)
except requests.exceptions.ConnectionError:
log_error(f"無法連接到 {url},網(wǎng)絡連接異常")
send_alert(f"網(wǎng)站 {url} 連接失敗")
except requests.exceptions.Timeout:
log_error(f"請求 {url} 超時")
send_alert(f"網(wǎng)站 {url} 響應超時")
2. DNS解析失敗
except requests.exceptions.InvalidURL:
log_error(f"無效的URL地址: {url}")
send_alert(f"網(wǎng)站 {url} URL格式錯誤")
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"服務器錯誤 {response.status_code} for {url}")
send_alert(f"網(wǎng)站 {url} 服務器內(nèi)部錯誤")
elif response.status_code >= 400:
log_warning(f"客戶端錯誤 {response.status_code} for {url}")
4. SSL證書驗證失敗
except requests.exceptions.SSLError as e:
log_error(f"SSL證書驗證失敗: {url} - {str(e)}")
send_alert(f"網(wǎng)站 {url} SSL證書問題")
5. 文件操作異常
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)}")
代碼實現(xiàn)
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
網(wǎng)站可用性監(jiān)控腳本
功能:監(jiān)控網(wǎng)站可用性、響應時間、生成報告和告警
作者: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) # 請求超時時間(秒)
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 = [] # 數(shù)據(jù)緩存
def check_website(self, url):
"""檢查單個網(wǎng)站的可用性"""
start_time = time.time()
timestamp = datetime.now()
try:
# 發(fā)送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
# 記錄監(jiān)控數(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)
}
# 檢查響應時間和狀態(tài)碼
if response_time > self.response_threshold:
logger.warning(f"網(wǎng)站 {url} 響應時間過長: {response_time:.3f}s")
self.send_alert(f"網(wǎng)站 {url} 響應時間過長 ({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}, 響應時間: {response_time:.3f}s")
return data
except requests.exceptions.ConnectionError:
logger.error(f"無法連接到網(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} 請求超時")
self.send_alert(f"網(wǎng)站 {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"檢查網(wǎng)站 {url} 時發(fā)生未知錯誤: {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("開始監(jiān)控所有網(wǎng)站...")
with ThreadPoolExecutor(max_workers=5) as executor:
# 提交所有監(jiān)控任務
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"處理網(wǎng)站 {url} 的結果時出錯: {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)
# 寫入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文件時出錯: {str(e)}")
def generate_report(self, days=7):
"""生成監(jiān)控報告"""
try:
# 讀取CSV數(shù)據(jù)
if not os.path.exists(self.csv_file):
logger.warning("監(jiān)控數(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("最近沒有監(jiān)控數(shù)據(jù)")
return
# 生成可用性統(tǒng)計
availability_stats = recent_data.groupby('url').agg({
'is_available': ['count', 'sum'],
'response_time': ['mean', 'min', 'max']
}).round(3)
print("\n=== 網(wǎng)站可用性監(jiān)控報告 ===")
print(f"統(tǒng)計周期: 最近{days}天")
print(f"數(shù)據(jù)總量: {len(recent_data)} 條記錄")
print("\n各網(wǎng)站統(tǒng)計:")
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" 平均響應時間: {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))
# 為每個網(wǎng)站繪制響應時間趨勢
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('網(wǎng)站響應時間趨勢')
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):
"""發(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}
時間: {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ā)送告警郵件時出錯: {str(e)}")
def run_continuous_monitoring(self):
"""持續(xù)監(jiān)控模式"""
logger.info(f"啟動持續(xù)監(jiān)控模式,監(jiān)控間隔: {self.interval}秒")
try:
while True:
self.monitor_all_websites()
logger.info(f"等待 {self.interval} 秒后進行下次監(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='請求超時時間(秒)')
parser.add_argument('-r', '--threshold', type=float, default=5.0, help='響應時間閾值(秒)')
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)控報告')
parser.add_argument('--days', type=int, default=7, help='報告統(tǒng)計天數(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)控實例
monitor = WebsiteMonitor(config)
if args.report:
# 生成報告模式
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()
使用說明
1. 基本使用
# 監(jiān)控單個網(wǎng)站 python website_monitor.py https://www.google.com # 監(jiān)控多個網(wǎng)站 python website_monitor.py https://www.google.com https://www.github.com https://www.stackoverflow.com # 設置監(jiān)控間隔為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
2. 持續(xù)監(jiān)控模式
# 啟動持續(xù)監(jiān)控 python website_monitor.py https://www.google.com https://www.github.com -m continuous -i 60
3. 生成報告
# 生成最近7天的監(jiān)控報告 python website_monitor.py --report # 生成最近30天的監(jiān)控報告 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'
}
高級特性
1. 數(shù)據(jù)分析
腳本會自動將監(jiān)控數(shù)據(jù)保存到CSV文件中,便于后續(xù)分析:
import pandas as pd
# 讀取監(jiān)控數(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ī)則:
# 添加自定義告警條件
if response_time > 10.0:
send_alert("網(wǎng)站響應極慢")
elif status_code == 500:
send_alert("服務器內(nèi)部錯誤")
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ā)處理多個網(wǎng)站的監(jiān)控,提高監(jiān)控效率。
2. 內(nèi)存管理
合理控制數(shù)據(jù)緩存大小,避免內(nèi)存泄漏。
3. 錯誤恢復
完善的異常處理機制確保腳本在遇到錯誤時能夠繼續(xù)運行。
安全考慮
1. 日志安全
敏感信息不會記錄到日志文件中。
2. 網(wǎng)絡安全
使用HTTPS協(xié)議進行網(wǎng)絡通信。
3. 權限控制
腳本運行所需的最小權限原則。
擴展建議
- 數(shù)據(jù)庫存儲:將監(jiān)控數(shù)據(jù)存儲到數(shù)據(jù)庫中,便于大規(guī)模數(shù)據(jù)分析
- Web界面:開發(fā)Web界面展示監(jiān)控數(shù)據(jù)和圖表
- 移動端告警:集成短信、微信等移動端告警方式
- 機器學習預測:使用機器學習算法預測 網(wǎng)站故障
- 分布式部署:在多個地理位置部署監(jiān)控節(jié)點,獲得更全面的監(jiān)控視角
這個網(wǎng)站可用性監(jiān)控腳本是一個功能完整、易于使用且高度可定制的監(jiān)控工具,能夠滿足從小型個人網(wǎng)站到大型企業(yè)系統(tǒng)的各種監(jiān)控需求。
以上就是使用Python實現(xiàn)網(wǎng)站可用性監(jiān)控腳本的詳細內(nèi)容,更多關于Python網(wǎng)站監(jiān)控的資料請關注腳本之家其它相關文章!
相關文章
PySide(PyQt)使用QPropertyAnimation制作動態(tài)界面的示例代碼
文章介紹了如何使用PySide或PyQt的QPropertyAnimation類來創(chuàng)建動態(tài)界面效果,感興趣的朋友一起看看吧2025-03-03
Python3實現(xiàn)的判斷環(huán)形鏈表算法示例
這篇文章主要介紹了Python3實現(xiàn)的判斷環(huán)形鏈表算法,涉及Python針對環(huán)形鏈表的遍歷、判斷相關操作技巧,需要的朋友可以參考下2019-03-03
Python 2.6.6升級到Python2.7.15的詳細步驟
這篇文章主要介紹了Python 2.6.6升級到Python2.7.15的詳細步驟,本文分步驟給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下2020-12-12

