300 أسطر
10 KiB
Python
300 أسطر
10 KiB
Python
#!/usr/bin/env python3
|
|
|
|
"""Collect availability metrics for https://mithal.space every minute."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import socket
|
|
import ssl
|
|
import sys
|
|
import time
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from html.parser import HTMLParser
|
|
from pathlib import Path
|
|
from typing import Any
|
|
import tempfile
|
|
from urllib import parse, request
|
|
|
|
|
|
TARGET_URL = "https://mithal.space"
|
|
METRICS_FILE = Path(__file__).with_name("metrics.json")
|
|
USER_AGENT = "Mozilla/5.0 (compatible; MithalMonitor/1.0)"
|
|
|
|
|
|
@dataclass
|
|
class SearchConfig:
|
|
"""Discovered search request details from the homepage."""
|
|
|
|
action_url: str
|
|
method: str
|
|
query_param: str
|
|
fixed_params: dict[str, str]
|
|
|
|
|
|
class SearchFormParser(HTMLParser):
|
|
"""Extract candidate search forms from the homepage HTML."""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.forms: list[dict[str, Any]] = []
|
|
self._current_form: dict[str, Any] | None = None
|
|
|
|
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
|
attributes = {key.lower(): value for key, value in attrs}
|
|
|
|
if tag.lower() == "form":
|
|
self._current_form = {
|
|
"action": attributes.get("action", ""),
|
|
"method": (attributes.get("method") or "GET").upper(),
|
|
"inputs": [],
|
|
}
|
|
return
|
|
|
|
if self._current_form is not None and tag.lower() == "input":
|
|
self._current_form["inputs"].append(
|
|
{
|
|
"type": (attributes.get("type") or "text").lower(),
|
|
"name": attributes.get("name", ""),
|
|
"value": attributes.get("value", ""),
|
|
}
|
|
)
|
|
|
|
def handle_endtag(self, tag: str) -> None:
|
|
if tag.lower() == "form" and self._current_form is not None:
|
|
self.forms.append(self._current_form)
|
|
self._current_form = None
|
|
|
|
|
|
def now_iso() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def http_get(url: str, timeout: int = 20) -> tuple[float, int, bytes]:
|
|
"""Perform a full HTTP GET and return latency, status code, and body."""
|
|
|
|
started_at = time.perf_counter()
|
|
req = request.Request(url, headers={"User-Agent": USER_AGENT})
|
|
|
|
with request.urlopen(req, timeout=timeout) as response:
|
|
body = response.read()
|
|
latency_ms = (time.perf_counter() - started_at) * 1000
|
|
return latency_ms, int(response.status), body
|
|
|
|
|
|
def fetch_homepage_html(base_url: str) -> str:
|
|
req = request.Request(base_url, headers={"User-Agent": USER_AGENT})
|
|
with request.urlopen(req, timeout=20) as response:
|
|
return response.read().decode("utf-8", errors="replace")
|
|
|
|
|
|
def discover_search_config(base_url: str) -> SearchConfig:
|
|
"""Inspect the homepage and derive the real search request shape."""
|
|
|
|
homepage_html = fetch_homepage_html(base_url)
|
|
parser = SearchFormParser()
|
|
parser.feed(homepage_html)
|
|
|
|
for form in parser.forms:
|
|
action = (form.get("action") or "").strip()
|
|
method = (form.get("method") or "GET").upper()
|
|
inputs = form.get("inputs") or []
|
|
|
|
query_input = None
|
|
fixed_params: dict[str, str] = {}
|
|
|
|
for input_item in inputs:
|
|
input_name = (input_item.get("name") or "").strip()
|
|
input_type = (input_item.get("type") or "text").lower()
|
|
input_value = input_item.get("value") or ""
|
|
|
|
if not input_name:
|
|
continue
|
|
|
|
if input_type in {"text", "search", "url", "email", "tel", "hidden"} and query_input is None:
|
|
if input_type != "hidden":
|
|
query_input = input_name
|
|
|
|
if input_type == "hidden" and input_value:
|
|
fixed_params[input_name] = input_value
|
|
|
|
if query_input is None:
|
|
for input_item in inputs:
|
|
input_name = (input_item.get("name") or "").strip()
|
|
input_type = (input_item.get("type") or "text").lower()
|
|
if input_name and input_type != "hidden":
|
|
query_input = input_name
|
|
break
|
|
|
|
if query_input and ("search" in action.lower() or any("search" in (item.get("name") or "").lower() for item in inputs)):
|
|
return SearchConfig(
|
|
action_url=parse.urljoin(base_url, action or "/search"),
|
|
method=method,
|
|
query_param=query_input,
|
|
fixed_params=fixed_params,
|
|
)
|
|
|
|
return SearchConfig(
|
|
action_url=parse.urljoin(base_url, "/search"),
|
|
method="GET",
|
|
query_param="q",
|
|
fixed_params={},
|
|
)
|
|
|
|
|
|
def measure_dns_lookup(hostname: str) -> float:
|
|
started_at = time.perf_counter()
|
|
socket.getaddrinfo(hostname, 443, type=socket.SOCK_STREAM)
|
|
return (time.perf_counter() - started_at) * 1000
|
|
|
|
|
|
def measure_ssl_certificate(hostname: str, port: int = 443) -> tuple[bool, str]:
|
|
pem_certificate = ssl.get_server_certificate((hostname, port), timeout=20)
|
|
|
|
with tempfile.NamedTemporaryFile("w", delete=True) as temp_file:
|
|
temp_file.write(pem_certificate)
|
|
temp_file.flush()
|
|
certificate = ssl._ssl._test_decode_cert(temp_file.name) # type: ignore[attr-defined]
|
|
|
|
not_after = certificate.get("notAfter")
|
|
if not not_after:
|
|
raise ValueError("SSL certificate does not expose an expiration date")
|
|
|
|
expiry_timestamp = ssl.cert_time_to_seconds(not_after)
|
|
expiry_dt = datetime.fromtimestamp(expiry_timestamp, tz=timezone.utc)
|
|
is_valid = datetime.now(timezone.utc) < expiry_dt
|
|
return is_valid, expiry_dt.isoformat()
|
|
|
|
|
|
def measure_search_response(base_url: str, query: str = "test") -> float:
|
|
search_config = discover_search_config(base_url)
|
|
params = dict(search_config.fixed_params)
|
|
params[search_config.query_param] = query
|
|
|
|
if search_config.method != "GET":
|
|
raise ValueError(f"Unsupported search method: {search_config.method}")
|
|
|
|
search_url = f"{search_config.action_url}?{parse.urlencode(params)}"
|
|
latency_ms, _, _ = http_get(search_url)
|
|
return latency_ms
|
|
|
|
|
|
def metric_record(name: str, value: Any, error_message: str | None = None) -> dict[str, Any]:
|
|
record = {name: value}
|
|
if error_message is not None:
|
|
record[f"{name}_error"] = error_message
|
|
return record
|
|
|
|
|
|
def collect_metrics(base_url: str = TARGET_URL) -> dict[str, Any]:
|
|
hostname = parse.urlparse(base_url).hostname or "mithal.space"
|
|
record: dict[str, Any] = {"timestamp": now_iso()}
|
|
|
|
try:
|
|
latency_ms, status_code, _ = http_get(base_url)
|
|
record.update(metric_record("latency_ms", round(latency_ms, 2)))
|
|
record.update(metric_record("status_code", status_code))
|
|
record["uptime"] = status_code == 200
|
|
except Exception as exc: # noqa: BLE001 - keep the monitor resilient.
|
|
record.update(metric_record("latency_ms", None, str(exc)))
|
|
record.update(metric_record("status_code", None, str(exc)))
|
|
record["uptime"] = None
|
|
record["uptime_error"] = str(exc)
|
|
|
|
try:
|
|
ssl_valid, ssl_expiry = measure_ssl_certificate(hostname)
|
|
record["ssl_valid"] = ssl_valid
|
|
record["ssl_expiry"] = ssl_expiry
|
|
except Exception as exc: # noqa: BLE001 - keep the monitor resilient.
|
|
record["ssl_valid"] = None
|
|
record["ssl_valid_error"] = str(exc)
|
|
record["ssl_expiry"] = None
|
|
record["ssl_expiry_error"] = str(exc)
|
|
|
|
try:
|
|
dns_lookup_ms = measure_dns_lookup(hostname)
|
|
record["dns_lookup_ms"] = round(dns_lookup_ms, 2)
|
|
except Exception as exc: # noqa: BLE001 - keep the monitor resilient.
|
|
record["dns_lookup_ms"] = None
|
|
record["dns_lookup_ms_error"] = str(exc)
|
|
|
|
try:
|
|
search_response_ms = measure_search_response(base_url)
|
|
record["search_response_ms"] = round(search_response_ms, 2)
|
|
except Exception as exc: # noqa: BLE001 - keep the monitor resilient.
|
|
record["search_response_ms"] = None
|
|
record["search_response_ms_error"] = str(exc)
|
|
|
|
return record
|
|
|
|
|
|
def load_metrics(path: Path) -> list[dict[str, Any]]:
|
|
if not path.exists():
|
|
return []
|
|
|
|
try:
|
|
with path.open("r", encoding="utf-8") as file_handle:
|
|
data = json.load(file_handle)
|
|
except Exception:
|
|
return []
|
|
|
|
if isinstance(data, list):
|
|
return [record for record in data if isinstance(record, dict)]
|
|
|
|
return []
|
|
|
|
|
|
def save_metrics(path: Path, records: list[dict[str, Any]]) -> None:
|
|
temp_path = path.with_suffix(".tmp")
|
|
with temp_path.open("w", encoding="utf-8") as file_handle:
|
|
json.dump(records, file_handle, ensure_ascii=False, indent=2)
|
|
file_handle.write("\n")
|
|
temp_path.replace(path)
|
|
|
|
|
|
def append_metric_record(record: dict[str, Any], path: Path = METRICS_FILE) -> None:
|
|
records = load_metrics(path)
|
|
records.append(record)
|
|
save_metrics(path, records)
|
|
|
|
|
|
def run_monitor(interval_seconds: int, once: bool, target_url: str, metrics_file: Path) -> None:
|
|
while True:
|
|
record = collect_metrics(target_url)
|
|
append_metric_record(record, metrics_file)
|
|
print(json.dumps(record, ensure_ascii=False))
|
|
|
|
if once:
|
|
return
|
|
|
|
time.sleep(interval_seconds)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Collect monitoring metrics for mithal.space")
|
|
parser.add_argument("--interval", type=int, default=60, help="Seconds between metric collections")
|
|
parser.add_argument("--once", action="store_true", help="Collect one record and exit")
|
|
parser.add_argument("--target-url", default=TARGET_URL, help="Website to monitor")
|
|
parser.add_argument("--metrics-file", default=str(METRICS_FILE), help="Path to metrics.json")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
metrics_file = Path(args.metrics_file)
|
|
|
|
try:
|
|
run_monitor(args.interval, args.once, args.target_url, metrics_file)
|
|
except KeyboardInterrupt:
|
|
return 0
|
|
except Exception as exc: # noqa: BLE001 - the outer loop should never crash silently.
|
|
print(f"monitor failed: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main()) |