Complete interview solution
هذا الالتزام موجود في:
@@ -0,0 +1,174 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Monitoring Script for mithal.space
|
||||
Collects: latency, uptime, SSL, DNS, search response
|
||||
"""
|
||||
|
||||
import requests
|
||||
import json
|
||||
import time
|
||||
import socket
|
||||
import ssl
|
||||
import datetime
|
||||
import os
|
||||
from urllib.parse import urlparse
|
||||
|
||||
# Configuration
|
||||
TARGET_URL = "https://mithal.space"
|
||||
SEARCH_QUERY = "?q=test" # adjust if search endpoint differs
|
||||
DATA_FILE = "monitoring-data.json"
|
||||
MAX_ENTRIES = 1440 # 24 hours * 60 minutes
|
||||
TIMEOUT = 10
|
||||
|
||||
def check_ssl_certificate(hostname, port=443):
|
||||
"""Check SSL certificate expiry date."""
|
||||
try:
|
||||
context = ssl.create_default_context()
|
||||
with socket.create_connection((hostname, port), timeout=TIMEOUT) as sock:
|
||||
with context.wrap_socket(sock, server_hostname=hostname) as ssock:
|
||||
cert = ssock.getpeercert()
|
||||
expiry_date = datetime.datetime.strptime(cert['notAfter'], '%b %d %H:%M:%S %Y %Z')
|
||||
days_remaining = (expiry_date - datetime.datetime.utcnow()).days
|
||||
return {
|
||||
"expiry_date": expiry_date.isoformat(),
|
||||
"days_remaining": days_remaining,
|
||||
"valid": days_remaining > 0
|
||||
}
|
||||
except Exception as e:
|
||||
return {
|
||||
"expiry_date": None,
|
||||
"days_remaining": None,
|
||||
"valid": False,
|
||||
"error": str(e)
|
||||
}
|
||||
|
||||
def resolve_dns(hostname):
|
||||
"""Measure DNS resolution time."""
|
||||
try:
|
||||
start = time.time()
|
||||
socket.gethostbyname(hostname)
|
||||
dns_time = round((time.time() - start) * 1000, 2)
|
||||
return dns_time
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def check_latency_and_uptime(url):
|
||||
"""Check HTTP response time and status code."""
|
||||
try:
|
||||
start = time.time()
|
||||
response = requests.get(url, timeout=TIMEOUT)
|
||||
latency = round((time.time() - start) * 1000, 2)
|
||||
status_code = response.status_code
|
||||
is_up = 200 <= status_code < 400
|
||||
return {
|
||||
"latency": latency,
|
||||
"status_code": status_code,
|
||||
"is_up": is_up,
|
||||
"error": None
|
||||
}
|
||||
except requests.RequestException as e:
|
||||
return {
|
||||
"latency": None,
|
||||
"status_code": None,
|
||||
"is_up": False,
|
||||
"error": str(e)
|
||||
}
|
||||
|
||||
def check_search_response(url, query="?q=test"):
|
||||
"""Measure search endpoint response time."""
|
||||
try:
|
||||
search_url = f"{url}{query}"
|
||||
start = time.time()
|
||||
response = requests.get(search_url, timeout=TIMEOUT)
|
||||
search_latency = round((time.time() - start) * 1000, 2)
|
||||
return {
|
||||
"search_latency": search_latency,
|
||||
"search_status": response.status_code,
|
||||
"search_error": None
|
||||
}
|
||||
except requests.RequestException as e:
|
||||
return {
|
||||
"search_latency": None,
|
||||
"search_status": None,
|
||||
"search_error": str(e)
|
||||
}
|
||||
|
||||
def collect_metrics():
|
||||
"""Collect all metrics for mithal.space."""
|
||||
parsed = urlparse(TARGET_URL)
|
||||
hostname = parsed.hostname or "mithal.space"
|
||||
|
||||
dns_time = resolve_dns(hostname)
|
||||
ssl_info = check_ssl_certificate(hostname)
|
||||
health = check_latency_and_uptime(TARGET_URL)
|
||||
search = check_search_response(TARGET_URL, SEARCH_QUERY)
|
||||
|
||||
metrics = {
|
||||
"timestamp": datetime.datetime.utcnow().isoformat() + "Z",
|
||||
"url": TARGET_URL,
|
||||
"dns_ms": dns_time,
|
||||
"ssl_valid": ssl_info.get("valid", False),
|
||||
"ssl_days_remaining": ssl_info.get("days_remaining"),
|
||||
"ssl_expiry_date": ssl_info.get("expiry_date"),
|
||||
"latency_ms": health.get("latency"),
|
||||
"status_code": health.get("status_code"),
|
||||
"is_up": health.get("is_up", False),
|
||||
"error": health.get("error"),
|
||||
"search_latency_ms": search.get("search_latency"),
|
||||
"search_status_code": search.get("search_status"),
|
||||
"search_error": search.get("search_error")
|
||||
}
|
||||
return metrics
|
||||
|
||||
def save_metrics(metrics, filename=DATA_FILE, max_entries=MAX_ENTRIES):
|
||||
"""Save metrics to JSON file with size limit."""
|
||||
try:
|
||||
if os.path.exists(filename):
|
||||
with open(filename, 'r') as f:
|
||||
data = json.load(f)
|
||||
else:
|
||||
data = []
|
||||
data.append(metrics)
|
||||
if len(data) > max_entries:
|
||||
data = data[-max_entries:]
|
||||
with open(filename, 'w') as f:
|
||||
json.dump(data, f, indent=2)
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Error saving data: {e}")
|
||||
return False
|
||||
|
||||
def run_monitor():
|
||||
"""Main monitoring loop."""
|
||||
print(f"🚀 Starting monitoring for {TARGET_URL}")
|
||||
print(f"📊 Data will be saved to {DATA_FILE}")
|
||||
print("=" * 50)
|
||||
|
||||
while True:
|
||||
try:
|
||||
print(f"[{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Collecting metrics...")
|
||||
metrics = collect_metrics()
|
||||
|
||||
status = "✅ UP" if metrics['is_up'] else "❌ DOWN"
|
||||
print(f" Status: {status}")
|
||||
print(f" Latency: {metrics['latency_ms']} ms")
|
||||
print(f" DNS: {metrics['dns_ms']} ms")
|
||||
print(f" SSL: {metrics['ssl_days_remaining']} days remaining")
|
||||
print(f" Search: {metrics['search_latency_ms']} ms")
|
||||
|
||||
if save_metrics(metrics):
|
||||
print(" ✅ Data saved")
|
||||
else:
|
||||
print(" ❌ Failed to save data")
|
||||
print("-" * 50)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n🛑 Monitoring stopped by user")
|
||||
break
|
||||
except Exception as e:
|
||||
print(f"❌ Error in monitoring loop: {e}")
|
||||
|
||||
time.sleep(60)
|
||||
|
||||
if __name__ == "__main__":
|
||||
run_monitor()
|
||||
المرجع في مشكلة جديدة
حظر مستخدم