313 lines
14 KiB
Python
313 lines
14 KiB
Python
#!/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
|
|
|
|
|
|
__version__ = "0.0.1"
|
|
CONFIG_FILE = Path(__file__).with_name("hosts.json")
|
|
|
|
|
|
@dataclass
|
|
class Host:
|
|
target: str
|
|
method: str = "http"
|
|
interval: float = 10.0
|
|
|
|
|
|
class HostMonitorApp(tk.Tk):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.title(f"Monitor dostępności hostów v{__version__}")
|
|
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.host_after_ids: dict[str, str] = {}
|
|
self.selected_target: 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", "interval", "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", "interval": "Interwał [s]", "status": "Status", "tests": "Testy", "ok": "OK", "fail": "FAIL", "availability": "Dostępność", "avg": "Śr. czas", "last": "Ostatni test"}
|
|
widths = {"target": 240, "method": 70, "interval": 85, "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.bind("<<TreeviewSelect>>", self._select_host)
|
|
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 _select_host(self, _event: tk.Event[tk.Misc] | None = None) -> None:
|
|
selected = self.table.selection()
|
|
if len(selected) != 1:
|
|
self.selected_target = None
|
|
return
|
|
target = selected[0]
|
|
host = next((item for item in self.hosts if item.target == target), None)
|
|
if host is None:
|
|
self.selected_target = None
|
|
return
|
|
self.selected_target = target
|
|
self.target_var.set(host.target)
|
|
self.method_var.set(host.method)
|
|
self.interval_var.set(str(host.interval).rstrip("0").rstrip("."))
|
|
|
|
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
|
|
try:
|
|
interval = float(self.interval_var.get())
|
|
if interval <= 0:
|
|
raise ValueError
|
|
except ValueError:
|
|
messagebox.showerror("Błędny interwał", "Interwał musi być liczbą większą od zera.")
|
|
return
|
|
|
|
selected_host = next((host for host in self.hosts if host.target == self.selected_target), None)
|
|
duplicate = next((host for host in self.hosts if host.target == target and host is not selected_host), None)
|
|
if duplicate:
|
|
messagebox.showwarning("Host już istnieje", "Host o tym adresie jest już na liście.")
|
|
return
|
|
|
|
was_update = selected_host is not None
|
|
if selected_host:
|
|
old_target = selected_host.target
|
|
if self.running:
|
|
old_after_id = self.host_after_ids.pop(old_target, None)
|
|
if old_after_id:
|
|
self.after_cancel(old_after_id)
|
|
selected_host.target = target
|
|
selected_host.method = method
|
|
selected_host.interval = interval
|
|
if old_target != target:
|
|
if old_target in self.stats:
|
|
self.stats[target] = self.stats.pop(old_target)
|
|
self.selected_target = target
|
|
else:
|
|
existing = next((host for host in self.hosts if host.target == target), None)
|
|
if existing:
|
|
existing.method = method
|
|
existing.interval = interval
|
|
self.selected_target = target
|
|
else:
|
|
self.hosts.append(Host(target, method, interval))
|
|
self.selected_target = target
|
|
if not was_update:
|
|
self.target_var.set("")
|
|
self._refresh_table()
|
|
if was_update and self.running:
|
|
self._schedule_check(self.selected_target or target)
|
|
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)
|
|
if self.selected_target in selected:
|
|
self.selected_target = None
|
|
self._refresh_table()
|
|
self._save_config(silent=True)
|
|
|
|
def _toggle_monitoring(self) -> None:
|
|
if self.running:
|
|
self.running = False
|
|
for after_id in self.host_after_ids.values():
|
|
self.after_cancel(after_id)
|
|
self.host_after_ids.clear()
|
|
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.timeout_var.get())
|
|
if any(host.interval <= 0 for host in self.hosts):
|
|
raise ValueError
|
|
except ValueError:
|
|
messagebox.showerror("Błędne ustawienia", "Interwały hostów i timeout muszą być poprawnymi liczbami.")
|
|
return
|
|
self.running = True
|
|
self.start_button.configure(text="Zatrzymaj monitoring")
|
|
for host in self.hosts:
|
|
self._schedule_check(host.target, 0)
|
|
|
|
def _schedule_check(self, target: str, delay_ms: int | None = None) -> None:
|
|
if not self.running:
|
|
return
|
|
if delay_ms is None:
|
|
host = next((item for item in self.hosts if item.target == target), None)
|
|
if host is None:
|
|
return
|
|
delay_ms = max(1, int(host.interval * 1000))
|
|
self.host_after_ids[target] = self.after(delay_ms, lambda: self._start_check(target))
|
|
|
|
def _start_check(self, target: str) -> None:
|
|
if not self.running:
|
|
return
|
|
timeout = float(self.timeout_var.get())
|
|
host = next((item for item in self.hosts if item.target == target), None)
|
|
if host is None:
|
|
self.host_after_ids.pop(target, None)
|
|
return
|
|
with self.worker_lock:
|
|
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(target)
|
|
|
|
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(), f"{host.interval:g}", "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,))
|
|
if self.selected_target in current:
|
|
self.table.selection_set(self.selected_target)
|
|
self.table.focus(self.selected_target)
|
|
|
|
def _load_config(self) -> None:
|
|
if CONFIG_FILE.exists():
|
|
try:
|
|
data = json.loads(CONFIG_FILE.read_text(encoding="utf-8"))
|
|
default_interval = float(data.get("interval", 10))
|
|
self.hosts = [
|
|
Host(
|
|
str(item["target"]),
|
|
str(item.get("method", "http")),
|
|
float(item.get("interval", default_interval)),
|
|
)
|
|
for item in data.get("hosts", [])
|
|
if item.get("target")
|
|
]
|
|
self.interval_var.set(str(default_interval))
|
|
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, "interval": host.interval} 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
|
|
for after_id in self.host_after_ids.values():
|
|
self.after_cancel(after_id)
|
|
self.host_after_ids.clear()
|
|
self._save_config(silent=True)
|
|
self.executor.shutdown(wait=False, cancel_futures=True)
|
|
self.destroy()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
HostMonitorApp().mainloop()
|