Python网站可用性监控脚本

#!/usr/bin/env python3 # -*- coding: utf-8 -*- ''' 网站可用性监控脚本 用途：监控网站/API可用性，检测状态码、响应时间、SSL证书 依赖：pip install requests 使用方法：python3 website_monitor.py ''' import requests import json import time import ssl import socket from datetime import datetime, timedelta from concurrent.futures import ThreadPoolExecutor, as_completed # ========== 配置 ========== # 监控列表 SITES = [ {"name": "百度", "url": "https://www.baidu.com", "expect_status": 200, "timeout": 10}, {"name": "阿里云", "url": "https://www.aliyun.com", "expect_status": 200, "timeout": 10}, # {"name": "API测试", "url": "https://api.example.com/health", "expect_status": 200, "timeout": 5}, ] CHECK_INTERVAL = 60 # 检测间隔(秒)，0表示只检测一次 MAX_RETRIES = 2 # 失败重试次数 RETRY_DELAY = 3 # 重试间隔(秒) WARN_RESPONSE_TIME = 3000 # 响应时间告警阈值(毫秒) WARN_SSL_DAYS = 30 # SSL证书剩余天数告警阈值 WECOM_WEBHOOK = "" # 企业微信Webhook ALERT_EMAIL = "" # 告警邮箱 LOG_FILE = "/tmp/website_monitor.log" # ========================== def log_msg(msg): line = f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}" print(line) with open(LOG_FILE, "a", encoding="utf-8") as f: f.write(line + "\n") def send_alert(title, content): '''发送告警''' log_msg(f"【告警】{title}: {content}") # 企业微信告警 if WECOM_WEBHOOK: try: requests.post(WECOM_WEBHOOK, json={ "msgtype": "text", "text": {"content": f"【网站监控告警】{title}\n{content}"} }, timeout=5) except Exception as e: log_msg(f"企业微信告警发送失败: {e}") def check_ssl_cert(hostname, port=443): '''检查SSL证书有效期''' try: context = ssl.create_default_context() with socket.create_connection((hostname, port), timeout=5) as sock: with context.wrap_socket(sock, server_hostname=hostname) as ssock: cert = ssock.getpeercert() expiry_date = datetime.strptime(cert["notAfter"], "%b %d %H:%M:%S %Y %Z") days_left = (expiry_date - datetime.now()).days return {"valid": True, "expiry": expiry_date.strftime("%Y-%m-%d"), "days_left": days_left} except Exception as e: return {"valid": False, "error": str(e)} def check_site(site): '''检测单个网站''' name = site["name"] url = site["url"] expect_status = site.get("expect_status", 200) timeout = site.get("timeout", 10) result = { "name": name, "url": url, "status": "unknown", "http_status": None, "response_time": None, "error": None, "ssl": None } # 检查SSL证书（HTTPS） if url.startswith("https://"): hostname = url.split("://")[1].split("/")[0].split(":")[0] ssl_result = check_ssl_cert(hostname) result["ssl"] = ssl_result if ssl_result["valid"] and ssl_result["days_left"] <= WARN_SSL_DAYS: send_alert(f"{name} SSL证书即将过期", f"证书剩余 {ssl_result['days_left']} 天，到期日: {ssl_result['expiry']}") # 重试机制 for attempt in range(MAX_RETRIES + 1): try: start = time.time() response = requests.get(url, timeout=timeout, allow_redirects=True, headers={"User-Agent": "WebsiteMonitor/1.0"}) elapsed = (time.time() - start) * 1000 result["http_status"] = response.status_code result["response_time"] = round(elapsed, 0) if response.status_code == expect_status: result["status"] = "up" if elapsed > WARN_RESPONSE_TIME: send_alert(f"{name} 响应过慢", f"响应时间: {elapsed:.0f}ms (阈值: {WARN_RESPONSE_TIME}ms)") else: result["status"] = "down" result["error"] = f"状态码异常: 期望{expect_status}, 实际{response.status_code}" break except requests.exceptions.Timeout: result["error"] = f"请求超时({timeout}s)" if attempt < MAX_RETRIES: time.sleep(RETRY_DELAY) continue result["status"] = "down" except requests.exceptions.ConnectionError as e: result["error"] = f"连接失败: {str(e)[:100]}" if attempt < MAX_RETRIES: time.sleep(RETRY_DELAY) continue result["status"] = "down" except Exception as e: result["error"] = f"异常: {str(e)[:100]}" result["status"] = "down" break return result def print_results(results): '''打印检测结果''' print(f"\n{'='*70}") print(f" 网站可用性检测报告 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"{'='*70}") print(f"{'名称':<12} {'状态':<6} {'HTTP':<6} {'响应时间':<10} {'SSL剩余':<10} {'说明'}") print(f"{'-'*70}") up_count = 0 down_count = 0 for r in results: status_icon = "🟢" if r["status"] == "up" else "🔴" http = str(r["http_status"]) if r["http_status"] else "-" rt = f"{r['response_time']:.0f}ms" if r["response_time"] else "-" ssl_days = f"{r['ssl']['days_left']}天" if r["ssl"] and r["ssl"]["valid"] else "-" note = r["error"] or "" print(f"{r['name']:<12} {status_icon} {http:<6} {rt:<10} {ssl_days:<10} {note}") if r["status"] == "up": up_count += 1 else: down_count += 1 print(f"{'-'*70}") print(f"总计: {len(results)} 个 | 正常: {up_count} | 异常: {down_count}") print(f"{'='*70}") def main(): log_msg("网站可用性监控启动") log_msg(f"监控网站数: {len(SITES)}") log_msg(f"检测间隔: {CHECK_INTERVAL}秒" if CHECK_INTERVAL > 0 else "单次检测模式") def run_check(): results = [] with ThreadPoolExecutor(max_workers=10) as executor: futures = {executor.submit(check_site, site): site for site in SITES} for future in as_completed(futures): try: result = future.result() results.append(result) if result["status"] == "down": send_alert(f"{result['name']} 不可用", f"URL: {result['url']}\n错误: {result['error']}") except Exception as e: log_msg(f"检测异常: {e}") print_results(results) return results if CHECK_INTERVAL <= 0: run_check() else: while True: run_check() log_msg(f"等待 {CHECK_INTERVAL} 秒后进行下一次检测...") time.sleep(CHECK_INTERVAL) if __name__ == "__main__": main()

[{"name": "SITES", "label": "\u76d1\u63a7\u7f51\u7ad9\u5217\u8868(JSON)", "type": "textarea", "default": "[{\"name\":\"\u767e\u5ea6\",\"url\":\"https://www.baidu.com\",\"expect_status\":200,\"timeout\":10}]"}, {"name": "CHECK_INTERVAL", "label": "\u68c0\u6d4b\u95f4\u9694(\u79d2\uff0c0=\u5355\u6b21)", "default": "60"}, {"name": "WARN_RESPONSE_TIME", "label": "\u54cd\u5e94\u65f6\u95f4\u544a\u8b66\u9608\u503c(ms)", "default": "3000"}, {"name": "WECOM_WEBHOOK", "label": "\u4f01\u4e1a\u5fae\u4fe1Webhook(\u53ef\u9009)", "default": ""}]