Dodaj monitor dostępności hostów

host_checker.py

@@ -0,0 +1,174 @@
+#!/usr/bin/env python3
+"""Monitor dostępności hostów przez HTTP lub ICMP."""
+
+from __future__ import annotations
+
+import argparse
+import concurrent.futures
+import platform
+import subprocess
+import sys
+import time
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Iterable
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlparse
+from urllib.request import Request, urlopen
+
+
+@dataclass(frozen=True)
+class CheckResult:
+    target: str
+    method: str
+    available: bool
+    detail: str
+    elapsed_ms: float
+
+
+def normalize_url(target: str) -> str:
+    if "://" not in target:
+        return f"http://{target}"
+    return target
+
+
+def host_from_target(target: str) -> str:
+    parsed = urlparse(target if "://" in target else f"//{target}")
+    return parsed.hostname or target.split("/")[0]
+
+
+def check_http(target: str, timeout: float) -> CheckResult:
+    started = time.perf_counter()
+    url = normalize_url(target)
+    request = Request(url, headers={"User-Agent": "host-checker/1.0"})
+
+    try:
+        with urlopen(request, timeout=timeout) as response:
+            status = response.status
+            available = status < 400
+            detail = f"HTTP {status}"
+    except HTTPError as error:
+        # Odpowiedź HTTP, także 4xx/5xx, oznacza że host jest osiągalny.
+        available = True
+        detail = f"HTTP {error.code}"
+    except (URLError, TimeoutError, OSError) as error:
+        available = False
+        reason = getattr(error, "reason", error)
+        detail = str(reason)
+
+    return CheckResult(target, "http", available, detail, elapsed_ms(started))
+
+
+def check_icmp(target: str, timeout: float) -> CheckResult:
+    started = time.perf_counter()
+    host = host_from_target(target)
+    system = platform.system().lower()
+    if system == "windows":
+        command = ["ping", "-n", "1", "-w", str(max(1, int(timeout * 1000))), host]
+    else:
+        command = ["ping", "-c", "1", "-W", str(max(1, int(timeout))), host]
+
+    try:
+        completed = subprocess.run(
+            command,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            timeout=timeout + 1,
+            check=False,
+        )
+        available = completed.returncode == 0
+        detail = "odpowiedź ICMP" if available else "brak odpowiedzi"
+    except FileNotFoundError:
+        available = False
+        detail = "nie znaleziono programu ping"
+    except subprocess.TimeoutExpired:
+        available = False
+        detail = "przekroczono timeout"
+    except OSError as error:
+        available = False
+        detail = str(error)
+
+    return CheckResult(target, "icmp", available, detail, elapsed_ms(started))
+
+
+def elapsed_ms(started: float) -> float:
+    return (time.perf_counter() - started) * 1000
+
+
+def check(target: str, method: str, timeout: float) -> CheckResult:
+    if method == "http":
+        return check_http(target, timeout)
+    return check_icmp(target, timeout)
+
+
+def load_targets(cli_targets: list[str], filename: str | None) -> list[str]:
+    targets = list(cli_targets)
+    if filename:
+        for line in Path(filename).read_text(encoding="utf-8").splitlines():
+            line = line.strip()
+            if line and not line.startswith("#"):
+                targets.append(line)
+    # Zachowaj kolejność i usuń powtórzenia.
+    return list(dict.fromkeys(targets))
+
+
+def print_results(results: Iterable[CheckResult]) -> None:
+    timestamp = datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S%z")
+    for result in results:
+        state = "OK  " if result.available else "FAIL"
+        print(
+            f"{timestamp} | {state} | {result.method.upper():5} | "
+            f"{result.target} | {result.detail} | {result.elapsed_ms:.0f} ms",
+            flush=True,
+        )
+
+
+def run(args: argparse.Namespace, targets: list[str]) -> int:
+    while True:
+        with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as pool:
+            futures = [pool.submit(check, target, args.method, args.timeout) for target in targets]
+            results = [future.result() for future in futures]
+        print_results(results)
+
+        if args.once:
+            return 0 if all(result.available for result in results) else 1
+        try:
+            time.sleep(args.interval)
+        except KeyboardInterrupt:
+            print("\nZatrzymano.")
+            return 0
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Sprawdzanie dostępności hostów przez HTTP lub ICMP.")
+    parser.add_argument("targets", nargs="*", help="Hosty, adresy IP lub URL-e")
+    parser.add_argument("-f", "--file", help="Plik z hostami: jeden wpis w linii")
+    parser.add_argument("-m", "--method", choices=("http", "icmp"), default="http", help="Metoda testu (domyślnie: http)")
+    parser.add_argument("-t", "--timeout", type=float, default=3.0, help="Timeout w sekundach (domyślnie: 3)")
+    parser.add_argument("-i", "--interval", type=float, default=10.0, help="Odstęp między testami (domyślnie: 10 s)")
+    parser.add_argument("-w", "--workers", type=int, default=10, help="Liczba równoległych testów (domyślnie: 10)")
+    parser.add_argument("--once", action="store_true", help="Wykonaj tylko jeden test i zakończ")
+    return parser
+
+
+def main() -> int:
+    parser = build_parser()
+    args = parser.parse_args()
+    if args.timeout <= 0 or args.interval < 0 or args.workers < 1:
+        parser.error("timeout musi być > 0, interval >= 0, a workers >= 1")
+    try:
+        targets = load_targets(args.targets, args.file)
+    except OSError as error:
+        parser.error(f"nie można odczytać pliku: {error}")
+    if not targets:
+        parser.error("podaj co najmniej jeden host albo użyj --file")
+    try:
+        return run(args, targets)
+    except KeyboardInterrupt:
+        print("\nZatrzymano.")
+        return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())