first commit
هذا الالتزام موجود في:
327
q5-mithal-monitor/monitor.py
Normal file
327
q5-mithal-monitor/monitor.py
Normal file
@@ -0,0 +1,327 @@
|
||||
#!/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")
|
||||
|
||||
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 run_once():
|
||||
record = collect_metrics()
|
||||
save_jsonl(record)
|
||||
save_csv(record)
|
||||
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()
|
||||
المرجع في مشكلة جديدة
حظر مستخدم