diff --git a/q5-mithal-monitor/dashboard (1).html b/q5-mithal-monitor/dashboard (1).html new file mode 100644 index 0000000..6d2b2f2 --- /dev/null +++ b/q5-mithal-monitor/dashboard (1).html @@ -0,0 +1,150 @@ + + + + + +مراقبة mithal.space + + + +

🔍 مراقبة mithal.space

+
Uptime · Latency · SSL · DNS · Search — يُحدَّث كل 60 ثانية من latest.json
+ +
+
+
Uptime (آخر 24 ساعة)
+
+ + + + +
--%
+
+
+
+
آخر زمن استجابة (Latency)
+
--
+
+
+
شهادة SSL — الأيام المتبقية
+
--
+
+
+
زمن تحليل DNS
+
--
+
+
+
زمن استجابة البحث
+
--
+
+
+ +
+
زمن الاستجابة — آخر ساعة (ms)
+ +
+ + + + +
الوقتالحالةزمن الاستجابةDNSبحث
+ + + + diff --git a/q5-mithal-monitor/monitor.py b/q5-mithal-monitor/monitor.py new file mode 100644 index 0000000..967de9b --- /dev/null +++ b/q5-mithal-monitor/monitor.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python3 +""" +monitor.py — Monitoring script for mithal.space (Q5) + +Collects every minute: + - Latency: time-to-first-byte for the homepage + - Uptime: HTTP status code check + - SSL: certificate validity + days remaining until expiry + - DNS: DNS resolution time + - Search response: time to get a response from a search query + +Stores results as newline-delimited JSON (JSONL) for easy append + parsing, +and a rolling `latest.json` snapshot consumed by dashboard.html. + +Usage: + python3 monitor.py # runs forever, checks every 60s + python3 monitor.py --once # single check, useful for cron/testing +""" +import argparse +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=غيمة" +CHECK_INTERVAL_SECONDS = 60 +DATA_DIR = Path(__file__).parent +JSONL_LOG = DATA_DIR / "metrics.jsonl" +LATEST_SNAPSHOT = DATA_DIR / "latest.json" +MAX_HISTORY_IN_SNAPSHOT = 1440 # 24h at 1-minute resolution + + +def check_dns(hostname: str) -> float: + """Returns DNS resolution time in ms.""" + start = time.perf_counter() + socket.gethostbyname(hostname) + return round((time.perf_counter() - start) * 1000, 2) + + +def check_http(url: str, timeout: int = 10): + """Returns (status_code, latency_ms) or (None, None) on failure.""" + try: + start = time.perf_counter() + resp = requests.get(url, timeout=timeout, headers={"User-Agent": "mithal-monitor/1.0"}) + latency_ms = round((time.perf_counter() - start) * 1000, 2) + return resp.status_code, latency_ms + except requests.RequestException as exc: + return None, None + + +def check_ssl(hostname: str, port: int = 443, timeout: int = 10): + """Returns dict with ssl validity + days_remaining, or None on failure.""" + try: + ctx = ssl.create_default_context() + with socket.create_connection((hostname, port), timeout=timeout) as sock: + with ctx.wrap_socket(sock, server_hostname=hostname) as ssock: + cert = ssock.getpeercert() + not_after = datetime.strptime(cert["notAfter"], "%b %d %H:%M:%S %Y %Z").replace(tzinfo=timezone.utc) + days_remaining = (not_after - datetime.now(timezone.utc)).days + return { + "valid": True, + "expires_at": not_after.isoformat(), + "days_remaining": days_remaining, + } + except Exception as exc: + return {"valid": False, "error": str(exc)} + + +def check_search(url: str, timeout: int = 15): + """Sends a search query and measures response time.""" + status, latency_ms = check_http(url, timeout=timeout) + return {"status_code": status, "latency_ms": latency_ms, "ok": status == 200} + + +def run_check() -> dict: + hostname = urlparse(TARGET_URL).hostname + timestamp = datetime.now(timezone.utc).isoformat() + + dns_ms = None + try: + dns_ms = check_dns(hostname) + except Exception: + pass + + status_code, latency_ms = check_http(TARGET_URL) + ssl_info = check_ssl(hostname) + search_info = check_search(SEARCH_URL) + + return { + "timestamp": timestamp, + "target": TARGET_URL, + "uptime": { + "status_code": status_code, + "up": status_code == 200, + }, + "latency_ms": latency_ms, + "dns_ms": dns_ms, + "ssl": ssl_info, + "search": search_info, + } + + +def append_jsonl(record: dict): + with open(JSONL_LOG, "a", encoding="utf-8") as f: + f.write(json.dumps(record, ensure_ascii=False) + "\n") + + +def update_snapshot(record: dict): + history = [] + if LATEST_SNAPSHOT.exists(): + try: + history = json.loads(LATEST_SNAPSHOT.read_text(encoding="utf-8")).get("history", []) + except json.JSONDecodeError: + history = [] + history.append(record) + history = history[-MAX_HISTORY_IN_SNAPSHOT:] + + up_count = sum(1 for h in history if h["uptime"]["up"]) + uptime_pct = round((up_count / len(history)) * 100, 2) if history else 0.0 + + snapshot = { + "last_check": record, + "uptime_percent_24h": uptime_pct, + "history": history, + } + LATEST_SNAPSHOT.write_text(json.dumps(snapshot, ensure_ascii=False, indent=2), encoding="utf-8") + + +def main(): + parser = argparse.ArgumentParser(description="Monitor mithal.space") + parser.add_argument("--once", action="store_true", help="Run a single check and exit") + parser.add_argument("--interval", type=int, default=CHECK_INTERVAL_SECONDS) + args = parser.parse_args() + + while True: + record = run_check() + append_jsonl(record) + update_snapshot(record) + print(json.dumps(record, ensure_ascii=False)) + + if args.once: + break + time.sleep(args.interval) + + +if __name__ == "__main__": + main()