shukivaknin/gniza4linux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move schedule form to separate edit screen and fix crontab sync

Browse Source 
- Extract schedule add/edit form into ScheduleEditScreen (follows target_edit pattern)
- Fix toggle active: now properly installs/removes crontab entries with error reporting
- Delete also syncs crontab to remove deleted schedule entries
- Handle case where all schedules deactivated (calls remove instead of install)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
shuki
2026-03-06 04:45:27 +02:00
parent 1425c416eb
commit 48bd0ab1d4
5 changed files with 260 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tui/app.py
View File

@@ -11,6 +11,7 @@ from tui.screens.remote_edit import RemoteEditScreen
from tui.screens.snapshots import SnapshotsScreen
from tui.screens.retention import RetentionScreen
from tui.screens.schedule import ScheduleScreen
from tui.screens.schedule_edit import ScheduleEditScreen
from tui.screens.logs import LogsScreen
from tui.screens.settings import SettingsScreen
from tui.screens.wizard import WizardScreen
@@ -32,6 +33,7 @@ class GnizaApp(App):
"snapshots": SnapshotsScreen,
"retention": RetentionScreen,
"schedule": ScheduleScreen,
"schedule_edit": ScheduleEditScreen,
"logs": LogsScreen,
"settings": SettingsScreen,
"wizard": WizardScreen,

3
tui/gniza.tcss
View File

