d596a747a4
Uses rich.spinner.Spinner with set_interval refresh instead of Textual's LoadingIndicator which caused rendering failures. Spinner shows dots animation while running, changes to checkmark on finish. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
95 lines
3.8 KiB
Python
95 lines
3.8 KiB
Python
from textual.app import ComposeResult
|
|
from textual.screen import Screen
|
|
from textual.widgets import Header, Footer, Static, Button, Select
|
|
from textual.containers import Vertical, Horizontal
|
|
from textual import work
|
|
|
|
from tui.config import list_conf_dir, has_targets, has_remotes
|
|
from tui.backend import stream_cli
|
|
from tui.widgets import ConfirmDialog, OperationLog
|
|
|
|
|
|
class BackupScreen(Screen):
|
|
|
|
BINDINGS = [("escape", "go_back", "Back")]
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Header()
|
|
targets = list_conf_dir("targets.d")
|
|
remotes = list_conf_dir("remotes.d")
|
|
with Vertical(id="backup-screen"):
|
|
yield Static("Backup", id="screen-title")
|
|
if not targets:
|
|
yield Static("No targets configured. Add a target first.")
|
|
else:
|
|
yield Static("Target:")
|
|
yield Select(
|
|
[(t, t) for t in targets],
|
|
id="backup-target",
|
|
prompt="Select target",
|
|
)
|
|
yield Static("Remote (optional):")
|
|
yield Select(
|
|
[("Default (all)", "")] + [(r, r) for r in remotes],
|
|
id="backup-remote",
|
|
prompt="Select remote",
|
|
value="",
|
|
)
|
|
with Horizontal(id="backup-buttons"):
|
|
yield Button("Run Backup", variant="primary", id="btn-backup")
|
|
yield Button("Backup All", variant="warning", id="btn-backup-all")
|
|
yield Button("Back", id="btn-back")
|
|
yield Footer()
|
|
|
|
def on_button_pressed(self, event: Button.Pressed) -> None:
|
|
if event.button.id == "btn-back":
|
|
self.app.pop_screen()
|
|
elif event.button.id == "btn-backup":
|
|
target_sel = self.query_one("#backup-target", Select)
|
|
if not isinstance(target_sel.value, str):
|
|
self.notify("Please select a target", severity="error")
|
|
return
|
|
target = str(target_sel.value)
|
|
remote_sel = self.query_one("#backup-remote", Select)
|
|
remote = str(remote_sel.value) if isinstance(remote_sel.value, str) else ""
|
|
msg = f"Run backup for target '{target}'?"
|
|
if remote:
|
|
msg += f"\nRemote: {remote}"
|
|
self.app.push_screen(
|
|
ConfirmDialog(msg, "Confirm Backup"),
|
|
callback=lambda ok: self._do_backup(target, remote) if ok else None,
|
|
)
|
|
elif event.button.id == "btn-backup-all":
|
|
self.app.push_screen(
|
|
ConfirmDialog("Backup ALL targets now?", "Confirm Backup"),
|
|
callback=lambda ok: self._do_backup_all() if ok else None,
|
|
)
|
|
|
|
@work
|
|
async def _do_backup(self, target: str, remote: str) -> None:
|
|
log_screen = OperationLog(f"Backup: {target}")
|
|
self.app.push_screen(log_screen)
|
|
args = ["backup", f"--target={target}"]
|
|
if remote:
|
|
args.append(f"--remote={remote}")
|
|
rc = await stream_cli(log_screen.write, *args)
|
|
if rc == 0:
|
|
log_screen.write("\n[green]Backup completed successfully.[/green]")
|
|
else:
|
|
log_screen.write(f"\n[red]Backup failed (exit code {rc}).[/red]")
|
|
log_screen.finish()
|
|
|
|
@work
|
|
async def _do_backup_all(self) -> None:
|
|
log_screen = OperationLog("Backup All Targets")
|
|
self.app.push_screen(log_screen)
|
|
rc = await stream_cli(log_screen.write, "backup", "--all")
|
|
if rc == 0:
|
|
log_screen.write("\n[green]All backups completed.[/green]")
|
|
else:
|
|
log_screen.write(f"\n[red]Backup failed (exit code {rc}).[/red]")
|
|
log_screen.finish()
|
|
|
|
def action_go_back(self) -> None:
|
|
self.app.pop_screen()
|