"""View helpers: turn virsh/system output into plain-language data for templates.""" import os import re import shutil from datetime import datetime from pathlib import Path from typing import Any VM_STATES = { "running": ("Работает", "success"), "idle": ("Работает", "success"), "blocked": ("Работает", "success"), "paused": ("Приостановлена", "warning"), "pmsuspended": ("Спящий режим", "warning"), "in shutdown": ("Выключается", "warning"), "shut off": ("Выключена", "neutral"), "shutoff": ("Выключена", "neutral"), "crashed": ("Сбой", "danger"), "dying": ("Останавливается", "warning"), } OPERATION_STATES = { "success": ("Готово", "success"), "error": ("Ошибка", "danger"), "running": ("Выполняется", "info"), "queued": ("В очереди", "neutral"), } SERVICE_LABELS = { "libvirtd": "Виртуализация", "virtlogd": "Журналы машин", "cockpit": "Cockpit", "dashboard": "Экран сервера", "web": "Панель управления", } PORT_PRESETS = [ {"port": "22", "label": "SSH — удалённый доступ к Linux", "protocol": "tcp"}, {"port": "80", "label": "Сайт (HTTP)", "protocol": "tcp"}, {"port": "443", "label": "Сайт (HTTPS)", "protocol": "tcp"}, {"port": "3389", "label": "Удалённый рабочий стол Windows (RDP)", "protocol": "tcp"}, {"port": "25565", "label": "Minecraft", "protocol": "tcp"}, {"port": "3306", "label": "MySQL / MariaDB", "protocol": "tcp"}, {"port": "5432", "label": "PostgreSQL", "protocol": "tcp"}, {"port": "1194", "label": "OpenVPN", "protocol": "udp"}, {"port": "51820", "label": "WireGuard VPN", "protocol": "udp"}, {"port": "5060", "label": "SIP-телефония", "protocol": "udp"}, {"port": "10000-20000", "label": "Голос VoIP (RTP)", "protocol": "udp"}, ] WELL_KNOWN_PORTS = {int(item["port"]): item["label"].split(" (")[0].split(" — ")[0] for item in PORT_PRESETS if item["port"].isdigit()} VM_PRESETS = [ {"id": "mini", "title": "Мини", "hint": "Роутер, VPN, лёгкий сервис", "vcpus": 1, "memory": 1024, "disk_size": 10}, {"id": "standard", "title": "Стандарт", "hint": "Сайт, база данных, Linux-сервер", "vcpus": 2, "memory": 4096, "disk_size": 40}, {"id": "power", "title": "Мощная", "hint": "Windows, 1С, игровой сервер", "vcpus": 4, "memory": 8192, "disk_size": 80}, ] def vm_state(state: str) -> dict[str, Any]: key = (state or "").strip().lower() label, tone = VM_STATES.get(key, (state or "Неизвестно", "neutral")) return {"label": label, "tone": tone, "running": tone == "success", "raw": state} def operation_state(status: str) -> dict[str, str]: label, tone = OPERATION_STATES.get((status or "").lower(), (status or "—", "neutral")) return {"label": label, "tone": tone} def service_state(state: str) -> dict[str, str]: if state == "active": return {"label": "Работает", "tone": "success"} if state in ("activating", "reloading"): return {"label": "Запускается", "tone": "warning"} if state == "failed": return {"label": "Сбой", "tone": "danger"} return {"label": "Остановлена", "tone": "neutral"} def _kib_to_mb(value: str) -> int: match = re.search(r"(\d+)", value or "") if not match: return 0 amount = int(match.group(1)) low = (value or "").lower() if "mib" in low: return amount if "gib" in low: return amount * 1024 return amount // 1024 def parse_dominfo(text: str) -> dict[str, Any]: fields: dict[str, str] = {} for line in (text or "").splitlines(): if ":" in line: key, value = line.split(":", 1) fields[key.strip().lower()] = value.strip() try: vcpus = int(fields.get("cpu(s)", "0") or 0) except ValueError: vcpus = 0 autostart = fields.get("autostart", "").lower() return { "state": fields.get("state", ""), "vcpus": vcpus, "memory_mb": _kib_to_mb(fields.get("used memory", "")) or _kib_to_mb(fields.get("max memory", "")), "max_memory_mb": _kib_to_mb(fields.get("max memory", "")), "autostart": autostart in ("enable", "enabled", "yes", "on"), "autostart_known": bool(autostart), "persistent": fields.get("persistent", "") == "yes", "os_type": fields.get("os type", ""), "uuid": fields.get("uuid", ""), } def _table_rows(text: str) -> list[list[str]]: rows = [] for line in (text or "").splitlines(): stripped = line.strip() if not stripped or set(stripped) <= {"-", " "}: continue rows.append(stripped.split()) return rows[1:] if rows else [] def parse_domblklist(text: str) -> list[dict[str, str]]: disks = [] for parts in _table_rows(text): if len(parts) >= 4: kind, device, target, source = parts[0], parts[1], parts[2], " ".join(parts[3:]) elif len(parts) >= 2: kind, device, target, source = "file", "disk", parts[0], " ".join(parts[1:]) else: continue if source == "-": source = "" disks.append({ "type": kind, "device": device, "target": target, "source": source, "name": Path(source).name if source else "Пусто", "is_cdrom": device == "cdrom", "label": "Привод CD/DVD" if device == "cdrom" else "Диск", }) return disks def parse_domiflist(text: str) -> list[dict[str, str]]: interfaces = [] for parts in _table_rows(text): if len(parts) < 2: continue padded = parts + ["—"] * (5 - len(parts)) interfaces.append({"interface": padded[0], "type": padded[1], "source": padded[2], "model": padded[3], "mac": padded[4]}) return interfaces def format_mb(value: int | float) -> str: value = float(value or 0) if value >= 1024: gb = value / 1024 return f"{gb:.0f} ГБ" if gb >= 10 or gb.is_integer() else f"{gb:.1f} ГБ" return f"{value:.0f} МБ" def format_gb(value: float) -> str: if value >= 1024: return f"{value / 1024:.1f} ТБ" return f"{value:.0f} ГБ" if value >= 10 else f"{value:.1f} ГБ" def _meminfo() -> dict[str, int]: data = {} try: for line in Path("/proc/meminfo").read_text().splitlines(): key, value = line.split(":", 1) data[key] = int(value.split()[0]) except (OSError, ValueError): pass return data def host_stats(storage_path: Path) -> dict[str, Any]: cpu_count = os.cpu_count() or 1 try: load1 = float(Path("/proc/loadavg").read_text().split()[0]) except (OSError, ValueError, IndexError): load1 = 0.0 mem = _meminfo() mem_total = mem.get("MemTotal", 0) / 1024 / 1024 mem_used = max(0.0, mem_total - mem.get("MemAvailable", 0) / 1024 / 1024) path = storage_path if storage_path.exists() else Path("/") try: usage = shutil.disk_usage(path) disk_total, disk_used, disk_free = usage.total / 1024 ** 3, usage.used / 1024 ** 3, usage.free / 1024 ** 3 except OSError: disk_total = disk_used = disk_free = 0.0 def pct(used: float, total: float) -> int: return int(round(min(100.0, max(0.0, used / total * 100)))) if total else 0 return { "cpu_count": cpu_count, "cpu_pct": pct(load1, cpu_count), "load1": load1, "mem_total": format_gb(mem_total), "mem_used": format_gb(mem_used), "mem_pct": pct(mem_used, mem_total), "disk_total": format_gb(disk_total), "disk_used": format_gb(disk_used), "disk_free": format_gb(disk_free), "disk_pct": pct(disk_used, disk_total), } def level_tone(pct: int) -> str: return "danger" if pct >= 90 else "warning" if pct >= 75 else "success" def greeting(now: datetime | None = None) -> str: hour = (now or datetime.now()).hour if 5 <= hour < 12: return "Доброе утро" if 12 <= hour < 18: return "Добрый день" if 18 <= hour < 23: return "Добрый вечер" return "Доброй ночи" def plural(count: int, one: str, few: str, many: str) -> str: count = abs(int(count)) if count % 10 == 1 and count % 100 != 11: return one if 2 <= count % 10 <= 4 and not 12 <= count % 100 <= 14: return few return many def port_service(port: int) -> str: return WELL_KNOWN_PORTS.get(int(port or 0), "") def format_bytes(value: int | float) -> str: value = float(value or 0) for unit in ("Б", "КБ", "МБ", "ГБ", "ТБ"): if value < 1024 or unit == "ТБ": return f"{value:.0f} {unit}" if value >= 10 or unit == "Б" else f"{value:.1f} {unit}" value /= 1024 return f"{value:.0f} ТБ"