"""Вкладка Пакеты: список с разбивкой, установка/удаление.""" import flet as ft from app.core.packages import (Package, apt_search, install, list_packages, remove, split_packages) from app.core.ssh import SSHClient from app.core.tasks import tasks from app.db.models import Server from app.ui.components import append_log, confirm_dialog, log_view, snack class PackagesTab(ft.Column): def __init__(self, page: ft.Page, srv: Server): super().__init__(expand=True, spacing=8) self._page = page self.srv = srv self._all: list[Package] = [] self._filter = "user" # all/system/user/deps self.search = ft.TextField( hint_text="Поиск по имени…", prefix_icon=ft.Icons.SEARCH, on_change=lambda e: self._render_table(), expand=True, dense=True, ) self.chips = ft.Row([ ft.Chip(label=ft.Text("Пользовательские"), on_select=lambda e: self._set_filter("user"), selected=True), ft.Chip(label=ft.Text("Системные"), on_select=lambda e: self._set_filter("system")), ft.Chip(label=ft.Text("Зависимости"), on_select=lambda e: self._set_filter("deps")), ft.Chip(label=ft.Text("Все"), on_select=lambda e: self._set_filter("all")), ]) self.count_text = ft.Text("", size=12, color=ft.Colors.GREY_500) self.table = ft.DataTable( columns=[ ft.DataColumn(ft.Text("Пакет")), ft.DataColumn(ft.Text("Версия")), ft.DataColumn(ft.Text("Тип")), ft.DataColumn(ft.Text("")), ], rows=[], expand=True, ) self.table_col = ft.Column([self.table], scroll=ft.ScrollMode.AUTO, expand=True) self.log = log_view() self.log_container = ft.Container( height=200, border=ft.Border.all(1, ft.Colors.GREY_800), border_radius=8, padding=8, content=self.log, visible=False) self.controls = [ ft.Container(height=8), ft.Row([ ft.FilledButton("Установить пакет", icon=ft.Icons.ADD, on_click=self._install_dialog), ft.OutlinedButton("Обновить список", icon=ft.Icons.REFRESH, on_click=lambda e: self._reload()), ft.OutlinedButton("Autoremove", icon=ft.Icons.CLEANING_SERVICES, on_click=self._autoremove), self.count_text, ]), ft.Row([self.search], spacing=10), self.chips, ft.Container(expand=True, content=self.table_col), self.log_container, ] self._page.run_task(self._load) def _set_filter(self, f): self._filter = f for c in self.chips.controls: c.selected = False idx = {"user": 0, "system": 1, "deps": 2, "all": 3}[f] self.chips.controls[idx].selected = True self._render_table() async def _load(self): try: async with SSHClient(self.srv) as c: self._all = await list_packages(c) except Exception as e: snack(self._page, f"Ошибка загрузки: {e}", error=True) self._render_table() def _reload(self): self.count_text.value = "загрузка…" self._page.update() self._page.run_task(self._load) def _render_table(self): groups = split_packages(self._all) items = self._all if self._filter == "all" else groups.get(self._filter, []) q = (self.search.value or "").lower() if q: items = [p for p in items if q in p.name.lower()] self.count_text.value = f"показано {len(items)} из {len(self._all)}" kind_label = lambda p: ("системный" if p.is_system else "пользовательский" if p.manual else "зависимость") kind_color = lambda p: (ft.Colors.BLUE_300 if p.is_system else ft.Colors.GREEN_300 if p.manual else ft.Colors.GREY_500) self.table.rows = [ ft.DataRow(cells=[ ft.DataCell(ft.Text(p.name, selectable=True)), ft.DataCell(ft.Text(p.version, size=12)), ft.DataCell(ft.Text(kind_label(p), size=12, color=kind_color(p))), ft.DataCell(ft.IconButton( ft.Icons.DELETE_OUTLINE, icon_size=18, tooltip="Удалить", on_click=lambda e, p=p: self._remove(p), )), ]) for p in items[:300] ] if len(items) > 300: self.count_text.value += " (первые 300)" self._page.update() # ---------- действия ---------- def _show_log(self): self.log_container.visible = True self._page.update() def _install_dialog(self, e): name_tf = ft.TextField(label="Имя пакета (или несколько через пробел)", autofocus=True, width=400) results = ft.Column(tight=True, spacing=2) def search(e): async def _s(): results.controls = [ft.Text("поиск…", size=12)] dlg.content.update() try: async with SSHClient(self.srv) as c: lines = await apt_search(c, name_tf.value) results.controls = [ ft.TextButton( l[:90], on_click=lambda e, n=l.split(" ")[0]: setattr( name_tf, "value", n) or dlg.content.update(), style=ft.ButtonStyle(padding=2), ) for l in lines[:15] ] or [ft.Text("ничего не найдено", size=12)] except Exception as ex: results.controls = [ft.Text(f"ошибка: {ex}", size=12)] dlg.content.update() self._page.run_task(_s) def do_install(e): names = name_tf.value.split() if not names: return self._page.pop_dialog() self._show_log() async def job(log): async with SSHClient(self.srv) as c: log(f"$ apt-get install -y {' '.join(names)}") stream = await install(c, names) async for line in stream: log(line) tasks.start("install", f"install {' '.join(names)}", self.srv.id, job, on_log=lambda l: (append_log(self.log, l), self._page.update()), on_done=lambda r: self._reload()) dlg = ft.AlertDialog( modal=True, title=ft.Text("Установить пакет"), content=ft.Column([name_tf, results], tight=True, height=350, width=450), actions=[ ft.TextButton("Найти", on_click=search), ft.TextButton("Отмена", on_click=lambda e: self._close(dlg)), ft.FilledButton("Установить", on_click=do_install), ], ) self._page.show_dialog(dlg) def _close(self, dlg): self._page.pop_dialog() def _remove(self, p: Package): if p.is_system: snack(self._page, f"«{p.name}» — системный пакет, удаление заблокировано", error=True) return purge_cb = ft.Checkbox(label="Очистить конфигурацию (purge)", value=False) def do(): self._page.pop_dialog() self._show_log() async def job(log): async with SSHClient(self.srv) as c: stream = await remove(c, [p.name], purge_cb.value) async for line in stream: log(line) tasks.start("remove", f"remove {p.name}", self.srv.id, job, on_log=lambda l: (append_log(self.log, l), self._page.update()), on_done=lambda r: self._reload()) dlg = ft.AlertDialog( modal=True, title=ft.Text(f"Удалить {p.name}?"), content=purge_cb, actions=[ ft.TextButton("Отмена", on_click=lambda e: self._close(dlg)), ft.FilledButton("Удалить", on_click=lambda e: do(), style=ft.ButtonStyle(bgcolor=ft.Colors.RED_700)), ], ) self._page.show_dialog(dlg) def _autoremove(self, e): def do(): self._show_log() from app.core.packages import autoremove async def job(log): async with SSHClient(self.srv) as c: stream = await autoremove(c) async for line in stream: log(line) tasks.start("autoremove", "autoremove", self.srv.id, job, on_log=lambda l: (append_log(self.log, l), self._page.update())) confirm_dialog(self._page, "Autoremove?", "Будут удалены неиспользуемые зависимости.", do)