projektX/host_checker.py

#!/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())