Q5: Answer
هذا الالتزام موجود في:
97
q5-mithal-monitor/monitor.py
Normal file
97
q5-mithal-monitor/monitor.py
Normal file
@@ -0,0 +1,97 @@
|
||||
import requests
|
||||
import socket
|
||||
import ssl
|
||||
import time
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
|
||||
TARGET_HOST = "mithal.space"
|
||||
TARGET_URL = f"https://{TARGET_HOST}"
|
||||
DATA_FILE = "metrics.json"
|
||||
MAX_RECORDS = 1440
|
||||
|
||||
def get_dns_latency():
|
||||
"""حساب وقت تحليل الـ DNS بالملي ثانية"""
|
||||
start_time = time.time()
|
||||
try:
|
||||
socket.gethostbyname(TARGET_HOST)
|
||||
return round((time.time() - start_time) * 1000, 2)
|
||||
except Exception:
|
||||
return 0
|
||||
|
||||
def get_ssl_expiry_days():
|
||||
"""التحقق من شهادة SSL وحساب الأيام المتبقية"""
|
||||
try:
|
||||
context = ssl.create_default_context()
|
||||
with socket.create_connection((TARGET_HOST, 443), timeout=5) as sock:
|
||||
with context.wrap_socket(sock, server_hostname=TARGET_HOST) as ssock:
|
||||
ssl_info = ssock.getpeercert()
|
||||
expire_date_str = ssl_info['notAfter']
|
||||
expire_date = datetime.strptime(expire_date_str, "%b %d %H:%M:%S %Y %Z")
|
||||
remaining = expire_date - datetime.utcnow()
|
||||
return remaining.days
|
||||
except Exception:
|
||||
return 0
|
||||
|
||||
def check_health():
|
||||
"""جمع كل المقاييس المطلوبة"""
|
||||
metrics = {
|
||||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"uptime_status": "Down",
|
||||
"latency_ms": 0,
|
||||
"ssl_days_left": get_ssl_expiry_days(),
|
||||
"dns_latency_ms": get_dns_latency(),
|
||||
"search_latency_ms": 0
|
||||
}
|
||||
|
||||
try:
|
||||
response = requests.get(TARGET_URL, timeout=10)
|
||||
metrics["latency_ms"] = round(response.elapsed.total_seconds() * 1000, 2)
|
||||
if response.status_code == 200:
|
||||
metrics["uptime_status"] = "Up"
|
||||
else:
|
||||
metrics["uptime_status"] = f"Error {response.status_code}"
|
||||
except Exception:
|
||||
metrics["uptime_status"] = "Down"
|
||||
|
||||
try:
|
||||
search_res = requests.get(TARGET_URL, params={"q": "test"}, timeout=10)
|
||||
metrics["search_latency_ms"] = round(search_res.elapsed.total_seconds() * 1000, 2)
|
||||
except Exception:
|
||||
metrics["search_latency_ms"] = 0
|
||||
|
||||
return metrics
|
||||
|
||||
def save_metrics(new_metric):
|
||||
"""حفظ البيانات في ملف JSON"""
|
||||
data = []
|
||||
if os.path.exists(DATA_FILE):
|
||||
try:
|
||||
with open(DATA_FILE, "r") as f:
|
||||
data = json.load(f)
|
||||
except json.JSONDecodeError:
|
||||
data = []
|
||||
|
||||
data.append(new_metric)
|
||||
|
||||
if len(data) > MAX_RECORDS:
|
||||
data = data[-MAX_RECORDS:]
|
||||
|
||||
with open(DATA_FILE, "w") as f:
|
||||
json.dump(data, f, indent=4)
|
||||
|
||||
if __name__ == "__main__":
|
||||
print(f"Starting monitoring for {TARGET_HOST}... (Press Ctrl+C to stop)")
|
||||
while True:
|
||||
try:
|
||||
metrics = check_health()
|
||||
save_metrics(metrics)
|
||||
print(f"[{metrics['timestamp']}] Logged: {metrics['uptime_status']} | Latency: {metrics['latency_ms']}ms | SSL: {metrics['ssl_days_left']} days left")
|
||||
time.sleep(60)
|
||||
except KeyboardInterrupt:
|
||||
print("\nMonitoring stopped by user.")
|
||||
break
|
||||
except Exception as e:
|
||||
print(f"Unexpected error: {e}")
|
||||
time.sleep(60)
|
||||
المرجع في مشكلة جديدة
حظر مستخدم