@@ -42,6 +42,7 @@ DataTable {
/* Form screens */
#target-edit,
#remote-edit,
#schedule-edit,
#settings-screen,
#schedule-screen,
#backup-screen,
@@ -93,6 +94,7 @@ SelectionList {
#remotes-buttons,
#logs-buttons,
#sched-buttons,
#sched-edit-buttons,
#te-buttons,
#re-buttons,
#set-buttons {
@@ -107,6 +109,7 @@ SelectionList {
#remotes-buttons Button,
#logs-buttons Button,
#sched-buttons Button,
#sched-edit-buttons Button,
#te-buttons Button,
#re-buttons Button,
#set-buttons Button {

1
tui/screens/__init__.py
View File

@@ -8,6 +8,7 @@ from tui.screens.remote_edit import RemoteEditScreen
from tui.screens.snapshots import SnapshotsScreen
from tui.screens.retention import RetentionScreen
from tui.screens.schedule import ScheduleScreen
from tui.screens.schedule_edit import ScheduleEditScreen
from tui.screens.logs import LogsScreen
from tui.screens.settings import SettingsScreen
from tui.screens.wizard import WizardScreen

216
tui/screens/schedule.py
View File

@@ -1,35 +1,14 @@
import re
from textual.app import ComposeResult
from textual.screen import Screen
from textual.widgets import Header, Footer, Static, Button, DataTable, Input, Select, SelectionList
from textual.widgets import Header, Footer, Static, Button, DataTable
from textual.containers import Vertical, Horizontal
from textual import work
from tui.config import list_conf_dir, parse_conf, write_conf, update_conf_key, CONFIG_DIR
from tui.config import list_conf_dir, parse_conf, update_conf_key, CONFIG_DIR
from tui.models import Schedule
from tui.backend import run_cli
from tui.widgets import ConfirmDialog, OperationLog
_NAME_RE = re.compile(r'^[a-zA-Z][a-zA-Z0-9_-]{0,31}$')
SCHEDULE_TYPES = [
("Hourly", "hourly"),
("Daily", "daily"),
("Weekly", "weekly"),
("Monthly", "monthly"),
("Custom cron", "custom"),
]
HOURLY_INTERVALS = [
("Every Hour", "1"),
("Every 2 Hours", "2"),
("Every 3 Hours", "3"),
("Every 4 Hours", "4"),
("Every 6 Hours", "6"),
("Every 8 Hours", "8"),
("Every 12 Hours", "12"),
]
class ScheduleScreen(Screen):
@@ -47,88 +26,10 @@ class ScheduleScreen(Screen):
yield Button("Delete", variant="error", id="btn-delete")
yield Button("Show crontab", id="btn-show")
yield Button("Back", id="btn-back")
yield Static("", id="sched-divider")
yield Static("Add Schedule", id="sched-form-title")
yield Static("Name:")
yield Input(id="sched-name", placeholder="Schedule name")
yield Static("Type:")
yield Select(SCHEDULE_TYPES, id="sched-type", value="daily")
yield Static("Schedule Hours:", classes="sched-hourly-field")
yield Select(HOURLY_INTERVALS, id="sched-interval", value="1", classes="sched-hourly-field")
yield Static("Time (HH:MM):", classes="sched-time-field")
yield Input(id="sched-time", value="02:00", placeholder="02:00", classes="sched-time-field")
yield Static("Schedule Days:", classes="sched-daily-days-field")
yield SelectionList[str](
("Sunday", "0"),
("Monday", "1"),
("Tuesday", "2"),
("Wednesday", "3"),
("Thursday", "4"),
("Friday", "5"),
("Saturday", "6"),
id="sched-daily-days",
classes="sched-daily-days-field",
)
yield Static("Schedule Day:", classes="sched-weekly-day-field")
yield Select(
[("Sunday", "0"), ("Monday", "1"), ("Tuesday", "2"), ("Wednesday", "3"),
("Thursday", "4"), ("Friday", "5"), ("Saturday", "6")],
id="sched-weekly-day",
value="0",
classes="sched-weekly-day-field",
)
yield Static("Schedule Day:", classes="sched-monthly-field")
yield Select(
[("1st of the month", "1"), ("7th of the month", "7"),
("14th of the month", "14"), ("21st of the month", "21"),
("28th of the month", "28")],
id="sched-monthly-day",
value="1",
classes="sched-monthly-field",
)
yield Static("Custom cron (5 fields):", classes="sched-cron-field")
yield Input(id="sched-cron", placeholder="0 2 * * *", classes="sched-cron-field")
yield Static("Targets (empty=all):")
yield SelectionList[str](
*self._build_target_choices(),
id="sched-targets",
)
yield Static("Remotes (empty=all):")
yield SelectionList[str](
*self._build_remote_choices(),
id="sched-remotes",
)
yield Footer()
def _build_target_choices(self) -> list[tuple[str, str]]:
return [(name, name) for name in list_conf_dir("targets.d")]
def _build_remote_choices(self) -> list[tuple[str, str]]:
return [(name, name) for name in list_conf_dir("remotes.d")]
def on_mount(self) -> None:
self._refresh_table()
self._update_type_visibility()
def on_select_changed(self, event: Select.Changed) -> None:
if event.select.id == "sched-type":
self._update_type_visibility()
def _update_type_visibility(self) -> None:
type_sel = self.query_one("#sched-type", Select)
stype = str(type_sel.value) if isinstance(type_sel.value, str) else "daily"
for w in self.query(".sched-hourly-field"):
w.display = stype == "hourly"
for w in self.query(".sched-time-field"):
w.display = stype in ("daily", "weekly", "monthly")
for w in self.query(".sched-daily-days-field"):
w.display = stype == "daily"
for w in self.query(".sched-weekly-day-field"):
w.display = stype == "weekly"
for w in self.query(".sched-monthly-field"):
w.display = stype == "monthly"
for w in self.query(".sched-cron-field"):
w.display = stype == "custom"
def _refresh_table(self) -> None:
table = self.query_one("#sched-table", DataTable)
@@ -151,11 +52,13 @@ class ScheduleScreen(Screen):
if event.button.id == "btn-back":
self.app.pop_screen()
elif event.button.id == "btn-add":
self._save_schedule()
from tui.screens.schedule_edit import ScheduleEditScreen
self.app.push_screen(ScheduleEditScreen(), callback=lambda _: self._refresh_table())
elif event.button.id == "btn-edit":
name = self._selected_schedule()
if name:
self._load_schedule(name)
from tui.screens.schedule_edit import ScheduleEditScreen
self.app.push_screen(ScheduleEditScreen(name), callback=lambda _: self._refresh_table())
else:
self.notify("Select a schedule first", severity="warning")
elif event.button.id == "btn-delete":
@@ -176,98 +79,6 @@ class ScheduleScreen(Screen):
elif event.button.id == "btn-show":
self._show_crontab()
def _load_schedule(self, name: str) -> None:
"""Load a schedule's config into the form for editing."""
data = parse_conf(CONFIG_DIR / "schedules.d" / f"{name}.conf")
s = Schedule.from_conf(name, data)
self.query_one("#sched-name", Input).value = name
self.query_one("#sched-type", Select).value = s.schedule
self.query_one("#sched-time", Input).value = s.time
self.query_one("#sched-cron", Input).value = s.cron
# Hourly interval
if s.schedule == "hourly" and s.day:
self.query_one("#sched-interval", Select).value = s.day
# Daily days
if s.schedule == "daily" and s.day:
days_list = self.query_one("#sched-daily-days", SelectionList)
day_vals = set(s.day.split(","))
for idx in range(days_list.option_count):
opt = days_list.get_option_at_index(idx)
if opt.value in day_vals:
days_list.select(opt.value)
else:
days_list.deselect(opt.value)
# Weekly day
if s.schedule == "weekly" and s.day:
self.query_one("#sched-weekly-day", Select).value = s.day
# Monthly day
if s.schedule == "monthly" and s.day:
self.query_one("#sched-monthly-day", Select).value = s.day
# Targets
if s.targets:
target_vals = set(s.targets.split(","))
tlist = self.query_one("#sched-targets", SelectionList)
for idx in range(tlist.option_count):
opt = tlist.get_option_at_index(idx)
if opt.value in target_vals:
tlist.select(opt.value)
else:
tlist.deselect(opt.value)
# Remotes
if s.remotes:
remote_vals = set(s.remotes.split(","))
rlist = self.query_one("#sched-remotes", SelectionList)
for idx in range(rlist.option_count):
opt = rlist.get_option_at_index(idx)
if opt.value in remote_vals:
rlist.select(opt.value)
else:
rlist.deselect(opt.value)
self._update_type_visibility()
self.query_one("#sched-form-title", Static).update("Edit Schedule")
self.notify(f"Editing schedule '{name}'")
def _save_schedule(self) -> None:
name = self.query_one("#sched-name", Input).value.strip()
if not name:
self.notify("Name is required", severity="error")
return
if not _NAME_RE.match(name):
self.notify("Invalid name.", severity="error")
return
conf = CONFIG_DIR / "schedules.d" / f"{name}.conf"
type_sel = self.query_one("#sched-type", Select)
stype = str(type_sel.value) if isinstance(type_sel.value, str) else "daily"
if stype == "hourly":
interval_sel = self.query_one("#sched-interval", Select)
day_val = str(interval_sel.value) if isinstance(interval_sel.value, str) else "1"
elif stype == "daily":
selected_days = sorted(self.query_one("#sched-daily-days", SelectionList).selected)
day_val = ",".join(selected_days)
elif stype == "weekly":
wday_sel = self.query_one("#sched-weekly-day", Select)
day_val = str(wday_sel.value) if isinstance(wday_sel.value, str) else "0"
elif stype == "monthly":
mday_sel = self.query_one("#sched-monthly-day", Select)
day_val = str(mday_sel.value) if isinstance(mday_sel.value, str) else "1"
else:
day_val = ""
sched = Schedule(
name=name,
schedule=stype,
time=self.query_one("#sched-time", Input).value.strip() or "02:00",
day=day_val,
cron=self.query_one("#sched-cron", Input).value.strip(),
targets=",".join(self.query_one("#sched-targets", SelectionList).selected),
remotes=",".join(self.query_one("#sched-remotes", SelectionList).selected),
)
is_new = not conf.exists()
write_conf(conf, sched.to_conf())
self.notify(f"Schedule '{name}' {'created' if is_new else 'updated'}.")
self._refresh_table()
self.query_one("#sched-name", Input).value = ""
self.query_one("#sched-form-title", Static).update("Add Schedule")
def _toggle_active(self, name: str) -> None:
conf = CONFIG_DIR / "schedules.d" / f"{name}.conf"
data = parse_conf(conf)
@@ -282,7 +93,19 @@ class ScheduleScreen(Screen):
@work
async def _sync_crontab(self) -> None:
"""Reinstall crontab with only active schedules."""
await run_cli("schedule", "install")
# Check if any schedule is active
has_active = False
for name in list_conf_dir("schedules.d"):
data = parse_conf(CONFIG_DIR / "schedules.d" / f"{name}.conf")
if data.get("SCHEDULE_ACTIVE", "yes") == "yes":
has_active = True
break
if has_active:
rc, stdout, stderr = await run_cli("schedule", "install")
else:
rc, stdout, stderr = await run_cli("schedule", "remove")
if rc != 0:
self.notify(f"Crontab sync failed: {stderr or stdout}", severity="error")
def _delete_schedule(self, name: str) -> None:
conf = CONFIG_DIR / "schedules.d" / f"{name}.conf"
@@ -290,6 +113,7 @@ class ScheduleScreen(Screen):
conf.unlink()
self.notify(f"Schedule '{name}' deleted.")
self._refresh_table()
self._sync_crontab()
@work
async def _install_schedules(self) -> None:

234
tui/screens/schedule_edit.py Normal file
View File

@@ -0,0 +1,234 @@
import re
from textual.app import ComposeResult
from textual.screen import Screen
from textual.widgets import Header, Footer, Static, Button, Input, Select, SelectionList
from textual.containers import Vertical, Horizontal
from tui.config import list_conf_dir, parse_conf, write_conf, CONFIG_DIR
from tui.models import Schedule
_NAME_RE = re.compile(r'^[a-zA-Z][a-zA-Z0-9_-]{0,31}$')
SCHEDULE_TYPES = [
("Hourly", "hourly"),
("Daily", "daily"),
("Weekly", "weekly"),
("Monthly", "monthly"),
("Custom cron", "custom"),
]
HOURLY_INTERVALS = [
("Every Hour", "1"),
("Every 2 Hours", "2"),
("Every 3 Hours", "3"),
("Every 4 Hours", "4"),
("Every 6 Hours", "6"),
("Every 8 Hours", "8"),
("Every 12 Hours", "12"),
]
class ScheduleEditScreen(Screen):
BINDINGS = [("escape", "cancel", "Cancel")]
def __init__(self, name: str = ""):
super().__init__()
self._edit_name = name
self._is_new = not name
def compose(self) -> ComposeResult:
yield Header()
title = "Add Schedule" if self._is_new else f"Edit Schedule: {self._edit_name}"
sched = Schedule()
if not self._is_new:
data = parse_conf(CONFIG_DIR / "schedules.d" / f"{self._edit_name}.conf")
sched = Schedule.from_conf(self._edit_name, data)
with Vertical(id="schedule-edit"):
yield Static(title, id="screen-title")
if self._is_new:
yield Static("Name:")
yield Input(value="", placeholder="Schedule name", id="sched-name")
yield Static("Type:")
yield Select(SCHEDULE_TYPES, id="sched-type", value=sched.schedule)
yield Static("Schedule Hours:", classes="sched-hourly-field")
yield Select(
HOURLY_INTERVALS,
id="sched-interval",
value=sched.day if sched.schedule == "hourly" and sched.day else "1",
classes="sched-hourly-field",
)
yield Static("Time (HH:MM):", classes="sched-time-field")
yield Input(
id="sched-time",
value=sched.time,
placeholder="02:00",
classes="sched-time-field",
)
yield Static("Schedule Days:", classes="sched-daily-days-field")
yield SelectionList[str](
("Sunday", "0"),
("Monday", "1"),
("Tuesday", "2"),
("Wednesday", "3"),
("Thursday", "4"),
("Friday", "5"),
("Saturday", "6"),
id="sched-daily-days",
classes="sched-daily-days-field",
)
yield Static("Schedule Day:", classes="sched-weekly-day-field")
yield Select(
[("Sunday", "0"), ("Monday", "1"), ("Tuesday", "2"), ("Wednesday", "3"),
("Thursday", "4"), ("Friday", "5"), ("Saturday", "6")],
id="sched-weekly-day",
value=sched.day if sched.schedule == "weekly" and sched.day else "0",
classes="sched-weekly-day-field",
)
yield Static("Schedule Day:", classes="sched-monthly-field")
yield Select(
[("1st of the month", "1"), ("7th of the month", "7"),
("14th of the month", "14"), ("21st of the month", "21"),
("28th of the month", "28")],
id="sched-monthly-day",
value=sched.day if sched.schedule == "monthly" and sched.day else "1",
classes="sched-monthly-field",
)
yield Static("Custom cron (5 fields):", classes="sched-cron-field")
yield Input(
id="sched-cron",
value=sched.cron,
placeholder="0 2 * * *",
classes="sched-cron-field",
)
yield Static("Targets (empty=all):")
yield SelectionList[str](
*self._build_target_choices(),
id="sched-targets",
)
yield Static("Remotes (empty=all):")
yield SelectionList[str](
*self._build_remote_choices(),
id="sched-remotes",
)
with Horizontal(id="sched-edit-buttons"):
yield Button("Save", variant="primary", id="btn-save")
yield Button("Cancel", id="btn-cancel")
yield Footer()
def _build_target_choices(self) -> list[tuple[str, str]]:
return [(name, name) for name in list_conf_dir("targets.d")]
def _build_remote_choices(self) -> list[tuple[str, str]]:
return [(name, name) for name in list_conf_dir("remotes.d")]
def on_mount(self) -> None:
self._update_type_visibility()
if not self._is_new:
self._load_selections()
def _load_selections(self) -> None:
"""Pre-select items in SelectionLists when editing."""
data = parse_conf(CONFIG_DIR / "schedules.d" / f"{self._edit_name}.conf")
sched = Schedule.from_conf(self._edit_name, data)
# Daily days
if sched.schedule == "daily" and sched.day:
days_list = self.query_one("#sched-daily-days", SelectionList)
day_vals = set(sched.day.split(","))
for idx in range(days_list.option_count):
opt = days_list.get_option_at_index(idx)
if opt.value in day_vals:
days_list.select(opt.value)
# Targets
if sched.targets:
target_vals = set(sched.targets.split(","))
tlist = self.query_one("#sched-targets", SelectionList)
for idx in range(tlist.option_count):
opt = tlist.get_option_at_index(idx)
if opt.value in target_vals:
tlist.select(opt.value)
# Remotes
if sched.remotes:
remote_vals = set(sched.remotes.split(","))
rlist = self.query_one("#sched-remotes", SelectionList)
for idx in range(rlist.option_count):
opt = rlist.get_option_at_index(idx)
if opt.value in remote_vals:
rlist.select(opt.value)
def on_select_changed(self, event: Select.Changed) -> None:
if event.select.id == "sched-type":
self._update_type_visibility()
def _update_type_visibility(self) -> None:
type_sel = self.query_one("#sched-type", Select)
stype = str(type_sel.value) if isinstance(type_sel.value, str) else "daily"
for w in self.query(".sched-hourly-field"):
w.display = stype == "hourly"
for w in self.query(".sched-time-field"):
w.display = stype in ("daily", "weekly", "monthly")
for w in self.query(".sched-daily-days-field"):
w.display = stype == "daily"
for w in self.query(".sched-weekly-day-field"):
w.display = stype == "weekly"
for w in self.query(".sched-monthly-field"):
w.display = stype == "monthly"
for w in self.query(".sched-cron-field"):
w.display = stype == "custom"
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "btn-cancel":
self.dismiss(None)
elif event.button.id == "btn-save":
self._save()
def _save(self) -> None:
if self._is_new:
name = self.query_one("#sched-name", Input).value.strip()
if not name:
self.notify("Name is required", severity="error")
return
if not _NAME_RE.match(name):
self.notify("Invalid name. Use letters, digits, _ - (max 32 chars, start with letter).", severity="error")
return
conf = CONFIG_DIR / "schedules.d" / f"{name}.conf"
if conf.exists():
self.notify(f"Schedule '{name}' already exists.", severity="error")
return
else:
name = self._edit_name
type_sel = self.query_one("#sched-type", Select)
stype = str(type_sel.value) if isinstance(type_sel.value, str) else "daily"
if stype == "hourly":
interval_sel = self.query_one("#sched-interval", Select)
day_val = str(interval_sel.value) if isinstance(interval_sel.value, str) else "1"
elif stype == "daily":
selected_days = sorted(self.query_one("#sched-daily-days", SelectionList).selected)
day_val = ",".join(selected_days)
elif stype == "weekly":
wday_sel = self.query_one("#sched-weekly-day", Select)
day_val = str(wday_sel.value) if isinstance(wday_sel.value, str) else "0"
elif stype == "monthly":
mday_sel = self.query_one("#sched-monthly-day", Select)
day_val = str(mday_sel.value) if isinstance(mday_sel.value, str) else "1"
else:
day_val = ""
sched = Schedule(
name=name,
schedule=stype,
time=self.query_one("#sched-time", Input).value.strip() or "02:00",
day=day_val,
cron=self.query_one("#sched-cron", Input).value.strip(),
targets=",".join(self.query_one("#sched-targets", SelectionList).selected),
remotes=",".join(self.query_one("#sched-remotes", SelectionList).selected),
)
conf = CONFIG_DIR / "schedules.d" / f"{name}.conf"
write_conf(conf, sched.to_conf())
self.notify(f"Schedule '{name}' saved.")
self.dismiss(name)
def action_cancel(self) -> None:
self.dismiss(None)