321 أسطر
14 KiB
Python
321 أسطر
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
monitor.py
|
|
==========
|
|
سكربت مراقبة ذاتي الاكتفاء (بدون أي مكتبات خارجية) لموقع mithal.space.
|
|
|
|
يقوم بأمرين معًا:
|
|
1. تشغيل خمسة فحوصات كل 60 ثانية (Latency, Uptime, SSL, DNS, Search)
|
|
وتخزين النتائج في ملف JSON Lines محلي (data.jsonl).
|
|
2. تشغيل خادم HTTP بسيط يقدّم dashboard.html ويعرض بيانات مجمّعة
|
|
عبر /api/metrics حتى تعمل اللوحة مباشرة بفتح المتصفح على الخادم.
|
|
|
|
التشغيل:
|
|
python monitor.py
|
|
python monitor.py --interval 60 --port 8080
|
|
python monitor.py --once # فحص واحد فقط بدون تشغيل خادم (مناسب لجدولة cron)
|
|
|
|
النشر على غيمة (ghaymah.systems):
|
|
هذا الملف مستقل تمامًا (لا يحتاج pip install لأي حزمة)، لذا يمكن نشره
|
|
داخل أي حاوية Python رسمية بأمر تشغيل واحد:
|
|
FROM python:3.11-slim
|
|
COPY monitor.py dashboard.html /app/
|
|
WORKDIR /app
|
|
EXPOSE 8080
|
|
CMD ["python", "monitor.py"]
|
|
اضبط Health Check Path على /health ومنفذ الخدمة على 8080 من console غيمة.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import socket
|
|
import ssl
|
|
import threading
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from datetime import datetime, timedelta, timezone
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import urlparse, urlencode
|
|
|
|
# --------------------------------------------------------------------------
|
|
# الإعدادات (قابلة للتعديل عبر متغيرات البيئة)
|
|
# --------------------------------------------------------------------------
|
|
TARGET_URL = os.environ.get("TARGET_URL", "https://mithal.space")
|
|
SEARCH_PATH = os.environ.get("SEARCH_PATH", "/search")
|
|
SEARCH_QUERY_PARAM = os.environ.get("SEARCH_QUERY_PARAM", "q")
|
|
SEARCH_QUERY_VALUE = os.environ.get("SEARCH_QUERY_VALUE", "test")
|
|
REQUEST_TIMEOUT = float(os.environ.get("REQUEST_TIMEOUT", 10))
|
|
DATA_FILE = os.environ.get("DATA_FILE", "data.jsonl")
|
|
MAX_RECORDS = int(os.environ.get("MAX_RECORDS", 3000)) # ~50 ساعة عند فحص كل دقيقة
|
|
|
|
_parsed = urlparse(TARGET_URL)
|
|
HOSTNAME = _parsed.hostname
|
|
PORT_443 = 443
|
|
|
|
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
DASHBOARD_FILE = os.path.join(BASE_DIR, "dashboard.html")
|
|
|
|
_store_lock = threading.Lock()
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 1) الفحوصات الخمسة
|
|
# --------------------------------------------------------------------------
|
|
def check_dns(hostname: str = HOSTNAME) -> dict:
|
|
"""يقيس زمن تحليل اسم النطاق (DNS resolution)."""
|
|
start = time.perf_counter()
|
|
try:
|
|
socket.getaddrinfo(hostname, PORT_443)
|
|
return {"ok": True, "dns_ms": round((time.perf_counter() - start) * 1000, 2)}
|
|
except socket.gaierror as e:
|
|
return {"ok": False, "dns_ms": round((time.perf_counter() - start) * 1000, 2), "error": str(e)}
|
|
|
|
|
|
def check_ssl(hostname: str = HOSTNAME, port: int = 443) -> dict:
|
|
"""يفحص صلاحية شهادة SSL وتاريخ انتهائها."""
|
|
try:
|
|
ctx = ssl.create_default_context()
|
|
with socket.create_connection((hostname, port), timeout=REQUEST_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 {"ok": True, "valid": True, "expires_at": not_after.isoformat(), "days_remaining": days_remaining}
|
|
except ssl.SSLCertVerificationError as e:
|
|
return {"ok": True, "valid": False, "error": str(e)}
|
|
except Exception as e:
|
|
return {"ok": False, "valid": False, "error": str(e)}
|
|
|
|
|
|
def _timed_get(url: str):
|
|
"""طلب GET بسيط عبر urllib مع قياس الزمن وإرجاع (status_code, elapsed_ms, error)."""
|
|
start = time.perf_counter()
|
|
req = urllib.request.Request(url, headers={"User-Agent": "mithal-monitor/1.0"})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as resp:
|
|
elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
|
|
return resp.getcode(), elapsed_ms, None
|
|
except urllib.error.HTTPError as e:
|
|
elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
|
|
return e.code, elapsed_ms, None # كود HTTP فعلي (مثل 404/500) وليس خطأ اتصال
|
|
except Exception as e:
|
|
elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
|
|
return None, elapsed_ms, str(e)
|
|
|
|
|
|
def check_latency(url: str = TARGET_URL) -> dict:
|
|
"""يقيس زمن استجابة الصفحة الرئيسية ويحدد حالة توفر الموقع (uptime)."""
|
|
status_code, elapsed_ms, error = _timed_get(url)
|
|
up = status_code is not None and 200 <= status_code < 400
|
|
return {"ok": error is None, "up": up, "status_code": status_code, "latency_ms": elapsed_ms, "error": error}
|
|
|
|
|
|
def check_search(base_url: str = TARGET_URL) -> dict:
|
|
"""يرسل استعلام بحث فعلي ويقيس زمن الرد بمعزل عن الصفحة الرئيسية."""
|
|
query = urlencode({SEARCH_QUERY_PARAM: SEARCH_QUERY_VALUE})
|
|
search_url = base_url.rstrip("/") + SEARCH_PATH + "?" + query
|
|
status_code, elapsed_ms, error = _timed_get(search_url)
|
|
return {"ok": error is None, "status_code": status_code, "search_response_ms": elapsed_ms, "error": error}
|
|
|
|
|
|
def run_full_check() -> dict:
|
|
"""ينفّذ جميع الفحوصات الخمسة ويجمعها في سجل واحد جاهز للتخزين."""
|
|
dns_result = check_dns()
|
|
latency_result = check_latency()
|
|
ssl_result = check_ssl()
|
|
search_result = check_search()
|
|
|
|
return {
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"target": TARGET_URL,
|
|
"up": latency_result.get("up", False),
|
|
"status_code": latency_result.get("status_code"),
|
|
"latency_ms": latency_result.get("latency_ms"),
|
|
"dns_ms": dns_result.get("dns_ms"),
|
|
"ssl_valid": ssl_result.get("valid"),
|
|
"ssl_days_remaining": ssl_result.get("days_remaining"),
|
|
"ssl_expires_at": ssl_result.get("expires_at"),
|
|
"search_response_ms": search_result.get("search_response_ms"),
|
|
"search_status_code": search_result.get("status_code"),
|
|
"error": latency_result.get("error") or ssl_result.get("error") or search_result.get("error"),
|
|
}
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 2) التخزين (JSON Lines) - قابل للتبديل بسهولة إلى append_to_csv أدناه
|
|
# --------------------------------------------------------------------------
|
|
def append_to_store(record: dict, path: str = DATA_FILE, max_lines: int = MAX_RECORDS):
|
|
"""يضيف سجل فحص جديد كسطر JSON، ويقلّم الملف عند تجاوز max_lines."""
|
|
with _store_lock:
|
|
with open(path, "a", encoding="utf-8") as f:
|
|
f.write(json.dumps(record, ensure_ascii=False) + "\n")
|
|
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
lines = f.readlines()
|
|
if len(lines) > max_lines:
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
f.writelines(lines[-max_lines:])
|
|
|
|
|
|
def append_to_csv(record: dict, path: str = "data.csv"):
|
|
"""بديل اختياري: تخزين نفس السجل بصيغة CSV مسطّحة (جدول بيانات)."""
|
|
import csv
|
|
file_exists = os.path.exists(path)
|
|
with _store_lock:
|
|
with open(path, "a", newline="", encoding="utf-8") as f:
|
|
writer = csv.DictWriter(f, fieldnames=list(record.keys()))
|
|
if not file_exists:
|
|
writer.writeheader()
|
|
writer.writerow(record)
|
|
|
|
|
|
def read_all_records(path: str = DATA_FILE) -> list:
|
|
if not os.path.exists(path):
|
|
return []
|
|
records = []
|
|
with _store_lock:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
records.append(json.loads(line))
|
|
except json.JSONDecodeError:
|
|
continue
|
|
return records
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 3) التجميع لأجل /api/metrics (uptime 24h, latency آخر ساعة، SSL، آخر 10)
|
|
# --------------------------------------------------------------------------
|
|
def build_metrics_payload() -> dict:
|
|
records = read_all_records()
|
|
now = datetime.now(timezone.utc)
|
|
|
|
def parse_ts(r):
|
|
try:
|
|
return datetime.fromisoformat(r["timestamp"])
|
|
except Exception:
|
|
return None
|
|
|
|
last_24h = [r for r in records if (ts := parse_ts(r)) and now - ts <= timedelta(hours=24)]
|
|
uptime_percent = round(100 * sum(1 for r in last_24h if r.get("up")) / len(last_24h), 2) if last_24h else None
|
|
|
|
last_hour = [r for r in records if (ts := parse_ts(r)) and now - ts <= timedelta(hours=1)]
|
|
latency_series = [
|
|
{"timestamp": r["timestamp"], "latency_ms": r.get("latency_ms")}
|
|
for r in last_hour if r.get("latency_ms") is not None
|
|
]
|
|
|
|
latest = records[-1] if records else {}
|
|
ssl_info = {
|
|
"valid": latest.get("ssl_valid"),
|
|
"days_remaining": latest.get("ssl_days_remaining"),
|
|
"expires_at": latest.get("ssl_expires_at"),
|
|
}
|
|
|
|
last_10 = records[-10:][::-1]
|
|
|
|
return {
|
|
"uptime_24h_percent": uptime_percent,
|
|
"checks_in_24h": len(last_24h),
|
|
"latency_series": latency_series,
|
|
"ssl": ssl_info,
|
|
"last_checks": last_10,
|
|
"latest": latest,
|
|
"generated_at": now.isoformat(),
|
|
}
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 4) حلقة المراقبة في الخلفية (تعمل كل CHECK_INTERVAL ثانية)
|
|
# --------------------------------------------------------------------------
|
|
def collector_loop(interval: int):
|
|
while True:
|
|
try:
|
|
record = run_full_check()
|
|
append_to_store(record)
|
|
status = "UP" if record["up"] else "DOWN"
|
|
print(f"[{record['timestamp']}] {status} http={record['status_code']} "
|
|
f"latency={record['latency_ms']}ms dns={record['dns_ms']}ms "
|
|
f"ssl_days={record['ssl_days_remaining']} search={record['search_response_ms']}ms")
|
|
if record.get("error"):
|
|
print(f" -> ملاحظة: {record['error']}")
|
|
except Exception as e:
|
|
print(f"[collector] خطأ غير متوقع: {e}")
|
|
time.sleep(interval)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 5) خادم HTTP بسيط (بدون Flask) يقدّم dashboard.html + /api/metrics + /health
|
|
# --------------------------------------------------------------------------
|
|
class DashboardHandler(BaseHTTPRequestHandler):
|
|
def _send_json(self, payload: dict, status: int = 200):
|
|
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_GET(self):
|
|
if self.path in ("/", "/dashboard.html"):
|
|
if os.path.exists(DASHBOARD_FILE):
|
|
with open(DASHBOARD_FILE, "rb") as f:
|
|
body = f.read()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/html; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
else:
|
|
self._send_json({"error": "dashboard.html not found next to monitor.py"}, 404)
|
|
elif self.path == "/health":
|
|
self._send_json({"status": "healthy"})
|
|
elif self.path == "/api/metrics":
|
|
self._send_json(build_metrics_payload())
|
|
else:
|
|
self._send_json({"error": "not found"}, 404)
|
|
|
|
def log_message(self, fmt, *args):
|
|
pass # تعطيل سجلات الخادم الافتراضية المزعجة؛ نطبع فقط سجلات الفحص
|
|
|
|
|
|
def run_server(port: int):
|
|
server = ThreadingHTTPServer(("0.0.0.0", port), DashboardHandler)
|
|
print(f"Dashboard server running on http://0.0.0.0:{port} (health: /health, api: /api/metrics)")
|
|
server.serve_forever()
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# نقطة الدخول
|
|
# --------------------------------------------------------------------------
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Monitor mithal.space + serve dashboard.")
|
|
parser.add_argument("--interval", type=int, default=60, help="ثواني بين كل فحص (افتراضي 60)")
|
|
parser.add_argument("--port", type=int, default=int(os.environ.get("PORT", 8080)))
|
|
parser.add_argument("--once", action="store_true", help="نفّذ فحصًا واحدًا فقط بدون تشغيل الخادم (مناسب لـ cron)")
|
|
args = parser.parse_args()
|
|
|
|
if args.once:
|
|
record = run_full_check()
|
|
append_to_store(record)
|
|
print(json.dumps(record, indent=2, ensure_ascii=False))
|
|
return
|
|
|
|
collector_thread = threading.Thread(target=collector_loop, args=(args.interval,), daemon=True)
|
|
collector_thread.start()
|
|
|
|
run_server(args.port)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except KeyboardInterrupt:
|
|
print("\nتم إيقاف المراقبة بواسطة المستخدم.")
|