349 أسطر
14 KiB
Python
349 أسطر
14 KiB
Python
#!/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() |