121 أسطر
3.8 KiB
Python
121 أسطر
3.8 KiB
Python
import time
|
|
import requests
|
|
import json
|
|
import ssl
|
|
import socket
|
|
import os
|
|
import asyncio
|
|
from datetime import datetime, timezone
|
|
from urllib.parse import urlparse
|
|
from fastapi import FastAPI
|
|
from fastapi.responses import JSONResponse, FileResponse
|
|
from contextlib import asynccontextmanager
|
|
|
|
# --- تعريف الـ Lifespan لحل مشكلة الـ startup وتوليد الخلفية ---
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
task = asyncio.create_task(background_monitor())
|
|
yield
|
|
task.cancel()
|
|
|
|
app = FastAPI(lifespan=lifespan)
|
|
|
|
TARGET_URL = "https://mithal.space"
|
|
LOG_FILE = "metrics.json"
|
|
|
|
# --- دالة الفحص الخلفية بـ FastAPI ---
|
|
async def background_monitor():
|
|
while True:
|
|
parsed_url = urlparse(TARGET_URL)
|
|
hostname = parsed_url.netloc
|
|
|
|
start_time = time.time()
|
|
status_code = 0
|
|
latency = 0
|
|
ssl_expiry = "N/A"
|
|
dns_time = 0
|
|
search_time = 0
|
|
|
|
try:
|
|
dns_start = time.time()
|
|
socket.gethostbyname(hostname)
|
|
dns_time = round((time.time() - dns_start) * 1000, 2)
|
|
except:
|
|
dns_time = -1
|
|
|
|
try:
|
|
response = requests.get(TARGET_URL, timeout=10)
|
|
latency = round((time.time() - start_time) * 1000, 2)
|
|
status_code = response.status_code
|
|
except:
|
|
latency = -1
|
|
status_code = 503
|
|
|
|
try:
|
|
search_start = time.time()
|
|
requests.get(f"{TARGET_URL}/search?q=test", timeout=5)
|
|
search_time = round((time.time() - search_start) * 1000, 2)
|
|
except:
|
|
search_time = -1
|
|
|
|
try:
|
|
context = ssl.create_default_context()
|
|
with socket.create_connection((hostname, 443), timeout=5) as sock:
|
|
with context.wrap_socket(sock, server_hostname=hostname) as ssock:
|
|
cert = ssock.getpeercert()
|
|
expiry_date = datetime.strptime(cert['notAfter'], '%b %d %H:%M:%S %Y %Z').replace(tzinfo=timezone.utc)
|
|
days_left = (expiry_date - datetime.now(timezone.utc)).days
|
|
ssl_expiry = f"صحيحة (تنتهي خلال {days_left} يوم)"
|
|
except:
|
|
ssl_expiry = "خطأ في جلب SSL"
|
|
|
|
timestamp_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
record = {
|
|
"timestamp": timestamp_str,
|
|
"status_code": status_code,
|
|
"latency_ms": latency,
|
|
"dns_ms": dns_time,
|
|
"search_response_ms": search_time,
|
|
"ssl_status": ssl_expiry
|
|
}
|
|
|
|
try:
|
|
if os.path.exists(LOG_FILE):
|
|
with open(LOG_FILE, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
else:
|
|
data = []
|
|
except:
|
|
data = []
|
|
|
|
data.append(record)
|
|
if len(data) > 20:
|
|
data = data[-20:]
|
|
|
|
with open(LOG_FILE, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=4, ensure_ascii=False)
|
|
|
|
print(f"[{timestamp_str}] Background Checked {TARGET_URL} - Status: {status_code} - Latency: {latency}ms")
|
|
|
|
# الانتظار لمدة 30 ثانية أو دقيقة بحسب رغبتك
|
|
await asyncio.sleep(60)
|
|
|
|
# --- المسارات (Endpoints) ---
|
|
@app.get("/")
|
|
def index():
|
|
if os.path.exists("dashboard.html"):
|
|
return FileResponse("dashboard.html")
|
|
return {"message": "Dashboard file not found"}
|
|
|
|
@app.get("/api/metrics")
|
|
def get_metrics():
|
|
if os.path.exists(LOG_FILE):
|
|
try:
|
|
with open(LOG_FILE, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
return JSONResponse(content=data)
|
|
except:
|
|
return JSONResponse(content=[])
|
|
return JSONResponse(content=[])
|
|
|