debian-manager/app/ui/pages/group_ops.py

"""Групповые операции по серверам."""
import flet as ft

from app.core import servers as servers_core
from app.core.group_ops import (GroupResult, op_apt_update, op_install_package,
                                op_reboot, op_shell, op_upgrade, run_group)
from app.ui.components import snack


class GroupOpsPage(ft.Column):
    def __init__(self, page: ft.Page, go):
        super().__init__(expand=True, scroll=ft.ScrollMode.AUTO, spacing=12)
        self._page = page
        self.go = go

        servers = servers_core.list_servers()
        self.checks = {s.id: ft.Checkbox(label=f"{s.name} ({s.host})", value=False,
                                         data=s.id) for s in servers}

        self.op_dd = ft.Dropdown(
            label="Операция", value="update", width=300,
            options=[
                ft.dropdown.Option("update", "apt update"),
                ft.dropdown.Option("upgrade", "upgrade"),
                ft.dropdown.Option("dist", "dist-upgrade"),
                ft.dropdown.Option("reboot", "reboot (с ожиданием)"),
                ft.dropdown.Option("shell", "произвольная команда"),
                ft.dropdown.Option("install", "установить пакет"),
            ],
            on_select=lambda e: self._toggle_param(),
        )
        self.param_tf = ft.TextField(label="Команда / пакет", visible=False, width=400)
        self.rolling_cb = ft.Checkbox(label="Rolling (по одному, ждать возврата)", value=False)
        self.failfast_cb = ft.Checkbox(label="Остановиться при первой ошибке", value=False)
        self.results_col = ft.Column(spacing=6)

        self.controls = [
            ft.Text("Групповые операции", size=28, weight=ft.FontWeight.BOLD),
            ft.Row([
                ft.Card(content=ft.Container(padding=14, width=380, content=ft.Column([
                    ft.Text("Серверы", weight=ft.FontWeight.BOLD),
                    *self.checks.values(),
                    ft.TextButton("выбрать все", on_click=lambda e: self._select_all(True)),
                    ft.TextButton("снять все", on_click=lambda e: self._select_all(False)),
                ]))),
                ft.Card(content=ft.Container(padding=14, expand=True, content=ft.Column([
                    ft.Text("Операция", weight=ft.FontWeight.BOLD),
                    self.op_dd,
                    self.param_tf,
                    self.rolling_cb,
                    self.failfast_cb,
                    ft.FilledButton("Запустить", icon=ft.Icons.PLAY_ARROW,
                                    on_click=self._run),
                ], spacing=10))),
            ], vertical_alignment=ft.CrossAxisAlignment.START),
            ft.Text("Результаты", size=18, weight=ft.FontWeight.BOLD),
            self.results_col,
        ]

    def _select_all(self, v):
        for cb in self.checks.values():
            cb.value = v
        self._page.update()

    def _toggle_param(self):
        self.param_tf.visible = self.op_dd.value in ("shell", "install")
        self._page.update()

    def _run(self, e):
        selected = [s for s in servers_core.list_servers()
                    if self.checks.get(s.id) and self.checks[s.id].value]
        if not selected:
            snack(self._page, "Выберите хотя бы один сервер", error=True)
            return

        op_name = self.op_dd.value
        if op_name == "update":
            op = op_apt_update()
        elif op_name == "upgrade":
            op = op_upgrade(False)
        elif op_name == "dist":
            op = op_upgrade(True)
        elif op_name == "reboot":
            op = op_reboot(wait=True)
        elif op_name == "shell":
            if not self.param_tf.value:
                snack(self._page, "Введите команду", error=True)
                return
            op = op_shell(self.param_tf.value)
        else:
            if not self.param_tf.value:
                snack(self._page, "Введите имя пакета", error=True)
                return
            op = op_install_package(self.param_tf.value.split())

        self.results_col.controls = [
            ft.Text(f"запущено на {len(selected)} серверах…", color=ft.Colors.GREY_500)]
        self._page.update()

        async def job():
            def on_status(res: GroupResult):
                self._render_results(res)
            res = await run_group(
                selected, op,
                rolling=self.rolling_cb.value,
                fail_fast=self.failfast_cb.value,
                on_status=on_status,
            )
            self._render_results(res)
            snack(self._page, f"готово: ✅{res.ok_count} ❌{res.err_count}")
        self._page.run_task(job)

    def _render_results(self, res: GroupResult):
        self.results_col.controls = []
        for r in res.results.values():
            icon = {"pending": ft.Icons.HOURGLASS_EMPTY, "running": ft.Icons.SYNC,
                    "success": ft.Icons.CHECK_CIRCLE, "error": ft.Icons.ERROR}.get(r.status)
            color = {"success": ft.Colors.GREEN_400, "error": ft.Colors.RED_400,
                     "running": ft.Colors.BLUE_300}.get(r.status, ft.Colors.GREY_500)
            log_preview = r.log.strip().splitlines()[-3:] if r.log.strip() else []
            self.results_col.controls.append(ft.Card(content=ft.Container(
                padding=10,
                content=ft.Column([
                    ft.Row([ft.Icon(icon, color=color, size=18),
                            ft.Text(r.name, weight=ft.FontWeight.BOLD),
                            ft.Text(r.status, size=12, color=color)]),
                    *[ft.Text(l, size=10, font_family="Consolas",
                              color=ft.Colors.GREY_500) for l in log_preview],
                ], spacing=2),
            )))
        self._page.update()