#!/usr/bin/env python3 """ مراقب موقع mithal.space (Website Monitor) ========================================== يجمع كل دقيقة: - Latency : زمن استجابة HTTP للصفحة الرئيسية - Uptime : هل الموقع متاح (status code 2xx/3xx) - SSL : صلاحية الشهادة وعدد الأيام المتبقية على انتهائها - DNS : زمن تحليل اسم النطاق (DNS resolution time) - Search : زمن الرد على طلب بحث (search query) داخل الموقع يخزّن كل فحص كسطر JSON مستقل في data.jsonl (JSON Lines) وأيضاً كصف في data.csv، بحيث يمكن قراءة الملف تراكمياً دون الحاجة لإعادة كتابته. الاستخدام: python3 monitor.py # يعمل باستمرار، فحص كل 60 ثانية python3 monitor.py --once # فحص واحد فقط (مناسب لجدولة cron) python3 monitor.py --interval 30 # تغيير الفاصل الزمني (بالثواني) """ import argparse import concurrent.futures import csv import json import os import socket import ssl import sys import time from datetime import datetime, timezone import requests # ---------------------------------------------------------------------------- # الإعدادات (Configuration) — عدّل هذه القيم حسب موقعك # ---------------------------------------------------------------------------- SITE_URL = "https://mithal.space" SITE_HOST = "mithal.space" SITE_PORT = 443 # رابط البحث الذي سيتم اختباره — عدّله ليطابق مسار البحث الفعلي في موقعك # مثال: https://mithal.space/search?q=test SEARCH_URL = "https://mithal.space/search?q=test" REQUEST_TIMEOUT = 10 # ثانية CHECK_INTERVAL_SECONDS = 60 # كل دقيقة DATA_DIR = os.path.dirname(os.path.abspath(__file__)) JSONL_PATH = os.path.join(DATA_DIR, "data.jsonl") CSV_PATH = os.path.join(DATA_DIR, "data.csv") JS_PATH = os.path.join(DATA_DIR, "data.js") CSV_FIELDS = [ "timestamp", "dns_time_ms", "http_up", "http_status_code", "http_latency_ms", "ssl_valid", "ssl_days_remaining", "ssl_expiry_date", "search_up", "search_status_code", "search_latency_ms", "error", ] def now_iso(): return datetime.now(timezone.utc).isoformat() # ---------------------------------------------------------------------------- # 1) DNS — وقت تحليل اسم النطاق # ---------------------------------------------------------------------------- def measure_dns(hostname: str, timeout: float = REQUEST_TIMEOUT): """ ملاحظة مهمة: socket.getaddrinfo() لا يقبل معامل timeout أصلاً، وقد يتجمّد إلى ما لا نهاية إذا كان الـ DNS بطيئاً أو محجوباً (VPN/جدار حماية/شبكة شركة). لتفادي تجميد السكربت بالكامل، ننفّذ البحث في Thread منفصل ونفرض عليه مهلة زمنية يدوياً عبر future.result(timeout=...). """ executor = concurrent.futures.ThreadPoolExecutor(max_workers=1) start = time.perf_counter() try: future = executor.submit(socket.getaddrinfo, hostname, None) future.result(timeout=timeout) elapsed_ms = (time.perf_counter() - start) * 1000 return {"time_ms": round(elapsed_ms, 2), "error": None} except concurrent.futures.TimeoutError: return {"time_ms": None, "error": f"DNS timeout after {timeout}s (تحقق من الاتصال بالإنترنت أو VPN/DNS)"} except socket.gaierror as e: return {"time_ms": None, "error": str(e)} finally: # wait=False حتى لا ننتظر انتهاء الـ thread العالق في حال حدوث timeout executor.shutdown(wait=False) # ---------------------------------------------------------------------------- # 2) HTTP — Latency + Uptime (status code) # ---------------------------------------------------------------------------- def measure_http(url: str): try: start = time.perf_counter() resp = requests.get(url, timeout=REQUEST_TIMEOUT) elapsed_ms = (time.perf_counter() - start) * 1000 return { "up": resp.status_code < 400, "status_code": resp.status_code, "latency_ms": round(elapsed_ms, 2), "error": None, } except requests.exceptions.RequestException as e: return {"up": False, "status_code": None, "latency_ms": None, "error": str(e)} # ---------------------------------------------------------------------------- # 3) SSL — حالة الشهادة وتاريخ الانتهاء # ---------------------------------------------------------------------------- def _get_ca_context(): """يستخدم حزمة CA من certifi إن كانت مثبتة (أدق وأكثر استقراراً على ويندوز من الاعتماد على مخزن الشهادات الافتراضي)، وإلا يعود للسياق الافتراضي.""" try: import certifi return ssl.create_default_context(cafile=certifi.where()) except ImportError: return ssl.create_default_context() def _read_cert_expiry(der_bytes: bytes): """Read certificate expiry date using Python's built-in SSL module.""" try: pem_bytes = ssl.DER_cert_to_PEM_cert(der_bytes) # ssl._ssl._test_decode_cert() needs a file, # so we create a temporary certificate file. import tempfile with tempfile.NamedTemporaryFile( mode="w", suffix=".pem", delete=False ) as f: f.write(pem_bytes) cert_path = f.name try: cert_info = ssl._ssl._test_decode_cert(cert_path) expiry_str = cert_info.get("notAfter") if expiry_str: expiry_dt = datetime.strptime( expiry_str, "%b %d %H:%M:%S %Y %Z" ).replace(tzinfo=timezone.utc) return expiry_dt return None finally: os.unlink(cert_path) except Exception: return None def check_ssl(hostname: str, port: int = 443): """ يفصل بين أمرين مختلفين قد يُخلط بينهما: 1) trusted : هل تثق سلسلة الشهادات (chain of trust) بالشهادة عبر حزمة CA؟ قد تفشل هذه الخطوة محلياً (خصوصاً على ويندوز) دون أن يكون هناك أي خلل في شهادة الموقع نفسه. 2) expiry : تاريخ انتهاء الشهادة الفعلي — نقرأه دائماً من البايتات الخام للشهادة (getpeercert(binary_form=True)) بغض النظر عن نجاح التحقق من الثقة، لأن getpeercert() العادي يعيد قاموساً فارغاً {} إن لم يتم التحقق من الثقة. """ trusted = False trust_error = None days_remaining = None expiry_date = None # 1) محاولة اتصال بالتحقق الكامل من الثقة try: context = _get_ca_context() with socket.create_connection((hostname, port), timeout=REQUEST_TIMEOUT) as sock: with context.wrap_socket(sock, server_hostname=hostname): trusted = True except ssl.SSLCertVerificationError as e: trust_error = f"فشل التحقق من سلسلة الثقة محلياً (قد يكون خلل CA bundle وليس عيباً بالشهادة): {e}" except (socket.timeout, socket.gaierror, ConnectionRefusedError, OSError) as e: return {"valid": False, "days_remaining": None, "expiry_date": None, "error": str(e)} # 2) قراءة تاريخ الانتهاء بشكل مستقل، حتى لو فشلت خطوة الثقة أعلاه try: unverified_context = ssl._create_unverified_context() with socket.create_connection((hostname, port), timeout=REQUEST_TIMEOUT) as sock: with unverified_context.wrap_socket(sock, server_hostname=hostname) as ssock: der_bytes = ssock.getpeercert(binary_form=True) expiry_dt = _read_cert_expiry(der_bytes) if der_bytes else None if expiry_dt is not None: days_remaining = (expiry_dt - datetime.now(timezone.utc)).days expiry_date = expiry_dt.date().isoformat() elif trust_error is None: trust_error = "تعذّر قراءة تاريخ الانتهاء: ثبّت حزمة cryptography عبر: pip install cryptography" except (socket.timeout, socket.gaierror, ConnectionRefusedError, OSError) as e: if trust_error is None: trust_error = str(e) return { "valid": trusted, "days_remaining": days_remaining, "expiry_date": expiry_date, "error": trust_error, } # ---------------------------------------------------------------------------- # 4) Search Response — زمن الرد على طلب بحث # ---------------------------------------------------------------------------- def measure_search(url: str): try: start = time.perf_counter() resp = requests.get(url, timeout=REQUEST_TIMEOUT) elapsed_ms = (time.perf_counter() - start) * 1000 return { "up": resp.status_code < 400, "status_code": resp.status_code, "latency_ms": round(elapsed_ms, 2), "error": None, } except requests.exceptions.RequestException as e: return {"up": False, "status_code": None, "latency_ms": None, "error": str(e)} # ---------------------------------------------------------------------------- # تجميع كل المقاييس في فحص واحد # ---------------------------------------------------------------------------- def collect_metrics(): dns = measure_dns(SITE_HOST) http = measure_http(SITE_URL) ssl_info = check_ssl(SITE_HOST, SITE_PORT) search = measure_search(SEARCH_URL) errors = [e for e in (dns.get("error"), http.get("error"), ssl_info.get("error"), search.get("error")) if e] return { "timestamp": now_iso(), "dns": dns, "http": http, "ssl": ssl_info, "search": search, "error": "; ".join(errors) if errors else None, } # ---------------------------------------------------------------------------- # التخزين: JSON Lines + CSV # ---------------------------------------------------------------------------- def save_jsonl(record: dict): with open(JSONL_PATH, "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") def save_csv(record: dict): file_exists = os.path.isfile(CSV_PATH) row = { "timestamp": record["timestamp"], "dns_time_ms": record["dns"].get("time_ms"), "http_up": record["http"].get("up"), "http_status_code": record["http"].get("status_code"), "http_latency_ms": record["http"].get("latency_ms"), "ssl_valid": record["ssl"].get("valid"), "ssl_days_remaining": record["ssl"].get("days_remaining"), "ssl_expiry_date": record["ssl"].get("expiry_date"), "search_up": record["search"].get("up"), "search_status_code": record["search"].get("status_code"), "search_latency_ms": record["search"].get("latency_ms"), "error": record.get("error"), } with open(CSV_PATH, "a", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=CSV_FIELDS) if not file_exists: writer.writeheader() writer.writerow(row) def save_js(): """تحديث data.js تلقائياً لإتاحتها للمتصفح دون الحاجة لـ fetch/upload""" if not os.path.isfile(JSONL_PATH): return try: records = [] with open(JSONL_PATH, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line: try: records.append(json.loads(line)) except Exception: pass with open(JS_PATH, "w", encoding="utf-8") as f: f.write(f"window.MONITOR_DATA = {json.dumps(records, ensure_ascii=False)};\n") except Exception as e: print(f"تنبيه: تعذّر تحديث data.js: {e}", flush=True) def run_once(): record = collect_metrics() save_jsonl(record) save_csv(record) save_js() status = "UP" if record["http"]["up"] else "DOWN" print( f"[{record['timestamp']}] HTTP={status} " f"latency={record['http']['latency_ms']}ms " f"dns={record['dns']['time_ms']}ms " f"ssl_days_left={record['ssl']['days_remaining']} " f"search={record['search']['latency_ms']}ms", flush=True, ) return record def main(): parser = argparse.ArgumentParser(description="مراقب موقع mithal.space") parser.add_argument("--once", action="store_true", help="فحص واحد فقط ثم الخروج (مناسب لـ cron)") parser.add_argument( "--interval", type=int, default=CHECK_INTERVAL_SECONDS, help="الفاصل الزمني بالثواني بين الفحوصات" ) args = parser.parse_args() if args.once: run_once() return print(f"بدء المراقبة المستمرة لـ {SITE_URL} كل {args.interval} ثانية... (Ctrl+C للإيقاف)", flush=True) while True: try: run_once() except Exception as e: # لا نسمح لأي خطأ غير متوقع بإيقاف الحلقة الرئيسية print(f"خطأ غير متوقع أثناء الفحص: {e}", flush=True) time.sleep(args.interval) if __name__ == "__main__": main()