137 أسطر
4.7 KiB
Python
137 أسطر
4.7 KiB
Python
import argparse
|
|
import csv
|
|
import json
|
|
import socket
|
|
import ssl
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
import requests
|
|
|
|
TARGET_URL = "https://mithal.space"
|
|
# عدّل هذا لمسار بحث فعلي في الموقع لو مختلف
|
|
SEARCH_URL = "https://mithal.space/search?q=test"
|
|
|
|
CSV_FILE = Path("metrics.csv")
|
|
JSON_FILE = Path("latest.json")
|
|
HISTORY_JSON = Path("history.json")
|
|
CHECK_INTERVAL = 60 # ثانية
|
|
HISTORY_MAX = 1440 # الاحتفاظ بآخر 24 ساعة (فحص كل دقيقة)
|
|
|
|
|
|
def check_dns(hostname: str) -> float:
|
|
"""يقيس وقت تحليل DNS بالمللي ثانية."""
|
|
start = time.perf_counter()
|
|
try:
|
|
socket.gethostbyname(hostname)
|
|
return round((time.perf_counter() - start) * 1000, 2)
|
|
except socket.gaierror:
|
|
return -1.0
|
|
|
|
|
|
def check_http(url: str) -> dict:
|
|
"""يقيس زمن الاستجابة وكود الحالة."""
|
|
try:
|
|
start = time.perf_counter()
|
|
resp = requests.get(url, timeout=10)
|
|
elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
|
|
return {"status_code": resp.status_code, "latency_ms": elapsed_ms, "up": resp.status_code < 400}
|
|
except requests.RequestException as e:
|
|
return {"status_code": None, "latency_ms": -1, "up": False, "error": str(e)}
|
|
|
|
|
|
def check_ssl(hostname: str, port: int = 443) -> dict:
|
|
"""يفحص حالة شهادة SSL وعدد الأيام المتبقية على انتهائها."""
|
|
try:
|
|
ctx = ssl.create_default_context()
|
|
with socket.create_connection((hostname, port), timeout=10) as sock:
|
|
with ctx.wrap_socket(sock, server_hostname=hostname) as ssock:
|
|
cert = ssock.getpeercert()
|
|
expiry = datetime.strptime(cert["notAfter"], "%b %d %H:%M:%S %Y %Z")
|
|
expiry = expiry.replace(tzinfo=timezone.utc)
|
|
days_left = (expiry - datetime.now(timezone.utc)).days
|
|
return {"valid": True, "expires_on": expiry.isoformat(), "days_left": days_left}
|
|
except Exception as e:
|
|
return {"valid": False, "error": str(e)}
|
|
|
|
|
|
def check_search(url: str) -> dict:
|
|
"""يرسل طلب بحث ويقيس زمن الرد."""
|
|
try:
|
|
start = time.perf_counter()
|
|
resp = requests.get(url, timeout=10)
|
|
elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
|
|
return {"status_code": resp.status_code, "latency_ms": elapsed_ms}
|
|
except requests.RequestException as e:
|
|
return {"status_code": None, "latency_ms": -1, "error": str(e)}
|
|
|
|
|
|
def run_check() -> dict:
|
|
hostname = urlparse(TARGET_URL).hostname
|
|
timestamp = datetime.now(timezone.utc).isoformat()
|
|
|
|
http_result = check_http(TARGET_URL)
|
|
dns_ms = check_dns(hostname)
|
|
ssl_result = check_ssl(hostname)
|
|
search_result = check_search(SEARCH_URL)
|
|
|
|
return {
|
|
"timestamp": timestamp,
|
|
"up": http_result.get("up", False),
|
|
"status_code": http_result.get("status_code"),
|
|
"latency_ms": http_result.get("latency_ms"),
|
|
"dns_ms": dns_ms,
|
|
"ssl_valid": ssl_result.get("valid"),
|
|
"ssl_days_left": ssl_result.get("days_left"),
|
|
"search_latency_ms": search_result.get("latency_ms"),
|
|
}
|
|
|
|
|
|
def append_csv(row: dict):
|
|
is_new = not CSV_FILE.exists()
|
|
with open(CSV_FILE, "a", newline="", encoding="utf-8") as f:
|
|
writer = csv.DictWriter(f, fieldnames=list(row.keys()))
|
|
if is_new:
|
|
writer.writeheader()
|
|
writer.writerow(row)
|
|
|
|
|
|
def update_json(row: dict):
|
|
# أحدث قراءة (يقرأها الـ dashboard مباشرة)
|
|
JSON_FILE.write_text(json.dumps(row, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
# سجل تاريخي محدود (لآخر 24 ساعة) يستخدمه الـ dashboard للرسم البياني والجدول
|
|
history = []
|
|
if HISTORY_JSON.exists():
|
|
try:
|
|
history = json.loads(HISTORY_JSON.read_text(encoding="utf-8"))
|
|
except json.JSONDecodeError:
|
|
history = []
|
|
history.append(row)
|
|
history = history[-HISTORY_MAX:]
|
|
HISTORY_JSON.write_text(json.dumps(history, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--once", action="store_true", help="تشغيل فحص واحد فقط ثم الخروج")
|
|
args = parser.parse_args()
|
|
|
|
while True:
|
|
row = run_check()
|
|
append_csv(row)
|
|
update_json(row)
|
|
print(f"[{row['timestamp']}] up={row['up']} latency={row['latency_ms']}ms "
|
|
f"dns={row['dns_ms']}ms ssl_days_left={row['ssl_days_left']} "
|
|
f"search={row['search_latency_ms']}ms")
|
|
|
|
if args.once:
|
|
break
|
|
time.sleep(CHECK_INTERVAL)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|