commit 6ffbce804eee763022b6eb7e66f93ac1fcd72563 Author: Wojtek Date: Wed Oct 14 11:26:53 2026 +0200 Dodaj monitor dostępności hostów diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..55d5157 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +__pycache__/ +*.pyc +hosts.json +.venv/ +venv/ diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..c9ebf2d --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,3 @@ +{ + "python-envs.defaultEnvManager": "ms-python.python:system" +} \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..8efa64c --- /dev/null +++ b/README.md @@ -0,0 +1,44 @@ +# Host checker + +Prosty monitor dostępności hostów przez HTTP albo ICMP. Wymaga Python 3.10+ i nie korzysta z dodatkowych pakietów. + +## Aplikacja graficzna + +Uruchom: + +```bash +python3 app.py +``` + +Aplikacja pozwala dodawać, aktualizować i usuwać hosty, wybierać metodę `HTTP`/`ICMP`, ustawić interwał oraz timeout. Konfiguracja jest automatycznie zapisywana w pliku `hosts.json` obok aplikacji. Tabela pokazuje bieżący status, liczbę testów, liczbę odpowiedzi OK/FAIL, procent dostępności i średni czas odpowiedzi. + +Przycisk „Uruchom monitoring” rozpoczyna cykliczne sprawdzanie wszystkich zapisanych hostów. Wyniki są zbierane równolegle, więc wolny host nie blokuje pozostałych. + +## Przykłady + +Jeden test HTTP: + +```bash +python3 host_checker.py https://example.com --once +``` + +Monitorowanie hostów przez ICMP co 5 sekund: + +```bash +python3 host_checker.py 192.168.1.1 router.local -m icmp -i 5 +``` + +Lista hostów z pliku: + +```text +# komentarze są ignorowane +https://example.com +https://example.org/health +192.168.1.1 +``` + +```bash +python3 host_checker.py --file hosts.txt -m http --timeout 2 +``` + +Kod wyjścia przy `--once` wynosi `0`, jeśli wszystkie testy się powiodły, albo `1`, gdy przynajmniej jeden host jest niedostępny. W przypadku ICMP system musi mieć dostępny program `ping`; na niektórych systemach wymagane są dodatkowe uprawnienia. diff --git a/app.py b/app.py new file mode 100644 index 0000000..00fa2bf --- /dev/null +++ b/app.py @@ -0,0 +1,229 @@ +#!/usr/bin/env python3 +"""Graficzna aplikacja do monitorowania dostępności hostów.""" + +from __future__ import annotations + +import concurrent.futures +import json +import queue +import threading +import tkinter as tk +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from tkinter import messagebox, ttk + +from host_checker import CheckResult, check + + +CONFIG_FILE = Path(__file__).with_name("hosts.json") + + +@dataclass +class Host: + target: str + method: str = "http" + + +class HostMonitorApp(tk.Tk): + def __init__(self) -> None: + super().__init__() + self.title("Monitor dostępności hostów") + self.geometry("1120x620") + self.minsize(850, 450) + self.hosts: list[Host] = [] + self.stats: dict[str, dict[str, float | int | bool | str]] = {} + self.results: queue.Queue[tuple[str, CheckResult]] = queue.Queue() + self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=10) + self.running = False + self.after_id: str | None = None + self.worker_lock = threading.Lock() + self._build_ui() + self._load_config() + self.protocol("WM_DELETE_WINDOW", self._close) + self.after(200, self._consume_results) + + def _build_ui(self) -> None: + controls = ttk.Frame(self, padding=10) + controls.pack(fill="x") + + ttk.Label(controls, text="Host / URL:").grid(row=0, column=0, sticky="w") + self.target_var = tk.StringVar() + ttk.Entry(controls, textvariable=self.target_var, width=38).grid(row=0, column=1, padx=5) + ttk.Label(controls, text="Metoda:").grid(row=0, column=2, sticky="e") + self.method_var = tk.StringVar(value="http") + ttk.Combobox(controls, textvariable=self.method_var, values=("http", "icmp"), state="readonly", width=8).grid(row=0, column=3, padx=5) + ttk.Button(controls, text="Dodaj / aktualizuj", command=self._add_host).grid(row=0, column=4, padx=5) + ttk.Button(controls, text="Usuń zaznaczone", command=self._remove_selected).grid(row=0, column=5, padx=5) + + ttk.Label(controls, text="Interwał [s]:").grid(row=1, column=0, sticky="w", pady=(8, 0)) + self.interval_var = tk.StringVar(value="10") + ttk.Spinbox(controls, from_=1, to=86400, textvariable=self.interval_var, width=8).grid(row=1, column=1, sticky="w", padx=5, pady=(8, 0)) + ttk.Label(controls, text="Timeout [s]:").grid(row=1, column=2, sticky="e", pady=(8, 0)) + self.timeout_var = tk.StringVar(value="3") + ttk.Spinbox(controls, from_=0.1, to=300, increment=0.5, textvariable=self.timeout_var, width=8).grid(row=1, column=3, sticky="w", padx=5, pady=(8, 0)) + self.start_button = ttk.Button(controls, text="Uruchom monitoring", command=self._toggle_monitoring) + self.start_button.grid(row=1, column=4, padx=5, pady=(8, 0)) + ttk.Button(controls, text="Zapisz konfigurację", command=self._save_config).grid(row=1, column=5, padx=5, pady=(8, 0)) + + table_frame = ttk.Frame(self, padding=(10, 0, 10, 10)) + table_frame.pack(fill="both", expand=True) + columns = ("target", "method", "status", "tests", "ok", "fail", "availability", "avg", "last") + self.table = ttk.Treeview(table_frame, columns=columns, show="headings", selectmode="extended") + headings = {"target": "Host / URL", "method": "Metoda", "status": "Status", "tests": "Testy", "ok": "OK", "fail": "FAIL", "availability": "Dostępność", "avg": "Śr. czas", "last": "Ostatni test"} + widths = {"target": 260, "method": 70, "status": 75, "tests": 55, "ok": 45, "fail": 50, "availability": 90, "avg": 80, "last": 180} + for column in columns: + self.table.heading(column, text=headings[column]) + self.table.column(column, width=widths[column], anchor="center") + self.table.column("target", anchor="w") + scrollbar = ttk.Scrollbar(table_frame, orient="vertical", command=self.table.yview) + self.table.configure(yscrollcommand=scrollbar.set) + self.table.pack(side="left", fill="both", expand=True) + scrollbar.pack(side="right", fill="y") + self.table.tag_configure("up", foreground="#16803c") + self.table.tag_configure("down", foreground="#b42318") + self.table.tag_configure("unknown", foreground="#666666") + + self.status_var = tk.StringVar(value="Gotowy. Dodaj hosty i uruchom monitoring.") + ttk.Label(self, textvariable=self.status_var, relief="sunken", anchor="w", padding=5).pack(fill="x", side="bottom") + + def _add_host(self) -> None: + target = self.target_var.get().strip() + method = self.method_var.get() + if not target: + messagebox.showwarning("Brak hosta", "Wpisz adres hosta lub URL.") + return + existing = next((host for host in self.hosts if host.target == target), None) + if existing: + existing.method = method + else: + self.hosts.append(Host(target, method)) + self.target_var.set("") + self._refresh_table() + self._save_config(silent=True) + + def _remove_selected(self) -> None: + selected = set(self.table.selection()) + if not selected: + return + self.hosts = [host for host in self.hosts if host.target not in selected] + for target in selected: + self.stats.pop(target, None) + self._refresh_table() + self._save_config(silent=True) + + def _toggle_monitoring(self) -> None: + if self.running: + self.running = False + if self.after_id: + self.after_cancel(self.after_id) + self.start_button.configure(text="Uruchom monitoring") + self.status_var.set("Monitoring zatrzymany.") + return + if not self.hosts: + messagebox.showwarning("Brak hostów", "Dodaj co najmniej jeden host.") + return + try: + float(self.interval_var.get()) + float(self.timeout_var.get()) + except ValueError: + messagebox.showerror("Błędne ustawienia", "Interwał i timeout muszą być liczbami.") + return + self.running = True + self.start_button.configure(text="Zatrzymaj monitoring") + self._schedule_check(0) + + def _schedule_check(self, delay_ms: int | None = None) -> None: + if not self.running: + return + if delay_ms is None: + delay_ms = max(1, int(float(self.interval_var.get()) * 1000)) + self.after_id = self.after(delay_ms, self._start_check) + + def _start_check(self) -> None: + if not self.running: + return + timeout = float(self.timeout_var.get()) + with self.worker_lock: + for host in self.hosts: + future = self.executor.submit(check, host.target, host.method, timeout) + future.add_done_callback(lambda item, target=host.target: self._put_result(target, item)) + self._schedule_check() + + def _put_result(self, target: str, future: concurrent.futures.Future[CheckResult]) -> None: + try: + self.results.put((target, future.result())) + except Exception as error: # nie przerywaj monitorowania przez pojedynczy host + self.results.put((target, CheckResult(target, "?", False, str(error), 0))) + + def _consume_results(self) -> None: + changed = False + while True: + try: + target, result = self.results.get_nowait() + except queue.Empty: + break + stats = self.stats.setdefault(target, {"tests": 0, "ok": 0, "fail": 0, "total_ms": 0.0, "available": False, "detail": "", "last": ""}) + stats["tests"] = int(stats["tests"]) + 1 + stats["ok"] = int(stats["ok"]) + int(result.available) + stats["fail"] = int(stats["fail"]) + int(not result.available) + stats["total_ms"] = float(stats["total_ms"]) + result.elapsed_ms + stats["available"] = result.available + stats["detail"] = result.detail + stats["last"] = datetime.now().strftime("%H:%M:%S") + changed = True + if changed: + self._refresh_table() + self.after(200, self._consume_results) + + def _refresh_table(self) -> None: + existing = set(self.table.get_children()) + current = {host.target for host in self.hosts} + for target in existing - current: + self.table.delete(target) + for host in self.hosts: + stats = self.stats.get(host.target, {}) + tests = int(stats.get("tests", 0)) + ok = int(stats.get("ok", 0)) + fail = int(stats.get("fail", 0)) + availability = f"{ok / tests * 100:.1f}%" if tests else "-" + avg = f"{float(stats.get('total_ms', 0)) / tests:.0f} ms" if tests else "-" + last = f"{stats.get('last', '-')} ({stats.get('detail', '-')})" if tests else "-" + values = (host.target, host.method.upper(), "OK" if stats.get("available") else ("FAIL" if tests else "-"), tests, ok, fail, availability, avg, last) + tag = "up" if stats.get("available") else ("down" if tests else "unknown") + if host.target in existing: + self.table.item(host.target, values=values, tags=(tag,)) + else: + self.table.insert("", "end", iid=host.target, values=values, tags=(tag,)) + + def _load_config(self) -> None: + if CONFIG_FILE.exists(): + try: + data = json.loads(CONFIG_FILE.read_text(encoding="utf-8")) + self.hosts = [Host(str(item["target"]), str(item.get("method", "http"))) for item in data.get("hosts", []) if item.get("target")] + self.interval_var.set(str(data.get("interval", 10))) + self.timeout_var.set(str(data.get("timeout", 3))) + self._refresh_table() + except (OSError, ValueError, KeyError, TypeError) as error: + messagebox.showwarning("Błąd konfiguracji", f"Nie udało się odczytać hosts.json: {error}") + + def _save_config(self, silent: bool = False) -> None: + try: + data = {"interval": float(self.interval_var.get()), "timeout": float(self.timeout_var.get()), "hosts": [{"target": host.target, "method": host.method} for host in self.hosts]} + CONFIG_FILE.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") + if not silent: + self.status_var.set(f"Zapisano konfigurację: {CONFIG_FILE.name}") + except (OSError, ValueError) as error: + messagebox.showerror("Błąd zapisu", str(error)) + + def _close(self) -> None: + self.running = False + if self.after_id: + self.after_cancel(self.after_id) + self._save_config(silent=True) + self.executor.shutdown(wait=False, cancel_futures=True) + self.destroy() + + +if __name__ == "__main__": + HostMonitorApp().mainloop() diff --git a/host_checker.py b/host_checker.py new file mode 100644 index 0000000..25b3ee6 --- /dev/null +++ b/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())